GodopuandClaude Opus 5
135d76f903
Review round 2 (planner-reviewer-codex-01, NOT PASS at d09c162):
- R1 remote.mirror_job: admission under the registry lock. The same
dispatch_token is idempotent (a finished attempt never runs again), a
lower attempt or the same attempt under another token is refused before
any write, and only a strictly newer attempt replaces the record
(register_job(supersede_older_attempt=True); events.log kept, seqs stay
monotonic).
- R2 remote.handle_dispatch: the --exec command runs in its own session;
at timeout_sec, or when the runner is interrupted (SIGTERM now unwinds
it), the whole process group is terminated (TERM, then KILL) before the
error is published.
- R3 terminal barrier: a terminal publish raises an attempt-scoped barrier
under the lock that allocates seqs, stops the heartbeat timer, gives it
MAM_HEARTBEAT_DRAIN_S (default 1 s) to exit and then kills it, and only
then takes its own seq. Beats are admitted under that lock
(next_seq(admit=...)); a beat still on the wire is cancelled with the
timer rather than holding the terminal back; and a late non-terminal
event never publishes a running state over a terminal record.
- R4 job_subscriber: a retained state renews the idle clock only when its
seq is newer than everything already accepted; re-delivered or older
snapshots still recover a terminal status.
- tests: five regressions, all failing on 77f2829; T1-14's wall clock
15 -> 30 s against public-broker latency.
- docs: MESSAGING 2.5 (barrier, state renewal, detection vs final close),
heartbeat docstring, loop SKILL.md (remote rows need the Python runner),
VERSIONS v4.5.0 items 1/4/5 and the test count (25).
Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01HCGpJXhSZasJLU38ksDghe
518 lines
23 KiB
Python
Executable File
518 lines
23 KiB
Python
Executable File
#!/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
|
|
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 (`_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 <topic_prefix>/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
|
|
|
|
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
|
|
|
|
# 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())
|