#!/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 ``/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 --event started [--detail "..."] [--data '{...}'] publish_event.py --pick-pending --agent-session herdr:claude --event completed publish_event.py --job --event completed --retained """ from __future__ import annotations import argparse import hashlib import hmac import json import logging import sys from typing import Any, Dict, Optional from lib_py.delegation import heartbeat as heartbeat_timer 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, default_registry_dir, events_topic_for, load_job, 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", } # Re-exported from the transport module, which owns the single implementation. # Two copies of these numbers meant tuning one silently diverged from the other. from lib_py.delegation.publisher import ( # noqa: F401 (re-exported for callers) CONNECT_ACK_TIMEOUT, PUBLISH_ACK_TIMEOUT, OneShotPublisher, ) 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, extra=()) -> None: """Connect, publish one QoS-1 message, wait for the broker ACK, disconnect. Delegates to the transport module so the CLI path and the service path share one implementation. Keeping a second copy here is what let the persistent transport's handshake and not-connected bugs exist in only one of them. Raises on any failure so ``with_retry`` can re-run the whole sequence. ``extra`` (Issue #18): further ``(topic, body, retain)`` messages sent over the same connection after the event - the retained job-state message, so it costs no second broker handshake. """ OneShotPublisher().publish_raw(config, topic, body, retain, extra=extra) def state_topic_for(job: Dict[str, Any]) -> str: prefix = job.get("topic_prefix") or mqtt_common.topic_prefix_for(job["job_id"]) return f"{prefix}/state" def build_state_payload(job: Dict[str, Any], prepared: "PreparedEvent") -> Dict[str, Any]: """The retained job-state message published next to every event (Issue #18). A subscriber that (re)connects receives it at once and learns the job's status and the last seq without waiting for the next heartbeat; a seq it never saw in between is a detected loss, not a silent one. Signed like an event so a forged state cannot close a job. """ status = EVENT_TO_STATUS.get(prepared.event) or "running" if prepared.event not in TERMINAL_EVENTS and job.get("status") in TERMINAL_EVENTS: # A late non-terminal event must not publish a retained `running` over # the terminal snapshot (review round 2, codex R3). status = job["status"] data = {k: v for k, v in (prepared.payload.get("data") or {}).items() if k in ("dispatch_token", "attempt_id")} payload = build_payload(job_id=prepared.job_id, seq=prepared.seq, event="state", detail=prepared.event, data=data, auth_token=job.get("auth_token")) payload["status"] = status payload["last_event"] = prepared.event if job.get("auth_token"): # re-sign: `status`/`last_event` were added after build_payload signed payload["data"].pop("hmac_sig", None) sign_payload = {k: v for k, v in payload.items() if k != "data"} sign_payload["data"] = dict(payload["data"]) msg = json.dumps(sign_payload, sort_keys=True, separators=(",", ":")).encode() payload["data"]["hmac_sig"] = hmac.new(job["auth_token"].encode(), msg, hashlib.sha256).hexdigest() return payload class PreparedEvent(object): """Everything decided before the wire: payload, topic, broker, retain flag. Splitting this out lets a different transport (one-shot vs persistent) be swapped in without touching the parts that must not vary: the seq is already consumed, the HMAC is already computed over the final payload, and the audit record is written from these exact fields. """ __slots__ = ("job_id", "event", "detail", "seq", "topic", "retain", "payload", "body", "config") def __init__(self, job_id, event, detail, seq, topic, retain, payload, body, config): self.job_id = job_id self.event = event self.detail = detail self.seq = seq self.topic = topic self.retain = retain self.payload = payload self.body = body self.config = config def prepare_event(job_id, event, detail="", data=None, registry_dir=DEFAULT_REGISTRY_DIR, retained=False, dispatch_token=None, attempt_id=None, admit=None) -> PreparedEvent: """Load the job, consume the next seq, and build the signed payload. Consuming the seq here (before any transport runs) is deliberate: a failed publish must still burn the number, so a replayed packet can never carry a seq the subscriber would accept. ``admit`` is checked under the lock that allocates the seq (``mqtt_common.next_seq``); a refusal raises ``AdmissionRefused`` and consumes nothing. """ job = load_job(job_id, registry_dir) 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, registry_dir) if admit is None else next_seq(job_id, registry_dir, admit=admit) payload_data = dict(data) if data else {} eff_dispatch_token = dispatch_token if eff_dispatch_token is not None and "dispatch_token" not in payload_data: payload_data["dispatch_token"] = eff_dispatch_token eff_attempt_id = attempt_id if eff_attempt_id is not None and "attempt_id" not in payload_data: payload_data["attempt_id"] = eff_attempt_id payload = build_payload( job_id=job_id, seq=seq, event=event, detail=detail, data=payload_data, auth_token=job.get("auth_token"), ) body = json.dumps(payload, ensure_ascii=False).encode("utf-8") retain = retained or event in TERMINAL_EVENTS return PreparedEvent(job_id, event, detail, seq, topic, retain, payload, body, config) def record_local_event(prepared: PreparedEvent, registry_dir: str) -> None: """Append the event to the job's on-disk events.log BEFORE any wire publish. The delegate wrapper reads this log as liveness evidence once its subscriber has idled out (`wrapper.recent_disk_heartbeat`). Recorded only after the publish - which blocks on the broker ACK and its retries - a slow broker's heartbeat reached the disk too late for that grace window, and live jobs were closed as deadline_timeout. Best effort, like the rest of the record. """ registry.append_event(prepared.job_id, registry_dir, prepared.payload) def record_outcome(prepared: PreparedEvent, registry_dir: str, publish_ok: bool, publish_error: Optional[str], expected_dispatch_token: Optional[str] = None, expected_attempt: Optional[int] = None, local_event_recorded: bool = False) -> None: """Persist the audit record, the registry event log and the status sync. All three are best effort and all three happen whether or not the publish succeeded — that is the contract G-1/G-2/G-4 pin down. A caller that already wrote the registry event log before publishing (``record_local_event``) passes ``local_event_recorded=True`` so the line is not appended twice. """ mqtt_common.append_event(prepared.job_id, { "event": "published", "source_event": prepared.event, "seq": prepared.seq, "topic": prepared.topic, "retain": prepared.retain, "timestamp": prepared.payload["timestamp"], "detail": prepared.detail, "payload": prepared.payload, "published": publish_ok, "publish_error": publish_error, }) if not local_event_recorded: registry.append_event(prepared.job_id, registry_dir, prepared.payload) new_status = EVENT_TO_STATUS.get(prepared.event) if new_status: try: exp_token = expected_dispatch_token if exp_token is None and isinstance(prepared.payload.get("data"), dict): exp_token = prepared.payload["data"].get("dispatch_token") exp_att = expected_attempt if exp_att is None and isinstance(prepared.payload.get("data"), dict): exp_att = prepared.payload["data"].get("attempt_id") # Agent authority on purpose: an agent may never move a terminal # record, which is what the operator-only requests are gated for. mqtt_common.update_job_status(prepared.job_id, registry_dir, _source="publish_event", status=new_status, expected_dispatch_token=exp_token, expected_attempt=exp_att) except Exception as exc: # pragma: no cover - best effort logger.warning("status sync failed: %s", exc) 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 _beat_admission(job_id, registry_dir, dispatch_token, attempt_id): """Admission for one heartbeat beat, evaluated under the lock that allocates its seq (review round 2, codex R3): never after this attempt's terminal barrier, on a terminal record, or after a re-dispatch.""" def admit(record): status = record.get("status") if status in TERMINAL_EVENTS: return "job is {}".format(status) if heartbeat_timer.barrier_blocks(job_id, registry_dir, dispatch_token, attempt_id): return "the terminal barrier for this attempt is raised" if dispatch_token and record.get("dispatch_token") != dispatch_token: return "dispatch token changed (re-dispatched)" if attempt_id is not None and record.get("attempt_id", record.get("iteration", 1)) != attempt_id: return "attempt changed (re-dispatched)" return None return admit def main(argv=None, _beat: bool = False) -> int: """CLI entry point. ``_beat`` is set only by the heartbeat timer (``heartbeat.run``): the beat is admitted under the seq lock (``_beat_admission``) and a refusal returns 4, which stops the timer.""" 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("--finalize-report", default=None, metavar="PATH", help="atomically finalize this report and declare its " "completion manifest before publishing") parser.add_argument("--side-effect", action="append", default=[], dest="side_effects", metavar="KIND=REFERENCE", help="evidence for an external side effect this job performed") parser.add_argument("--attempt-id", type=int, default=None, help="attempt this finalization belongs to (default: the " "job's current iteration)") parser.add_argument("--expect-dispatch-token", default=None, help="expected dispatch token for CAS verification") parser.add_argument("--heartbeat", type=int, default=None, metavar="N", help="after publishing, start a detached timer that publishes a " "`progress` heartbeat every N seconds until the job ends " "(Issue #18; not with a terminal event)") parser.add_argument("--no-state", action="store_true", help="do not publish the retained /state message") parser.add_argument("--watch-pid", type=int, default=None, help="bind the heartbeat timer to this process (default: the nearest " "agent-CLI ancestor; 0 = no watch)") parser.add_argument("-v", "--verbose", action="store_true") args = parser.parse_args(argv) if args.heartbeat is not None and (args.event in TERMINAL_EVENTS or args.heartbeat < 1): logger.error("--heartbeat needs a positive period and a non-terminal event") return 1 if not mqtt_common.load_env_or_report(): return 1 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 # Finalize before publishing: a `completed` event whose report is still being # written would invite exactly the partial-report adoption the contract # forbids. Failing here fails the publish, on purpose. if args.finalize_report: from lib_py.delegation import finalize # same package: no sibling hop (see __init__) try: side_effects = finalize.parse_side_effects(args.side_effects) except ValueError as exc: logger.error("%s", exc) return 1 attempt_id = args.attempt_id if attempt_id is None: try: rec = load_job(job_id, args.registry_dir) attempt_id = rec.get("attempt_id", rec.get("iteration", 1)) except FileNotFoundError: attempt_id = 1 try: manifest = finalize.finalize_report( args.finalize_report, job_id, attempt_id, side_effects=side_effects, declared_by="agent") except (OSError, FileNotFoundError) as exc: logger.error("could not finalize report: %s", exc) return 1 logger.info("finalized %s (sha256 %s...)", manifest["report_path"], manifest["sha256"][:12]) check_token = args.expect_dispatch_token if check_token is None and isinstance(data, dict) and "dispatch_token" in data: check_token = data.get("dispatch_token") try: rec = load_job(job_id, args.registry_dir) actual_tok = rec.get("dispatch_token") rec_attempt = rec.get("attempt_id", rec.get("iteration", 1)) except FileNotFoundError: rec = None actual_tok = None rec_attempt = 1 check_attempt = args.attempt_id if check_attempt is None and isinstance(data, dict) and "attempt_id" in data: check_attempt = data.get("attempt_id") if rec is not None and actual_tok is not None: if check_token is None and check_attempt is None: if args.event in ("completed", "error", "cancelled") and rec_attempt > 1: logger.error("dispatch token required: redispatched job %s carries a dispatch token " "but none was provided (aborting terminal publish)", job_id) return 4 elif check_token is not None and actual_tok != check_token: logger.error("dispatch token mismatch: expected %s, got %s (aborting publish)", check_token, actual_tok) return 4 elif check_attempt is not None and rec_attempt != check_attempt: logger.error("attempt mismatch: expected %s, got %s (aborting publish)", check_attempt, rec_attempt) return 4 elif check_token is not None and actual_tok != check_token: logger.error("dispatch token mismatch: expected %s, got %s (aborting publish)", check_token, actual_tok) return 4 # Issue #18: a terminal event ends the heartbeat obligation. Review round 2 # (codex R3): first raise this attempt's barrier - under the lock that # allocates seqs, so no beat can take a seq after it - then stop the timer # and wait for it to drain its in-flight beats (killing it if it does not), # and only then take this event's seq. No heartbeat can follow the terminal # event, on disk or on the wire. if args.event in TERMINAL_EVENTS: if rec is not None: heartbeat_timer.raise_barrier( job_id, args.registry_dir, dispatch_token=args.expect_dispatch_token or actual_tok, attempt_id=args.attempt_id if args.attempt_id is not None else rec_attempt) if heartbeat_timer.stop_timer(job_id, args.registry_dir, drain=heartbeat_timer.drain_seconds()): logger.info("stopped the heartbeat timer for job %s", job_id) try: prepared = prepare_event( job_id=job_id, event=args.event, detail=args.detail, data=data, registry_dir=args.registry_dir, retained=args.retained, dispatch_token=args.expect_dispatch_token, attempt_id=args.attempt_id, admit=_beat_admission(job_id, args.registry_dir, check_token, check_attempt) if _beat else None, ) except mqtt_common.AdmissionRefused as exc: logger.info("heartbeat for job %s not published: %s", job_id, exc) return 4 except FileNotFoundError as exc: logger.error("%s", exc) return 1 extra_messages = [] if not args.no_state and rec is not None: state_payload = build_state_payload(rec, prepared) extra_messages.append((state_topic_for(rec), json.dumps(state_payload, ensure_ascii=False).encode("utf-8"), True)) # Local first: the event reaches events.log before the publish can block on # the broker ACK and its retries (see record_local_event). record_local_event(prepared, args.registry_dir) # Policy Note: seq consumption on failure is intentional (see prepare_event) # and the failure is documented explicitly in the audit record. publish = with_retry( _publish_once, attempts=args.attempts, exceptions=(OSError, TimeoutError, ConnectionError, ValueError), ) publish_ok = True publish_error: Optional[str] = None try: publish(prepared.config, prepared.topic, prepared.body, prepared.retain, extra_messages) except Exception as exc: publish_ok = False publish_error = str(exc) logger.error("publish failed after %d attempts: %s", args.attempts, exc) sync_expected_token = args.expect_dispatch_token if sync_expected_token is None and isinstance(data, dict): sync_expected_token = data.get("dispatch_token") sync_expected_attempt = args.attempt_id if sync_expected_attempt is None and isinstance(data, dict): sync_expected_attempt = data.get("attempt_id") record_outcome(prepared, args.registry_dir, publish_ok, publish_error, expected_dispatch_token=sync_expected_token, expected_attempt=sync_expected_attempt, local_event_recorded=True) # Issue #18: the automatic heartbeat. Started whether or not the wire # publish succeeded - the agent is working either way, and each beat lands # in events.log before its own publish, which is what local liveness reads. if args.heartbeat is not None: hb_token = sync_expected_token hb_attempt = sync_expected_attempt if rec is not None: hb_token = hb_token or rec.get("dispatch_token") hb_attempt = hb_attempt if hb_attempt is not None else rec.get("attempt_id", rec.get("iteration", 1)) try: pid = heartbeat_timer.start_timer(job_id, args.registry_dir, args.heartbeat, dispatch_token=hb_token, attempt_id=hb_attempt, watch_pid=args.watch_pid) logger.info("heartbeat timer started (pid %d, every %ss)", pid, args.heartbeat) except OSError as exc: logger.error("could not start the heartbeat timer: %s", exc) if not publish_ok: return 2 logger.info("published %s seq=%d job=%s retain=%s", args.event, prepared.seq, job_id, prepared.retain) return 0 if __name__ == "__main__": sys.exit(main())