message-bus

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) else {}

    if filepath in locks and locks[filepath]["worker_id"] == worker_id:
        del locks[filepath]
        with open(locks_file, "w") as f:
            json.dump(locks, f, indent=2)

        # Post unlock message and event
        post_message(bus_path, "FILE_UNLOCK", worker_id, {"filepath": filepath})

Worker Status Heartbeat

Workers post heartbeats every 5 minutes:

def update_worker_status(bus_path: str, worker_id: str, task_id: str, status: str, progress: int):
    status_file = f"{bus_path}/worker-status.json"
    statuses = json.load(open(status_file)) if os.path.exists(status_file) else {}

    statuses[worker_id] = {
        "task_id": task_id,
        "status": status,  # "RUNNING" | "COMPLETE" | "FAILED" | "BLOCKED"
        "progress_pct": progress,
        "last_heartbeat": datetime.utcnow().isoformat() + "Z"
    }

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

Board Deliberation Protocol

Phase 1: Assessment

Each director posts their assessment:

def post_board_assessment(bus_path: str, director: str, assessment: dict):
    board_path = f"{bus_path}/board"

    # read_file existing assessments
    assess_file = f"{board_path}/assessments.json"
    assessments = json.load(open(assess_file)) if os.path.exists(assess_file) else {}

    # Add this director's assessment
    assessments[director] = assessment

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

    # Post to main queue too
    post_message(bus_path, "BOARD_ASSESS", director, assessment)

Phase 2: Discussion

Directors respond to each other:

def post_board_discussion(bus_path: str, from_dir: str, to_dir: str,
                          msg_type: str, message: str, changes_verdict: bool):
    board_path = f"{bus_path}/board"

    discussion_msg = {
        "from": from_dir,
        "to": to_dir,
        "type": msg_type,  # "CHALLENGE" | "AGREE" | "QUESTION" | "CLARIFY"
        "message": message,
        "changes_my_verdict": changes_verdict,
        "timestamp": datetime.utcnow().isoformat() + "Z"
    }

    # Append to discussion log
    with open(f"{board_path}/discussion.jsonl", "a") as f:
        f.write_file(json.dumps(discussion_