413 lines
29 KiB
Markdown
413 lines
29 KiB
Markdown
# Messaging System Technical Analysis & Architecture Report
|
|
|
|
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) <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:
|
|
```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://<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`](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` 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**: Optional. Emitted at major check-points, long loops, or sub-task boundaries.
|
|
* **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.
|
|
|
|
---
|
|
|
|
## 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/<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
|
|
|
|
### 4.1 `registry.py` & `lib.sh` (Locking & Atomicity)
|
|
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)
|
|
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)
|
|
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:
|
|
```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 print logs or publish progress updates for the duration of the idle window, the subscriber aborts and exits with code 2.
|
|
|
|
---
|
|
|
|
### 4.4 `mqtt_common.py` (Logging, Env Vars & Config Resolution)
|
|
* **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, an error is logged and ambient search is refused (preventing silent fallback to unintended parent configs). Connection attempts fail-closed (`RuntimeError`).
|
|
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 <br> (lib.sh::atomic_dump_yaml)"]
|
|
```
|
|
|
|
### 5.1 Orchestration Wrappers (`multi-agent-mux-*`)
|
|
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`, and sleeps for `1` second.
|
|
* 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.
|
|
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. **Implement Signature-Based Payload Verification**:
|
|
Rather than sending a plaintext token, utilize HMAC signatures. The delegator and worker share a secret key; the worker publishes a signature of the payload (e.g. `HMAC-SHA256(secret_key, payload_bytes)`). The subscriber validates the signature, preventing token interception.
|
|
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.
|