# Messaging System Technical Analysis & Architecture Report Applies to: MAM v4.8.2 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`. ```mermaid graph TD subgraph "Secure Tailnet / Corporate Network" Broker["Private NATS Broker (nats:2.12-alpine)
Native: 4222 | MQTT: 1883 | WS: 8080"] subgraph "Hermes (Delegator/Orchestrator)" SubClient["job_subscriber.py
(Role: subscriber)"] end subgraph "Herdr Workspace (Agent Host)" PubClient["publish_event.py
(Role: publisher)"] end SubClient -- "Subscribe (QoS 1)
Auth: mam_agent / mam_observer
ACL: Read python/mqtt/jobs/+/events" --> Broker PubClient -- "Publish (QoS 1 + Retain Terminal)
Auth: mam_agent
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: ```conf # 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: ```conf # 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`](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://:8080/mqtt` or `wss://: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`](nats-docker/PRIVATE_SERVER.md) and [`nats-docker/NATS_REPORT.md`](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: ```conf # 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: ```json { "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 (`/.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 `/.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 (`/.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** `/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:`) ``` orchestrator ── retained /dispatch (signed) ──▶ remote runner (other host) orchestrator ◀── /events + retained /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 [--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. ```mermaid 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/.json`. A `last_seq` counter is initialized to `0`. * **Locking**: Exclusive fcntl file lock acquired over `.lock` during write. * **Durable Audit Log**: Writes `//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 `. * **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 ` in the background **before** launching the agent. * **Broker Connection**: Connects to the resolved host, issues a QoS 1 subscription to `python/mqtt/jobs//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 (`.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/.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: ```python 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 ` 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 `/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 Python entry points print one `ERROR: : cannot read ()` line and exit 1; the wrapper prints that line followed by `ERROR: cannot load (see the line above)` and exits 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. ```mermaid 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
(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/.subscriber.out` (loop iterations use `.mam/jobs/.iter__.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: ```bash 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)](#24-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/.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.