/message-bus

name: message-bus
description: "File-based message queue for inter-agent coordination. Used by workers AND board directors to communicate. Provides: progress updates, task completion signals, file locking, board deliberation. Core infrastructure for parallel execution."

Message Bus -- Inter-Agent Communication Protocol

File-based message queue enabling workers and board directors to coordinate via shared state.

Directory Structure

conductor/tracks/{track}/.message-bus/
├── queue.jsonl           # Append-only message log (all messages)
├── .lock_mutex           # OS-level mutex file for atomic lock operations (fcntl)
├── locks.json            # Current file locks
├── worker-status.json    # Worker heartbeats and states
├── events/               # Signal files for polling
│   ├── TASK_COMPLETE_1.1.event
│   └── FILE_UNLOCK_*.event
└── board/                # Board deliberation sessions
    ├── session-{ts}.json # Session metadata
    ├── assessments.json  # Director assessments (Phase 1)
    ├── discussion.jsonl  # Discussion messages (Phase 2)
    └── votes.json        # Final votes (Phase 3)

Message Types

Worker Messages

| Type | Purpose | Payload | |------|---------|---------| | `PROGRESS` | Task progress update | `{ task_id, progress_pct, current_subtask }` | | `TASK_COMPLETE` | Task finished | `{ task_id, commit_sha, files_modified, unblocks[] }` | | `TASK_FAILED` | Task failed | `{ task_id, error, stack_trace }` | | `FILE_LOCK` | Acquire file lock | `{ filepath, lock_type, expires_at }` | | `FILE_UNLOCK` | Release file lock | `{ filepath }` | | `BLOCKED` | Waiting on dependency | `{ task_id, waiting_for, resource }` |

Board Messages

| Type | Purpose | Payload | |------|---------|---------| | `BOARD_ASSESS` | Director assessment | `{ director, verdict, score, concerns[], recommendations[] }` | | `BOARD_DISCUSS` | Discussion message | `{ from, to, type, message, changes_my_verdict }` | | `BOARD_VOTE` | Final vote | `{ director, final_verdict, confidence, conditions[] }` | | `BOARD_RESOLVE` | Aggregated decision | `{ verdict, vote_summary, conditions[], dissent[] }` |

Message Format

All messages follow this structure:

{
  "id": "msg-{uuid}",
  "type": "PROGRESS | TASK_COMPLETE | BOARD_ASSESS | ...",
  "source": "worker-1.1-xxx | CA | orchestrator",
  "timestamp": "2026-02-01T12:00:00Z",
  "payload": { ... }
}

Worker Protocol

Posting Messages

def post_message(bus_path: str, msg_type: str, source: str, payload: dict):
    message = {
        "id": f"msg-{uuid4()}",
        "type": msg_type,
        "source": source,
        "timestamp": datetime.utcnow().isoformat() + "Z",
        "payload": payload
    }

    # Append to queue (atomic via file locking)
    with open(f"{bus_path}/queue.jsonl", "a") as f:
        f.write_file(json.dumps(message) + "\n")

    # Create event file for polling
    if msg_type in ["TASK_COMPLETE", "FILE_UNLOCK", "BOARD_RESOLVE"]:
        event_file = f"{bus_path}/events/{msg_type}_{payload.get('task_id', 'all')}.event"
        Path(event_file).touch()

Reading Messages

def read_messages(bus_path: str, since: str = None, msg_type: str = None) -> list:
    messages = []
    with open(f"{bus_path}/queue.jsonl", "r") as f:
        for line in f:
            msg = json.loads(line)
            if since and msg["timestamp"] < since:
                continue
            if msg_type and msg["type"] != msg_type:
                continue
            messages.append(msg)
    return messages

Polling for Events

def wait_for_event(bus_path: str, event_pattern: str, timeout: int = 300) -> bool:
    """Wait for event file to appear. Returns True if found, False if timeout."""
    import glob
    import time

    start = time.time()
    while time.time() - start < timeout:
        matches = glob.glob(f"{bus_path}/events/{event_pattern}")
        if matches:
            return True
        time.sleep(1)
    return False

File Lock Protocol

Acquiring Locks

import fcntl

def acquire_lock(bus_path: str, filepath: str, worker_id: str) -> bool:
    locks_file = f"{bus_path}/locks.json"
    mutex_file = f"{bus_path}/.lock_mutex"

    # Use an OS-level exclusive lock on a dedicated mutex file so that
    # the read → check → write sequence is atomic across concurrent processes.
    # Open in append mode — we only need the file to exist as a lock target,
    # not to store any content. Append mode avoids truncation overhead.
    lock_fd = open(mutex_file, "a")
    try:
        fcntl.flock(lock_fd, fcntl.LOCK_EX | fcntl.LOCK_NB)
    except BlockingIOError:
        lock_fd.close()
        return False  # Another process is mid-lock; retry later

    try:
        if os.path.exists(locks_file):
            with open(locks_file) as f:
                locks = json.load(f)
        else:
            locks = {}

        existing = locks.get(filepath)
        if existing and existing["worker_id"] != worker_id:
            # Check if the lock has expired (30-min timeout)
            if datetime.fromisoformat(existing["expires_at"]) > datetime.utcnow():
                return False  # Legitimately locked by another worker

        # Acquire lock
        locks[filepath] = {
            "worker_id": worker_id,
            "acquired_at": datetime.utcnow().isoformat() + "Z",
            "expires_at": (datetime.utcnow() + timedelta(minutes=30)).isoformat() + "Z"
        }

        with open(locks_file, "w") as f:
            json.dump(locks, f, indent=2)

        # Post lock message
        post_message(bus_path, "FILE_LOCK", worker_id, {"filepath": filepath})
        return True
    finally:
        fcntl.flock(lock_fd, fcntl.LOCK_UN)
        lock_fd.close()

Releasing Locks

def release_lock(bus_path: str, filepath: str, worker_id: str):
    locks_file = f"{bus_path}/locks.json"
    locks = json.load(open(locks_file)) if os.path.exists(locks_file