multi-agent-mux/MESSAGING.md

Messaging System Technical Analysis & Architecture Report

Applies to: MAM v4.8.1

This report provides a comprehensive, deep-dive analysis of the messaging system implemented in the multi-agent-mux-delegate-job skill. It covers the MQTT broker architecture, event protocols, job lifecycles, codebase internals, cross-system integration, and a list of known limitations along with production recommendations.

---

1. MQTT Broker Architecture: PoC vs. TLS Production

The messaging system is designed with a clear, decoupled transition pathway from a Proof of Concept (PoC) public broker setup to a secured, authenticated, and encrypted private production cluster. All configurations are resolved dynamically from the environment or overridden at the job level, requiring zero code modifications during deployment cut-over.

1.1 PoC Architecture (Public Sandbox)

In the initial development/testing phase, the system defaults to the public broker hosted by HiveMQ:

• Host/IP: broker.hivemq.com
• Protocol/Port: Plaintext MQTT over TCP on port 1883.
• Security & Auth: None. No username, password, TLS encryption, or access control list (ACL) constraints are applied.
• QoS Level: QoS 1 (At Least Once) is requested for publishes and subscriptions, ensuring acknowledgement at the network layer.

Risks and Limitations of the PoC Setup:

1. Zero Eavesdropping Protection: Because the broker is public and unencrypted, any internet user can subscribe to the root topic (python/mqtt/jobs/#) and read the exact prompt, agent sessions, and intermediate progress events.
2. Event Spoofing & Injection: Anyone can publish messages to any job topic. An attacker could publish a malicious completed or error event, prematurely terminating a running subscriber or causing the delegator to execute unauthorized post-validation hooks.
3. No Message Persistence: Public brokers do not guarantee queue persistence or durable sessions for disconnected clients. If a subscriber briefly drops offline, QoS 1 messages published during the disconnect window may be discarded.
4. Rate Limiting & Reliability: Public sandboxes are subject to arbitrary rate limits, traffic spikes, and transient connection resets, leading to network-level timeouts.

---

1.2 Production Architecture (Secure Private NATS Broker)

For production deployments, the system standardizes on a private, self-hosted NATS server (nats:2.12-alpine) with its built-in MQTT 3.1.1 protocol engine and JetStream persistence enabled, managed via nats-docker/docker/docker-compose.yaml.

graph TD
    subgraph "Secure Tailnet / Corporate Network"
        Broker["Private NATS Broker (nats:2.12-alpine) <br> Native: 4222 | MQTT: 1883 | WS: 8080"]
        
        subgraph "Hermes (Delegator/Orchestrator)"
            SubClient["job_subscriber.py <br> (Role: subscriber)"]
        end
        
        subgraph "Herdr Workspace (Agent Host)"
            PubClient["publish_event.py <br> (Role: publisher)"]
        end
        
        SubClient -- "Subscribe (QoS 1) <br> Auth: mam_agent / mam_observer <br> ACL: Read python/mqtt/jobs/+/events" --> Broker
        PubClient -- "Publish (QoS 1 + Retain Terminal) <br> Auth: mam_agent <br> ACL: Write python/mqtt/jobs/+/events" --> Broker
    end

Production Security & Hardening Controls:

1. Transport Layer Security & Overlay Networks: Within a trusted mesh (Tailscale / Tailnet, Model T), traffic routes over encrypted WireGuard overlays to private endpoints. For public WAN exposures (Model P), TLS v1.3 encryption is terminated via private CA certificates (MQTT_CA_CERTS), and mutual TLS (mTLS) is supported via client keypairs (MQTT_CERTFILE / MQTT_KEYFILE).
2. Strict Client Authentication & Multi-Tenancy: All clients authenticate against isolated NATS accounts (MAM, HOME, SYS) using dedicated credentials (MQTT_USERNAME / MQTT_PASSWORD). Anonymous access is explicitly disabled.
3. Role-Based Topic Access Control Lists (ACLs):
   • Worker / Agent (mam_agent): Granted full publish/subscribe access within the MAM account to manage job lifecycles:

     # nats-docker/docker/nats.conf
     accounts {
       MAM: {
         jetstream: enabled
         users: [
           { user: mam_agent, password: $MAM_BROKER_PASS }
         ]
       }
     }
     

   • Observer / Dashboard (mam_observer): Restricted to read-only access for monitoring streams while strictly preventing unauthorized command injection:

     # nats-docker/docker/nats.conf
     { user: mam_observer, password: $MAM_OBSERVER_PASS,
       permissions: {
         subscribe: { allow: ["python.mqtt.jobs.>"] }
         publish:   { deny:  [">"] }
       }
     }
     

4. Durable Message Queues & Session State:
   • JetStream is activated with a dedicated persistent store path (store_dir: "/data"), backing MQTT QoS 1 streams and persistent client sessions.
5. Retained Terminal Events: Terminal events (completed / error) are published with retain=True. NATS stores retained payloads in JetStream, allowing late-joining subscribers to instantly recover final states without polling.

---

1.3 NATS JetStream, Retained Messages & WebSocket Integration

The production deployment in nats-docker/docker/nats.conf includes key architectural primitives:

1. JetStream Requirement for MQTT Engine: nats-server requires JetStream enabled at both the server level and the account level (jetstream: enabled) for MQTT sessions and QoS 1 message persistence.
2. Retained Message Scope Boundary (N-1): Retained messages published via MQTT are stored in JetStream by NATS and delivered to subsequent MQTT subscribers. Note that native NATS pub/sub subscribers do not receive historical retained messages upon connection unless queried via JetStream KV/Object APIs.
3. MQTT-over-WebSocket /mqtt Path (N-7): For web dashboards and browser clients, NATS exposes WebSocket listeners on port 8080 (or 443 in TLS mode). Standard MQTT-over-WebSocket clients connect to the /mqtt path (e.g. ws://<host>:8080/mqtt or wss://<host>:8443/mqtt), with no_tls: true and same_origin: false configured for secure cross-origin streaming behind reverse proxies.
4. Remote Deployment Models: For full installation, Tailscale topology, and secret management guides, refer to nats-docker/PRIVATE_SERVER.md and nats-docker/NATS_REPORT.md.

---

1.4 Alternative: Hardened Mosquitto Reference

If an environment requires a dedicated Mosquitto broker instead of NATS, a reference /etc/mosquitto/mosquitto.conf configuration is maintained:

# Persistence settings
persistence true
persistence_location /var/lib/mosquitto/

# Authentication and Authorization
password_file /etc/mosquitto/auth/passwd
acl_file      /etc/mosquitto/auth/acl
allow_anonymous false

# Listener and TLS Configuration
listener 8883
cafile   /etc/mosquitto/certs/ca.crt
certfile /etc/mosquitto/certs/server.crt
keyfile  /etc/mosquitto/certs/server.key
tls_version tlsv1.3

---

2. Event Protocol Specification

The event protocol defines a strict, single-direction JSON wire schema. It acts as the contract between the worker agent (the publisher) and the delegator/orchestrator (the subscriber).

2.1 Wire Schema (JSON UTF-8, schema_version = 1)

Every event payload must adhere to the following schema structure:

{
  "schema_version": 1,
  "seq": 2,
  "job_id": "918b0612",
  "event": "progress",
  "timestamp": "2026-06-20T14:48:58Z",
  "detail": "Section 1: MQTT Broker Architecture completed",
  "data": {
    "auth_token": "URL-safe-base64-random-token-here",
    "custom_metric": 42
  }
}

2.2 Wire Schema Field Dictionary

Field	Type	Required	Description & Validation Rules
schema_version	Integer	Yes	Must be exactly 1. Subscribers discard payloads with mismatched version numbers to prevent parser crashes on schema drift.
seq	Integer	Yes	Monotonic sequence counter starting at 1 for the first publish. Incremented and stored in the job's registry file (last_seq) to survive agent pane crashes.
job_id	String	Yes	The 8-character hex string identifying the target job. Subscribers discard any messages whose job_id is unexpected or unrequested.
event	String	Yes	The event classification: started, progress, permission_required, completed, or error.
timestamp	String	Yes	ISO-8601 UTC timestamp with a trailing Z suffix. (Advisory only; never trusted for timeouts).
detail	String	Yes	Generalized, safe text description. Strict rules prohibit absolute paths, workspace paths, passwords, or raw environment variables.
data	Object	Yes	Metadata dictionary. Used in production to pass auth_token, dispatch_token (opaque UUID for worker attempt binding), attempt_id (monotonic integer attempt sequence), or structured execution metrics.

---

2.3 Event Type Dictionary and Schemas

1. started

• Emit Trigger: Emitted by the worker agent immediately upon boot inside the herdr session, indicating it has parsed the instructions and started execution.
• Payload Constraints: seq must be 1. Status in registry is transitioned to running.
• Example Detail: "Job 918b0612 started"

2. progress

• Emit Trigger: At least every heartbeat_sec while working (Issue #18 heartbeat obligation; publish_event.py --event started --heartbeat N automates it with detail: heartbeat), plus any check-point the agent wants to report.
• Payload Constraints: None.
• Example Detail: "Section 1: MQTT Broker Architecture completed"

3. permission_required

• Emit Trigger: Emitted when the agent needs human confirmation (e.g. to run a destructive command or read/write critical system files).
• Payload Constraints: detail contains the resource/action requested.
• Example Detail: "needs write permission to MESSAGING.md"

4. completed (Terminal)

• Emit Trigger: Successful job completion. The agent has generated all expected artifacts and verified correctness.
• Payload Constraints: Must be the final event. Published with retain=True.
• Example Detail: "deep report written and committed to git"

5. error (Terminal)

• Emit Trigger: Terminal failure. Agent encountered an unhandled exception, syntax error, or validation script fail.
• Payload Constraints: Must be the final event. Published with retain=True.
• Example Detail: "validation fail: missing files"

---

2.4 Integrity and Authentication Verification (HMAC-SHA256 Signatures)

To prevent unauthorized users from hijacking or spoofing events on public brokers:

1. When a job is registered, a cryptographic token (auth_token) is generated (secrets.token_urlsafe(32)).
2. The publisher reads this token and signs the JSON payload. Specifically, the publisher calculates an HMAC-SHA256 signature using the auth_token as the secret key over the serialized payload (with the hmac_sig field excluded).
3. The signature is attached as data.hmac_sig on the wire.
4. The subscriber (job_subscriber.py) reads the expected auth_token from the local registry and verifies the HMAC signature. Any message with a missing, invalid, or mismatched signature is discarded immediately with an "HMAC verify failed" log.
5. To prevent event drops, all publishers and subscribers must be updated simultaneously during deployment rollout, since the plaintext auth_token is never transmitted on the wire to prevent token interception.

---

2.5 Liveness and Delivery Contract (Issue #18)

No new event types: the existing events are promoted to a contract with obligations, periods and timeouts, so that an agent's life and the delivery of its prompt are decided by what the agent publishes, never by scraping a terminal (herdr agent_status, pane text, screen changes). This is what lets an agent on another host, which shares neither the pane nor the orchestrator's disk, take part on equal terms.

Rule	Contract	Where
Heartbeat obligation	While working, the agent publishes progress at least every N s (heartbeat_sec, default 30). publish_event.py --event started --heartbeat N starts a detached timer that publishes progress (detail: heartbeat) every N s with the same dispatch_token/attempt_id, so the agent does not have to remember. No heartbeat can follow a terminal event: the terminal publish first raises an attempt-scoped barrier (<registry>/<job>.heartbeat.barrier, written under the registry lock that also allocates seqs, so a beat either took its seq before it or is refused), then stops the timer through <registry>/<job>.heartbeat.pid and waits for it to drain its in-flight beats (killing it after MAM_HEARTBEAT_DRAIN_S, default 1 s - a beat still on the wire is cancelled, never allowed to trail the terminal), and only then takes its own seq; replacing the timer (another started --heartbeat for the job) drains or kills the previous one the same way before it hands over the pidfile. Starts, replacements and stops of a job's timer are serialized by a per-job lifecycle lock (<registry>/<job>.heartbeat.lock, separate from the seq lock): a start that finds the attempt ended arms nothing, an exiting timer never waits for that lock, and a timer whose pid could not be recorded is killed at once. The timer also stops itself on a terminal record or barrier, on re-dispatch, after timeout_sec, and within one period of the agent process it watches going away (--watch-pid, default the nearest agent-CLI ancestor of the publishing command; a remote runner watches itself) - a crashed agent stops heartbeating, an alive-but-stuck one does not.	lib_py/delegation/heartbeat.py, publish_event.py --heartbeat
Unresponsive	The orchestrator treats a job as unresponsive after K x N s without any event (heartbeat_misses, default 4 → 120 s; --idle-timeout overrides). K x N is when silence is detected; the wrapper's post-idle grace below can stretch the final close to about 2 x K x N. The subscriber's idle clock counts every accepted event, and a retained state only when its seq is newer than every event and state already accepted - the same snapshot re-delivered on a reconnect, or an older one, is not liveness; the wrapper's post-idle grace re-reads only the job's events.log (_recent_disk_heartbeat, written before every wire publish). herdr status and the pane are logged once as a diagnostic (liveness: ... herdr/pane says busy=yes/no (diagnostic only)) and never decide.	multi-agent-mux-delegate-job (finalize_unfinished), job_subscriber.py
Delivery	After the prompt is handed over (herdr accepted the submission, or the remote dispatch was published), the agent's started must arrive within T s (ack_timeout_sec, default 180; a real agent reads the brief before it runs the start command): the registry leaves pending, a started line lands in events.log, or the subscriber prints one. No started within T → the inject fails closed: rc 1, registry error (agent_blocked), and the prompt is never re-injected. send_keys_safe confirms an accepted-but-unconfirmed submission only by agent_status == working or the job's started; pane changes and busy tokens are no longer signals.	_wait_started, lib.sh: _sks_job_started
Termination	Only completed / error (and an operator cancelled) end a job. Terminal immutability (Issue #6 CAS) refuses anything that arrives after deadline_timeout closed the job.	engine.py
Wire durability	Every event (QoS 1) is followed, on the same connection, by a retained <topic_prefix>/state message (event: state, status, last_event, seq, signed like an event). A subscriber that connects or reconnects receives it at once: a terminal status ends the wait, a state whose seq is newer than every event and state already accepted resets the idle window (a re-delivered or older snapshot still recovers a terminal status but is not liveness), and a seq beyond the last one seen is recorded as seq_gap in the audit log (also for gaps between events). paho reconnects on its own; on_connect re-subscribes both topics and logs reconnected.	publish_event.py: build_state_payload, job_subscriber.py: _on_state
Credentials	Each host connects with its own .mam.env (Issue #12: the password is never in a record). Every event and state message carries the job's HMAC (auth_token). A remote dispatch is signed with the pre-shared MAM_DISPATCH_SECRET.	mqtt_common.py, remote.py
What screen scraping is still for	(a) operations that need the screen: prompt injection into a pane, trust/permission dialogs, the pre-injection idle handshake; (b) diagnostics on the same host when events stop. Neither is a liveness or delivery decision.	lib.sh

Registry record fields: heartbeat_sec, heartbeat_misses, ack_timeout_sec, idle_timeout_sec (derived K x N unless explicit), transport (herdr / remote). Wrapper options: --heartbeat, --heartbeat-misses, --ack-timeout (also MAM_HEARTBEAT_SEC, MAM_HEARTBEAT_MISSES, MAM_ACK_TIMEOUT_S for the loop runner).

Remote transport (--agent-session remote:<name>)

orchestrator ── retained <prefix>/dispatch (signed) ──▶ remote runner (other host)
orchestrator ◀── <prefix>/events + retained <prefix>/state ── remote runner / agent

remote.py dispatch publishes the brief, the job's auth_token, dispatch_token, attempt_id and the contract as one retained, signed message. remote.py run --session <name> [--exec CMD] on the other host mirrors the job into its own registry (same job id and identity), retires the retained dispatch, publishes started --heartbeat N, runs the command (env: JOB_ID, MAM_JOB_BRIEF, MAM_PUBLISH_CMD, ...) and publishes the terminal event unless the agent already did. The orchestrator's subscriber runs with --remote (automatic for transport: remote jobs) and mirrors started and terminal events into the orchestrator's registry through the transition boundary, because nothing else on that host will. The wrapper takes no herdr/pane/lsof/ps path for a remote session. Because the dispatch carries the job's auth_token, use the remote transport only on a private broker.

---

3. Job Lifecycle & State Transitions

The lifecycle of a delegated job progresses through a highly coordinated state machine, involving file-based registry claiming, asynchronous message subscription, and multi-faceted event publishing.

stateDiagram-v2
    [*] --> pending : register_job()
    pending --> running : pick_pending()
    running --> completed : publish_event(--event completed)
    running --> error : publish_event(--event error)
    running --> cancelled : update_status(..., cancelled)
    pending --> cancelled : update_status(..., cancelled)
    completed --> [*]
    error --> [*]
    cancelled --> [*]

3.1 Step-by-Step Lifecycle Phase Details

Phase 1: Registration (register)

• Trigger: A delegator triggers registry.py register (or the multi-agent-mux-delegate-job submit command).
• Registry State: Flips from non-existent to pending inside .mam/jobs/<job_id>.json. A last_seq counter is initialized to 0.
• Locking: Exclusive fcntl file lock acquired over .lock during write.
• Durable Audit Log: Writes <logs>/<job_id>/meta.json, sets status to pending in status.json, and appends a registered event line to events.ndjson.

Phase 2: Claiming (pick_pending)

• Trigger: An agent session starts up and calls registry.py pick --agent-session <session_label>.
• Registry State: Oldest matching pending record is scanned. Status is atomically updated to running. updated_at is stamped.
• Locking: Reads and writes occur inside the exclusive fcntl lock block.
• Durable Audit Log: Status is synced to running in status.json and a status_changed event is appended to events.ndjson.

Phase 3: Listening (subscribe)

• Trigger: The wrapper command launches job_subscriber.py --job <job_id> in the background before launching the agent.
• Broker Connection: Connects to the resolved host, issues a QoS 1 subscription to python/mqtt/jobs/<job_id>/events, and blocks on an event queue.
• Timeout Initialization: Dual timeouts (wall-clock budget and activity idle timer) are calculated and start ticking.

Phase 4: Execution & Progress Events (publish)

• Trigger: The agent executes prompts within herdr and runs publish_event.py at boot and checkpoint stages.
• Network Handshake: Publisher opens a fresh TCP/TLS socket to the broker, awaits CONNACK, publishes a single QoS 1 message, waits for PUBACK, and gracefully disconnects to avoid socket resource leaks.
• State Updates: Updates last_seq monotonically, updates status to running (if not already), and mirrors the published payload into the local audit logs (events.ndjson).
• Subscriber Capture: The subscriber captures the payload, performs bearer token checks, prints the formatted line to stdout, and resets its idle timer.

Phase 5: Terminal Finalization

• Trigger: Agent publishes --event completed or --event error.
• Registry Transition: State becomes completed or error.
• Retained Messaging: The terminal event is published with retain=True on the broker.
• Subscriber Exit: The subscriber processes the terminal event exactly once, terminates its background loops, and exits (code 0 for completed, 1 for error).

---

4. Code Internals Analysis

The core implementation logic of the delegation system resides within the Python package .agents/skills/lib_py/delegation/ (mqtt_common.py, job_subscriber.py, engine.py, cli.py, registry.py, publish_event.py, etc.). The entry scripts located in .agents/skills/multi-agent-mux-delegate-job/scripts/*.py (registry.py, publish_event.py, job_subscriber.py, mqtt_common.py) serve as wire-compatible shims that delegate directly into lib_py/delegation/.

4.1 registry.py & lib.sh (Locking & Atomicity)

• Underlying Module: .agents/skills/lib_py/delegation/registry.py (shim: .agents/skills/multi-agent-mux-delegate-job/scripts/registry.py) and .agents/skills/lib.sh.

Two concurrency control schemes co-exist in this workspace to coordinate state modification:

1. lib.sh::atomic_dump_yaml(): Used for workspace-wide herdr session inventory (agent-sessions.yaml).
   • Locking: Uses SQLite database transaction serialization via BEGIN IMMEDIATE on agent-sessions.db.
   • Safe Mutation: The mutation source code is passed in an environment variable AGENT_SESSIONS_MUTATION and executed dynamically using exec(compile(..., 'exec'), globals()). This isolates the execution and avoids command-injection vectors.
   • Atomicity: Updates the SQLite tables and then, if a session transitions to a finished state, writes to a temp file in the same directory using tempfile.mkstemp() and performs an os.replace() rename. POSIX guarantees the replacement is atomic, preventing half-written YAML reads. A .bak backup copy is also preserved.
2. registry.py::register_job() / pick_pending() / _atomic_write_record(): Used for job-level metadata JSON files (<job_id>.json).
   • Locking: Wraps operations in a registry_lock(registry_dir) context manager, implementing an advisory exclusive lock on .lock via fcntl.flock.
   • Atomicity: In _atomic_write_record(), it uses tempfile.mkstemp inside the parent registry folder, serializes the updated job record to the temp file, flushes it, triggers a physical disk sync via os.fsync(fh.fileno()), and executes os.replace to replace the main JSON record file. The file permission is restricted to 0o600 immediately.

---

4.2 publish_event.py (Retries and Handshakes)

• Underlying Module: .agents/skills/lib_py/delegation/publish_event.py (shim: .agents/skills/multi-agent-mux-delegate-job/scripts/publish_event.py).

The publisher script enforces robust error handling and fail-safe local persistence:

• Fresh Connection Pattern: Instead of maintaining a persistent socket connection (which is susceptible to socket timeouts or channel leaks), publish_event.py opens a fresh socket, completes the authentication/TLS handshake, publishes a single QoS 1 event, waits for PUBACK, and closes the connection.
• Guaranteed Disk Synchronization (B-14): Before attempting any network transmission over MQTT, publish_event.py records the event into the local registry (append_event and update_job_status). If the broker is unreachable or network publish fails, local audit logs and state machine files remain 100% accurate. The script returns exit code 2 at the very end to signal a transport failure without corrupting local state.
• Exponential Backoff: Wrapped in the with_retry() decorator from mqtt_common.py. In case of socket errors (OSError, TimeoutError, ConnectionError), it retries up to 3 times (configurable via --attempts) with backoff: \text{delay} = \min(\text{base\_delay} \times \text{factor}^{\text{attempt}-1}, \text{max\_delay}) Default parameters: base_delay = 0.5s, factor = 2.0, max_delay = 8.0s.
• ACK Handshake Deadlines:
  • CONNECT_ACK_TIMEOUT = 10s (stops hangs during broker downtime).
  • PUBLISH_ACK_TIMEOUT = 5s (guarantees QoS 1 message acknowledgment before marking as published).

---

4.3 job_subscriber.py (Timers and Queue Semantics)

• Underlying Module: .agents/skills/lib_py/delegation/job_subscriber.py (shim: .agents/skills/multi-agent-mux-delegate-job/scripts/job_subscriber.py).

The subscriber acts as the central execution watchdog:

• Queue Serialization: Uses a thread-safe queue.Queue internally. The Paho MQTT callback thread adds messages to the queue, and the main thread processes them sequentially. This separates network I/O from state machine validation.
• Local Disk Fallback Verification (B-15): On initial startup and upon any broker connection failure, _check_disk_fallback() immediately queries local job records (.mam/jobs/<job_id>.json) and audit logs (status.json). If the target job has already reached a terminal state locally, the subscriber completes immediately without waiting on a dead broker.
• Infrastructure Error Code Separation (F-4): The subscriber returns distinct exit codes:
  • Exit 0: Job completed successfully.
  • Exit 1: Job terminated with an application error event.
  • Exit 2: Activity idle or wall-clock timeout exceeded.
  • Exit 3: Broker infrastructure connection error (with disk fallback checked).
• State Machine Protection: To safeguard against QoS 1 duplicate delivery or out-of-order broker retries, the subscriber runs a terminal state machine. It records job completion in an internal terminal dictionary. Once a job is marked completed or error, any subsequent events for that job_id are ignored:

  if event in TERMINAL_EVENTS:
      if jid in terminal:
          logger.info("ignoring duplicate terminal %s for %s", event, jid)
          continue
      terminal[jid] = event
      pending.discard(jid)
  

• Dual Timeout Semantics:
  1. Wall-Clock Timeout: Calculated relative to absolute startup time (wall_deadline = start + wall_timeout). It acts as a hard budget limit, guarding against an agent hanging indefinitely.
  2. Activity Idle Timeout: Measured as the difference between the current monotonic time and the last packet arrival time (idle_left = idle_timeout - (now - last_event)). If the agent fails to publish any event (heartbeats and retained state included) for the duration of the idle window - K x N under the Issue #18 contract - the subscriber aborts and exits with code 2.
• Idle budget before the first event (Issue #18): a standalone subscriber allows ack_timeout_sec + idle_timeout_sec before the first accepted event (the injection has to produce started). The delegate wrapper passes --ack-timeout <wall clock> instead: it judges delivery itself with _wait_started (T from the injection, which happens after the subscriber is up) and reaps the subscriber when that fails, so the subscriber's own idle clock only starts with the first event.
• Retained state and seq gaps (Issue #18): the subscriber also subscribes to <topic_prefix>/state; a retained terminal state resolves the job at (re)connect, a state renews the idle clock only when its seq is newer than every event and state already accepted (a re-delivered or older snapshot does not), and a seq jump on either topic is written to the audit log as seq_gap. With --remote (or transport: remote) accepted started/terminal events are mirrored into the local job record.

---

4.4 mqtt_common.py (Logging, Env Vars & Config Resolution)

• Underlying Module: .agents/skills/lib_py/delegation/mqtt_common.py (shim: .agents/skills/multi-agent-mux-delegate-job/scripts/mqtt_common.py).

• Log Routing Isolation: Configured via setup_logging(). The root logger is bound to sys.stderr. This preserves the standard output stream (stdout) exclusively for clean JSON-lines payloads, enabling downstream bash tools to pipeline event feeds cleanly (e.g., job_subscriber.py ... | jq).

• Environment Variable Dictionary: The system parses and supports the following 10 configuration variables:

  Environment Variable	Default	Purpose
  MQTT_BROKER	broker.hivemq.com	Broker hostname or IP address (e.g., vm-ubuntu, 127.0.0.1)
  MQTT_PORT	1883	Broker port (1883 for plaintext/Tailscale, 8883 for TLS)
  MQTT_TLS	false	Enable TLS encryption (true / false / 1 / 0)
  MQTT_USERNAME	""	Authentication username (e.g., mam_agent, mam_observer)
  MQTT_PASSWORD	""	Authentication password
  MQTT_CA_CERTS	""	Path to CA certificate bundle for TLS verification
  MQTT_CERTFILE	""	Path to client certificate for mutual TLS (mTLS)
  MQTT_KEYFILE	""	Path to client private key for mutual TLS (mTLS)
  MQTT_CLIENT_ID_PREFIX	hermes	Prefix for dynamically generated random client IDs
  MQTT_KEEPALIVE	60	MQTT keepalive ping interval in seconds

• .mam.env Resolution Hierarchy (_load_dotenv): Configuration files are resolved with strict precedence rules:

  1. OS Environment Precedence: Any variable already defined in os.environ is preserved and never overwritten by file-based configs.
  2. Explicit Override (MAM_ENV_FILE): If MAM_ENV_FILE is set, only that specific file is parsed. If the specified file does not exist, it is skipped silently and ambient search is refused (no fallback to unintended parent configs): the process runs on its environment and the defaults. If it exists but cannot be read or decoded, that is an error (EnvFileError): the wrapper and the Python entry points print one ERROR: <path>: cannot read (<exception type>) line and exit 1.
  3. Workspace Root Auto-Discovery: If MAM_ENV_FILE is not set, the resolver searches candidate paths in order: MAM_REAL_ROOT, WORKSPACE_ROOT, upward directory walk searching for .agents or .git boundary markers, and os.getcwd().
  4. Public Broker Security Alert (B-17): If the final resolved host falls back to the public sandbox broker.hivemq.com, a prominent security warning is emitted.

• Broker Config Resolution (broker_config_from_job):

  1. Loads baseline settings from environment / .mam.env.
  2. Overlays job-specific overrides specified inside the job record JSON block (broker.*).

---

5. Cross-System Integration

The delegated messaging system functions as a critical control backplane, binding shell wrappers and monitoring loops across the orchestration stack.

graph LR
    User["User/Cron Client"] -->|submit| Wrap["multi-agent-mux-delegate-job (Bash)"]
    Wrap -->|registers| Reg["registry.py (Live Registry)"]
    Wrap -->|spawns background| Sub["job_subscriber.py"]
    Wrap -->|spawns herdr pane| Herdr["herdr Session (Agent Pane)"]
    
    Herdr -->|executes agent| Agent["Claude / Codex Agent"]
    Agent -->|publish_event.py| Broker["MQTT Broker"]
    Broker -->|delivers events| Sub
    Broker -->|delivers events| Mon["reconcile.sh (Monitor Loop)"]
    
    Mon -->|updates| Inv["agent-sessions.yaml <br> (lib.sh::atomic_dump_yaml)"]

5.1 Orchestration Wrappers (multi-agent-mux-*)

The system integrates a Compare-And-Swap (CAS) state transition engine implemented in .agents/skills/lib_py/delegation/engine.py (MAM_ENGINE_CAS=1 by default). The engine stores a fresh opaque dispatch_token (uuid4().hex) and attempt_id (monotonic integer) when a job is registered, claimed, or redispatched. Operator sources (source ∈ {registry_cli, operator, delegate_wrapper}) carry operator-level authority for lifecycle transitions (redispatch, finalize, close). For redispatched jobs (attempt_id > 1), terminal requests (completed, error, cancelled) from non-operator sources strictly require a valid pinned dispatch_token (or proved expected_attempt). Orchestration wrappers inject --expect-dispatch-token and --attempt-id into worker prompt instructions (publish_event.py) and subscriber daemons (job_subscriber.py). Any stale, redispatched, or orphaned worker events are refused before mutating registry status or emitting wire packets. Committed transitions are ordered separately by the monotonic transition_seq counter. The Python core implementation is guarded by the MAM_DELEGATION_PY=1 feature flag (enabled by default), which provides a transparent fallback rollback switch to legacy shell/shim execution paths if set to 0.

1. multi-agent-mux-delegate-job (submit):
   • Registers a job, spawns job_subscriber.py to capture standard output streams to .mam/jobs/<job_id>.subscriber.out (loop iterations use .mam/jobs/<job_id>.iter_<n>_<role>.subscriber.out).
   • Attempts a subscriber readiness handshake by polling for SUBSCRIBED every 0.2 seconds for up to 25 iterations. If readiness is not observed before the polling budget expires, it warns and proceeds with agent launch. Early subscriber exits with status 0 or 3 also permit continuation.
   • Boots the agent pane in herdr:

     herdr new-session -d -s "$sess" -c "$WORKDIR" \
       "printf '%s' \"$instructions\" | $bin --dangerously-skip-permissions; echo; read"
     

   • Pre-seeds agent instruction headers via stdin to enforce that the agent runs publish_event.py for its transitions.
   • Blocks on wait $sub_pid, and finally prints the audit log directory.
   • Subscriber Timeout and Fail-Closed Finalization: When the subscriber exits, finalize_unfinished is invoked across all termination paths (subscriber timeout, broker disconnection, inject failure, or preflight failure). Unfinished jobs are transitioned via registry.py close --request deadline_timeout|cancelled to guarantee terminal status in the registry and prevent zombie jobs from remaining in pending or running. If preflight checks or TUI idle handshakes fail before prompt delivery, the job is closed as cancelled with detail session_busy (exit code 5), allowing safe orchestration reschedule.
2. multi-agent-mux-monitor (reconcile.sh):
   • Wildcard Monitor Integration: Runs a unified background subscriber loop (reconcile.sh --subscribe) to capture progress, verify security tokens (HMAC) and sequences, write audit logs, and automatically clean up herdr sessions upon terminal events.
   • Reconciliation loop: Subscribes to the global job topic. On terminal events, it invokes lib.sh::atomic_dump_yaml to sync status drifts (e.g. setting herdr sessions to terminated in agent-sessions.yaml once the agent exits).
3. multi-agent-mux-create / stop / resume:
   • Integrates the job life status into session metadata updates, ensuring standard herdr cleanup triggers state updates in the registry and audit logs.

---

6. Known Limitations & Recommendations

6.1 Limitations

1. Single-Host File Locking Vulnerability: The advisory locking system previously relied heavily on fcntl.flock. While agent-sessions.yaml has been migrated to SQLite WAL to solve concurrent writes, the job metadata in .mam/jobs/ still relies on fcntl.flock which may behave non-atomically on NFS.
2. Bearer Token Leakage over Plaintext (Public Broker): The auth_token mechanism is a simple plaintext bearer comparison. If the transport layer is unencrypted (e.g., using broker.hivemq.com on port 1883), any eavesdropper on the network can steal the token and spoof legitimate events.
3. Subscriber Network Drop & Disk Fallback (Resolved via B-15 / Residual Active Reconnection Gap): job_subscriber.py implements on-disk status fallback (_check_disk_fallback) to recover state upon broker connection loss (B-15). An active in-session auto-reconnection loop during continuous execution remains a recommended enhancement.
4. Lack of Ordering Guarantees in QoS 1: QoS 1 guarantees delivery but not strict ordering. Under heavy backoff retries, a late-delivered progress event could land after a terminal event, causing state inconsistencies.

---

6.2 Recommendations

1. [Implemented] Migrate to SQLite WAL Backend: The agent-sessions.yaml locking mechanism in lib.sh has been upgraded to use a SQLite database (agent-sessions.db) configured with Write-Ahead Logging (PRAGMA journal_mode=WAL). The YAML file is now only updated as a finalized archive when a session reaches a terminal state (stopped, terminated, archived), eliminating flock contention during active session updates. Architecture Decision Note: This means agent-sessions.yaml is no longer a real-time view of currently running sessions. We have explicitly accepted the trade-off of giving up real-time text readability of running sessions in favor of robust concurrency and solving NFS flock limits. Tooling and status checks must now query the SQLite DB to observe live running states.
2. [Implemented] Implement Signature-Based Payload Verification: HMAC payload verification has been implemented using SHA-256 signatures (auth_token generation and verification). The delegator and worker share a secret key; the publisher generates an HMAC-SHA256 signature over the payload bytes and timestamp, which is attached as data.hmac_sig on the wire and verified by the subscriber (see §2.4 Integrity and Authentication Verification (HMAC-SHA256 Signatures)).
3. Enforce Mandatory NATS Broker-Side Authentication, JetStream and ACLs: Standardize on private nats-server with JetStream and account-level ACL isolation (nats-docker/docker/nats.conf). For public WAN exposures, terminate TLS v1.3 (MQTT_TLS=true) over port 8883.
4. Build Auto-Reconnecting Subscriber Loops: Upgrade job_subscriber.py to handle disconnect callbacks. Maintain a persistent queue in memory and allow the client to reconnect with exponential backoff, preventing socket dropout from terminating the orchestration flow.

---

Glossary: Session States vs Job States

This project manages two distinct state domains that are often confused:

Session States (YAML — .mam/agent-sessions.yaml)

Managed by .agents/skills/lib.sh and the 6 multi-agent-mux-* skills. Valid values (see lib.sh valid-status set):

State	Meaning	Set by
running	herdr session active, agent running	create, resume
stopped	stopped via multi-agent-mux-stop (default); conversation preserved for resume	stop
terminated	stopped with --purge-conversation, or herdr-dead detected; conversation deleted / session gone	stop --purge-conversation, monitor reconcile
archived	legacy value — no producer since --mode soft was removed; kept in the validation whitelist for rows written by older versions	(none)

Job States (Registry — .mam/jobs/<id>.json)

Managed by .agents/skills/multi-agent-mux-delegate-job/scripts/registry.py. Valid values:

State	Meaning	Set by
pending	job registered, agent not yet started	registry.py register
running	agent picked up the job, publishing events	publish_event.py --event started
completed	terminal event — agent finished successfully	publish_event.py --event completed
error	terminal event — agent failed	publish_event.py --event error
cancelled	job cancelled by orchestrator	registry.py cancel

Key distinction: Session states track the herdr container lifecycle (create→stop→resume). Job states track the delegated work lifecycle (submit→run→complete/error). A single session can host multiple sequential jobs; a job runs within exactly one session.