feat(issue-18): decide liveness and delivery by the event protocol, not the screen

Contract (no new event types): heartbeat every N s (automatic timer),
unresponsive after K x N s without an event, `started` within T s of the
injection, retained per-job state, and a remote transport.

- registry: heartbeat_sec / heartbeat_misses / ack_timeout_sec / transport
  fields; idle_timeout_sec derives K x N unless explicit (defaults 30/4/60
  keep the historical 120 s idle window).
- heartbeat.py + publish_event --heartbeat N: detached fixed-cadence timer
  publishing signed `progress` heartbeats (threaded wire publish, local
  record first); stopped by the terminal event via pidfile, by re-dispatch
  and by lifetime. Terminal events refuse --heartbeat.
- publish_event: retained `<prefix>/state` (status, last_event, seq, HMAC)
  sent on the same connection after every event (--no-state to skip);
  publisher.publish_raw takes extra messages.
- job_subscriber: subscribes the state topic, recovers terminal status from
  retained state, resets idle on any accepted message, records seq_gap and
  reconnected in the audit log, idle budget before the first event is
  T + K x N (--ack-timeout), --remote mirrors started/terminal events into
  the registry through the transition boundary.
- lib.sh send_keys_safe: an accepted-but-unconfirmed submission is confirmed
  by agent_status working or the job's own `started` (_sks_job_started);
  pane change and busy token are no longer delivery signals.
- wrapper: --heartbeat/--heartbeat-misses/--ack-timeout (MAM_* env),
  injected start command arms the timer, brief states the contract,
  _wait_started (registry / events.log / subscriber log, bounded by T and
  the wall clock) replaces the 10 s inject grace, R1 grace uses only the
  on-disk event age (herdr/pane logged once as a diagnostic),
  remote:<name> sessions skip preflight/idle/pane and dispatch over the
  broker (remote_dispatch).
- remote.py (+ scripts shim): signed retained dispatch (MAM_DISPATCH_SECRET)
  and a runner that mirrors the job into its own registry with the
  orchestrator's identity, publishes started --heartbeat, runs --exec and
  publishes the terminal event.
- loop runner: passes MAM_HEARTBEAT_SEC / MAM_HEARTBEAT_MISSES /
  MAM_ACK_TIMEOUT_S through; YAML rows with transport: remote delegate as
  remote:<name>.
- docs: MESSAGING.md 2.5 contract + remote transport, delegate-job SKILL.md,
  registry.md, loop SKILL.md.
- tests: tests/test_issue18_liveness_contract.py (16); issue 14 pane-signal
  tests flipped to started-based; issue 6 working-agent tests acknowledge
  with started; CLI snapshot/help captures refreshed.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01VCgu9Fa9iT7TbJUSzdq4Ff

.agents/skills/lib_py/delegation/job_subscriber.py

@@ -44,6 +44,8 @@ from lib_py.delegation.mqtt_common import (
 logger = logging.getLogger("delegate_job.job_subscriber")
 
 TERMINAL_EVENTS = ("completed", "error")
+TERMINAL_STATUSES = ("completed", "error", "cancelled")
+STATE_EVENT = "state"
 
 
 def _format_line(topic: str, payload: Dict[str, Any]) -> str:
@@ -133,6 +135,46 @@ class _Watcher:
         # by the transition boundary as an idempotent ack, so a replay after
         # restart cannot double-apply either.
         self.seen_intents = {}
+        # Issue #18: jobs for which at least one wire event was accepted. A seq
+        # gap is only meaningful after that - the initial last_seq comes from
+        # the registry and may be behind the wire (or ahead of a fresh remote).
+        self.accepted_any: Set[str] = set()
+
+    def _note_seq_gap(self, jid: str, expected: int, got: int, via: str, topic: str) -> None:
+        """Issue #18: a seq the subscriber never saw is a detected loss."""
+        logger.warning("seq gap for job %s: expected %d, got %d (via %s)", jid, expected, got, via)
+        mqtt_common.append_event(jid, {
+            "event": "seq_gap",
+            "expected_seq": expected,
+            "seq": got,
+            "via": via,
+            "topic": topic,
+        })
+
+    def _on_state(self, msg, payload: Dict[str, Any], jid: str) -> None:
+        """A retained `<prefix>/state` message (Issue #18): learn the job's
+        status and last seq at (re)connect without waiting for the next
+        heartbeat. Signed and token-checked like an event. Delivered to the
+        main loop as a state item so the idle window resets and a terminal
+        status ends the wait."""
+        seq = payload.get("seq")
+        if not isinstance(seq, int):
+            logger.warning("drop state for job %s: missing or invalid seq", jid)
+            return
+        last = self.last_seq.get(jid, 0)
+        if jid in self.accepted_any and seq > last + 1:
+            self._note_seq_gap(jid, last + 1, seq, "state", msg.topic)
+        if seq > last:
+            self.last_seq[jid] = seq
+        mqtt_common.append_event(jid, {
+            "event": "received",
+            "source_event": STATE_EVENT,
+            "seq": seq,
+            "topic": msg.topic,
+            "timestamp": payload.get("timestamp"),
+            "detail": "status={} last_event={}".format(payload.get("status"), payload.get("last_event")),
+        })
+        self.events.put((msg.topic, payload))
 
     def on_message(self, _client, _userdata, msg) -> None:
         # --- defensive parsing -------------------------------------------
@@ -175,6 +217,9 @@ class _Watcher:
                 logger.warning("drop event for job %s: attempt_id %r is older than expected %r",
                                jid, ev_att, expected_att)
                 return
+        if payload.get("event") == STATE_EVENT:
+            self._on_state(msg, payload, jid)
+            return
         # --- replay attack defense: check monotonic sequence ---
         seq = payload.get("seq")
         if seq is None or not isinstance(seq, int):
@@ -210,7 +255,10 @@ class _Watcher:
             logger.warning("drop event for job %s: seq %d is not monotonically increasing (last %d)",
                            jid, seq, self.last_seq.get(jid, 0))
             return
+        if jid in self.accepted_any and seq > self.last_seq.get(jid, 0) + 1:
+            self._note_seq_gap(jid, self.last_seq.get(jid, 0) + 1, seq, "events", msg.topic)
         self.last_seq[jid] = seq
+        self.accepted_any.add(jid)
         # Recorded only once the event is actually accepted. A late packet that the
         # sequence guard refuses must not mark its intent as seen, or the genuine
         # retry that follows would be mistaken for the replay of an event this
@@ -232,6 +280,22 @@ class _Watcher:
         self.events.put((msg.topic, payload))
 
 
+def _sync_remote_status(jid: str, status: str, registry_dir: str,
+                        dispatch_token: Optional[str]) -> None:
+    """Issue #18: mirror a wire event into the orchestrator's registry for a
+    job whose agent shares no disk with it. Goes through the same transition
+    boundary as the publisher's own status sync; a refused (late/duplicate)
+    transition is logged, never raised."""
+    try:
+        res = mqtt_common.transition_job_status(
+            jid, registry_dir, _source="job_subscriber", status=status,
+            expected_dispatch_token=dispatch_token)
+        if not res.committed and not res.idempotent:
+            logger.warning("remote status sync refused for %s -> %s: %s", jid, status, res.reason)
+    except Exception as exc:  # pragma: no cover - best effort
+        logger.warning("remote status sync failed for %s: %s", jid, exc)
+
+
 def _collect_jobs(args) -> List[Dict[str, Any]]:
     """Resolve the list of job records this invocation should watch."""
     if args.wait_any:
@@ -276,6 +340,13 @@ def _run_subscriber(args) -> int:
     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))
+    # Issue #18: until the first accepted event the "last event" is the
+    # injection itself, which happens after this subscriber started and may
+    # take up to the ack timeout T to produce `started` (a remote runner has
+    # to receive the dispatch first). The idle budget before the first event
+    # is therefore T + K x N; from the first event on it is K x N.
+    ack_timeout = args.ack_timeout if args.ack_timeout is not None else float(base_job.get("ack_timeout_sec", 0) or 0)
+    first_event_seen = False
 
     # All watched jobs share a broker in practice; connect using the first
     # job's broker and subscribe to each job's events topic.
@@ -283,16 +354,31 @@ def _run_subscriber(args) -> int:
     client = make_client("subscriber", config)
     client.on_message = watcher.on_message
 
+    # Issue #18: a remote job (no shared disk with the agent) has nobody else
+    # to move its registry record - the subscriber does it from the wire.
+    remote_sync = bool(args.remote) or any(j.get("transport") == "remote" for j in jobs)
+
     subscribed_topics = []
+    state_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")
+        state_topics.append(f"{prefix}/state")
+
+    connects = {"count": 0}
 
     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:
+        connects["count"] += 1
+        if connects["count"] > 1:
+            # Issue #18: paho reconnected on its own; re-subscribing here is
+            # what makes the retained state message arrive again at once.
+            logger.warning("broker reconnected (connect #%d); re-subscribing", connects["count"])
+            for jid in expected_ids:
+                mqtt_common.append_event(jid, {"event": "reconnected", "connect_count": connects["count"]})
+        for topic in subscribed_topics + state_topics:
             _c.subscribe(topic, qos=1)
             logger.info("subscribed to %s", topic)
 
@@ -361,7 +447,7 @@ def _run_subscriber(args) -> int:
                 return 2
             if not subscribed:
                 last_event = now
-            idle_left = idle_timeout - (now - last_event)
+            idle_left = (idle_timeout if first_event_seen else idle_timeout + ack_timeout) - (now - last_event)
             if idle_left <= 0:
                 _check_disk_fallback(pending, args.registry_dir, terminal, dispatch_tokens, attempts)
                 if not pending:
@@ -386,10 +472,30 @@ def _run_subscriber(args) -> int:
 
             last_event = time.monotonic()
             retention_checked = True
-            print(_format_line(topic, payload), flush=True)
+            first_event_seen = True
 
             jid = payload["job_id"]
             event = payload.get("event")
+            if event == STATE_EVENT:
+                # Issue #18: retained job state. Any state resets the idle
+                # window (the agent is alive enough to have published); a
+                # terminal status ends the wait exactly like the terminal event.
+                status = payload.get("status")
+                if status in TERMINAL_STATUSES and jid not in terminal:
+                    synth = "completed" if status == "completed" else "error"
+                    payload = dict(payload, event=synth, source="state",
+                                   detail="terminal status '{}' recovered from retained state".format(status))
+                    print(_format_line(topic, payload), flush=True)
+                    terminal[jid] = synth
+                    pending.discard(jid)
+                    if remote_sync:
+                        _sync_remote_status(jid, status, args.registry_dir, dispatch_tokens.get(jid))
+                continue
+
+            print(_format_line(topic, payload), flush=True)
+            if remote_sync and event in ("started", "completed", "error", "cancelled"):
+                _sync_remote_status(jid, {"started": "running"}.get(event, event),
+                                    args.registry_dir, dispatch_tokens.get(jid))
             if event in TERMINAL_EVENTS:
                 if jid in terminal:
                     # Already finalised: ignore duplicates / late reorders.
@@ -421,6 +527,9 @@ def main(argv=None) -> int:
                         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("--ack-timeout", type=float, default=None,
+                        help="extra idle budget before the first event, for the injection to "
+                             "produce `started` (default: job.ack_timeout_sec; Issue #18)")
     parser.add_argument("--expect-retention", action="store_true",
                         help="warn if no retained terminal event arrives promptly")
     parser.add_argument("--expect-dispatch-token", default=None,
@@ -428,6 +537,10 @@ def main(argv=None) -> int:
     parser.add_argument("--attempt-id", type=int, default=None,
                         help="minimum attempt id to accept events for")
     parser.add_argument("--registry-dir", default=default_registry_dir())
+    parser.add_argument("--remote", action="store_true",
+                        help="the agent shares no disk with this registry: mirror started/"
+                             "terminal events into the job record (Issue #18; automatic "
+                             "for jobs registered with --transport remote)")
     parser.add_argument("-v", "--verbose", action="store_true")
     args = parser.parse_args(argv)