# Grid Event Protocol

Version: 1.0
Status: Active
Last Updated: 2026-01-24

---

## Overview

This document defines the operational protocol for the Grid Event Bus. It covers how agents emit events, subscribe to topics, synchronize via barriers, and maintain the event stream for audit trails.

---

## Directory Structure

The Event Bus uses a file-based implementation with the following structure:

```
.grid/events/
  inbox/              # Pending events (agents write here)
    {timestamp}_{event_id}.json
  processed/          # Archived events (moved after processing)
    {date}/
      {timestamp}_{event_id}.json
  subscriptions/      # Agent subscription registrations
    {agent_id}.yaml
  barriers/           # Active barrier state
    {barrier_id}.yaml
  stream.log          # Append-only audit trail
```

### Directory Purposes

| Directory | Purpose | Retention |
|-----------|---------|-----------|
| `inbox/` | Pending events awaiting processing | Until processed |
| `processed/` | Archived events organized by date | 7 days default |
| `subscriptions/` | Agent subscription definitions | Until agent terminates |
| `barriers/` | Active synchronization barriers | Until released/timeout |
| `stream.log` | Append-only audit log | Session lifetime |

---

## Event Schema

### Base Event Structure

Every event MUST conform to this schema:

```yaml
event:
  # Required - Identity
  id: string              # UUID format: evt_{uuid_hex_12}
  type: string            # Topic path (e.g., "agent.spawned")
  source: string          # Emitting agent ID (e.g., "executor-001")
  timestamp: iso8601      # When event occurred (UTC)

  # Required - Routing
  target: string | null   # Specific agent, "broadcast", or null
  correlation_id: string  # Links related events (mission ID)
  causation_id: string    # ID of event that caused this one

  # Required - Version
  schema_version: "1.0"

  # Required - Payload
  payload:
    ref: string | null    # Blackboard path for large data
    # ... event-specific fields

  # Required - Metadata
  metadata:
    session_id: string
    phase: string | null
    block: string | null
    wave: number | null
```

### Event File Format

**Filename:** `{iso_timestamp}_{event_id}.json`

Example: `2026-01-24T163000.000Z_evt_550e8400.json`

**Content:** Full JSON event object

```json
{
  "id": "evt_550e8400e29b",
  "type": "agent.spawned",
  "source": "mc",
  "timestamp": "2026-01-24T16:30:00.000Z",
  "target": null,
  "correlation_id": "mission_abc123",
  "causation_id": null,
  "schema_version": "1.0",
  "payload": {
    "agent_id": "executor-001",
    "agent_type": "executor",
    "model": "opus",
    "purpose": "Execute block-01",
    "ref": null
  },
  "metadata": {
    "session_id": "sess_20260124_163000",
    "phase": "01-foundation",
    "block": "01",
    "wave": 1
  }
}
```

---

## Topic Hierarchy

### Naming Convention

Topics use hierarchical dot-notation:

```
{namespace}.{category}.{action}
```

### Core Namespaces

| Namespace | Purpose | Example Topics |
|-----------|---------|----------------|
| `system.*` | Grid-level lifecycle | `system.mission.start`, `system.barrier.release` |
| `phase.*` | Phase lifecycle | `phase.01-foundation.start`, `phase.01-foundation.complete` |
| `task.*` | Block/wave coordination | `task.block.assigned`, `task.wave.complete` |
| `agent.*` | Agent lifecycle/health | `agent.spawned`, `agent.heartbeat`, `agent.complete` |
| `verify.*` | Verification events | `verify.started`, `verify.complete`, `verify.gap_found` |
| `refine.*` | Refinement swarm | `refine.swarm.started`, `refine.visual.complete` |

### Topic Reference

#### System Events (`system.*`)

| Topic | Description | Key Payload Fields |
|-------|-------------|-------------------|
| `system.mission.start` | Mission begins | cluster, mode, session_id |
| `system.mission.complete` | Mission finished | cluster, duration_seconds, commits_made |
| `system.mission.failed` | Mission terminated | reason, last_position |
| `system.barrier.wait` | Barrier created | barrier_id, waiting_agents, required_count |
| `system.barrier.release` | Barrier released | barrier_id, waiting_duration_ms |
| `system.checkpoint.created` | Checkpoint written | checkpoint_path, reason, position |
| `system.checkpoint.resumed` | Resumed from checkpoint | checkpoint_path |
| `system.context.warning` | Context budget alert | usage_percent, threshold, agent_id |

#### Phase Events (`phase.*`)

| Topic | Description | Key Payload Fields |
|-------|-------------|-------------------|
| `phase.{id}.start` | Phase begins | phase_id, phase_name, total_blocks |
| `phase.{id}.complete` | Phase finished | phase_id, blocks_completed, duration_seconds |
| `phase.{id}.failed` | Phase failed | phase_id, failed_block, reason |

#### Task Events (`task.*`)

| Topic | Description | Key Payload Fields |
|-------|-------------|-------------------|
| `task.wave.start` | Wave begins | wave_number, blocks_in_wave, parallel_count |
| `task.wave.complete` | Wave finished | wave_number, blocks_completed |
| `task.block.assigned` | Block assigned | block_id, plan_path, assigned_to |
| `task.block.started` | Block execution started | block_id, executor_id, threads_total |
| `task.block.progress` | Progress update | block_id, thread_current, percent_complete |
| `task.block.complete` | Block finished | block_id, commits, summary_path |
| `task.block.failed` | Block failed | block_id, thread_failed, reason |
| `task.block.checkpoint` | Checkpoint hit | block_id, checkpoint_type, checkpoint_path |

#### Agent Events (`agent.*`)

| Topic | Description | Key Payload Fields |
|-------|-------------|-------------------|
| `agent.spawned` | Agent created | agent_id, agent_type, model, purpose |
| `agent.heartbeat` | Health check | agent_id, status, context_usage_percent |
| `agent.complete` | Agent finished | agent_id, result_status, result_path |
| `agent.error` | Agent error | agent_id, error_type, error_message, recoverable |
| `agent.stale` | Heartbeat timeout | agent_id, last_heartbeat_at, stale_duration_minutes |

#### Verification Events (`verify.*`)

| Topic | Description | Key Payload Fields |
|-------|-------------|-------------------|
| `verify.started` | Verification begins | verifier_id, target_block, plan_path |
| `verify.complete` | Verification finished | status, confidence_score, recommendation |
| `verify.gap_found` | Gap detected | gap_type, gap_item, severity, fix_required |

---

## Emitting Events

### How Agents Emit Events

Agents emit events by writing JSON files to the `inbox/` directory.

**Step 1: Generate Event ID**
```python
import uuid
event_id = f"evt_{uuid.uuid4().hex[:12]}"
```

**Step 2: Create Event Object**
```python
from datetime import datetime, timezone

event = {
    "id": event_id,
    "type": "agent.heartbeat",
    "source": agent_id,
    "timestamp": datetime.now(timezone.utc).isoformat(),
    "target": None,
    "correlation_id": context.correlation_id,
    "causation_id": context.last_event_id,
    "schema_version": "1.0",
    "payload": {
        "agent_id": agent_id,
        "status": "active",
        "current_action": "Writing authentication handler",
        "context_usage_percent": 45.2,
        "heartbeat_at": datetime.now(timezone.utc).isoformat(),
        "ref": None
    },
    "metadata": {
        "session_id": context.session_id,
        "phase": context.phase,
        "block": context.block,
        "wave": context.wave
    }
}
```

**Step 3: Write to Inbox**
```python
import json
import os

# Format timestamp for filename (remove colons for filesystem compatibility)
ts = event["timestamp"].replace(":", "")
filename = f"{ts}_{event_id}.json"
inbox_path = f".grid/events/inbox/{filename}"

# Ensure directory exists
os.makedirs(".grid/events/inbox", exist_ok=True)

# Write event
with open(inbox_path, "w") as f:
    json.dump(event, f, indent=2)
```

**Step 4: Append to Stream Log**
```python
stream_line = "|".join([
    event["timestamp"],
    event["id"],
    event["type"],
    event["source"],
    event["target"] or "broadcast",
    json.dumps(event["payload"], separators=(',', ':'))
])

with open(".grid/events/stream.log", "a") as f:
    f.write(stream_line + "\n")
```

### Emit Helper Functions

```python
def emit_event(event_type: str, payload: dict, target: str = None) -> str:
    """
    Emit an event to the Event Bus.
    Returns the event ID.
    """
    event_id = f"evt_{uuid.uuid4().hex[:12]}"
    timestamp = datetime.now(timezone.utc).isoformat()

    event = {
        "id": event_id,
        "type": event_type,
        "source": get_current_agent_id(),
        "timestamp": timestamp,
        "target": target,
        "correlation_id": get_correlation_id(),
        "causation_id": get_last_event_id(),
        "schema_version": "1.0",
        "payload": {**payload, "ref": payload.get("ref")},
        "metadata": get_current_metadata()
    }

    # Write to inbox
    ts = timestamp.replace(":", "")
    filename = f"{ts}_{event_id}.json"
    os.makedirs(".grid/events/inbox", exist_ok=True)
    with open(f".grid/events/inbox/{filename}", "w") as f:
        json.dump(event, f, indent=2)

    # Append to stream
    with open(".grid/events/stream.log", "a") as f:
        f.write(f"{timestamp}|{event_id}|{event_type}|{event['source']}|{target or 'broadcast'}|{json.dumps(payload)}\n")

    return event_id
```

---

## Subscribing to Events

### How Agents Subscribe

Agents register subscriptions by writing YAML files to `subscriptions/`.

**File:** `.grid/events/subscriptions/{agent_id}.yaml`

```yaml
# .grid/events/subscriptions/executor-001.yaml
agent_id: executor-001
registered_at: "2026-01-24T16:30:00.000Z"
subscriptions:
  # Exact topic match
  - topic: "task.block.assigned"
    filter:
      block_id: "01"

  # Wildcard - all system events
  - topic: "system.*"

  # Category wildcard
  - topic: "verify.*.complete"

  # All events for this mission
  - topic: "*"
    filter:
      correlation_id: "mission_abc123"
```

### Wildcard Patterns

| Pattern | Matches |
|---------|---------|
| `system.*` | All events starting with `system.` |
| `*.complete` | All events ending with `.complete` |
| `task.block.*` | All `task.block.` events |
| `agent.{id}.*` | Events for specific agent |
| `*` | All events (use with filter) |

### Pattern Matching Algorithm

```python
def topic_matches(event_type: str, pattern: str) -> bool:
    """Check if event type matches subscription pattern."""
    if pattern == "*":
        return True

    if pattern.endswith(".*"):
        prefix = pattern[:-2]
        return event_type.startswith(prefix + ".")

    if pattern.startswith("*."):
        suffix = pattern[2:]
        return event_type.endswith("." + suffix)

    pattern_parts = pattern.split(".")
    event_parts = event_type.split(".")

    if len(pattern_parts) != len(event_parts):
        return False

    for p, e in zip(pattern_parts, event_parts):
        if p == "*":
            continue
        if p != e:
            return False

    return True
```

### Polling for Events

```python
def poll_events(agent_id: str) -> list:
    """Poll inbox for events matching agent's subscriptions."""

    # Load subscriptions
    sub_path = f".grid/events/subscriptions/{agent_id}.yaml"
    if not os.path.exists(sub_path):
        return []

    with open(sub_path) as f:
        subs = yaml.safe_load(f)

    # Scan inbox
    inbox_path = ".grid/events/inbox/"
    events = []

    for filename in sorted(os.listdir(inbox_path)):
        with open(os.path.join(inbox_path, filename)) as f:
            event = json.load(f)

        # Check if matches any subscription
        for sub in subs.get("subscriptions", []):
            if topic_matches(event["type"], sub["topic"]):
                # Check filter if present
                if matches_filter(event, sub.get("filter", {})):
                    events.append(event)
                    break

    return events


def matches_filter(event: dict, filter_spec: dict) -> bool:
    """Check if event matches filter specification."""
    if not filter_spec:
        return True

    for key, value in filter_spec.items():
        # Check in payload
        if key in event.get("payload", {}):
            if event["payload"][key] != value:
                return False
        # Check in metadata
        elif key in event.get("metadata", {}):
            if event["metadata"][key] != value:
                return False
        # Check in root
        elif key in event:
            if event[key] != value:
                return False
        else:
            return False

    return True
```

---

## Barrier Synchronization Protocol

### What Are Barriers?

Barriers are synchronization points where multiple agents must reach before any can proceed. Used for:
- Wave completion (all executors in wave must finish)
- Phase transitions
- Checkpoint resolution
- Custom coordination

### Barrier Types

| Type | Condition | Use Case |
|------|-----------|----------|
| `wave` | All executors complete | Wave-level parallelism |
| `phase` | All blocks complete | Phase transitions |
| `checkpoint` | User response received | Human-in-the-loop |
| `verification` | Recognizer returns CLEAR | Quality gate |
| `n_of_m` | N of M agents signal | Flexible coordination |

### Creating a Barrier

**File:** `.grid/events/barriers/{barrier_id}.yaml`

```yaml
# .grid/events/barriers/wave_1_phase_01.yaml
barrier_id: "wave_1_phase_01"
barrier_type: "wave"
created_at: "2026-01-24T16:30:00.000Z"
status: "waiting"           # waiting | released | timeout | cancelled

condition:
  type: "all_complete"      # all_complete | any_complete | n_of_m
  required_agents:
    - "executor-001"
    - "executor-002"
    - "executor-003"
  required_count: 3
  timeout_seconds: 600

state:
  signaled_agents:
    - agent_id: "executor-001"
      signaled_at: "2026-01-24T16:40:00.000Z"
      result: "SUCCESS"
  waiting_agents:
    - "executor-002"
    - "executor-003"

release_event_id: null
```

### Signaling a Barrier

When an agent completes its work, it signals the barrier:

```python
def signal_barrier(barrier_id: str, agent_id: str, result: str) -> bool:
    """
    Signal barrier completion.
    Returns True if barrier was released.
    """
    barrier_path = f".grid/events/barriers/{barrier_id}.yaml"

    with open(barrier_path) as f:
        barrier = yaml.safe_load(f)

    if barrier["status"] != "waiting":
        return False

    # Record signal
    barrier["state"]["signaled_agents"].append({
        "agent_id": agent_id,
        "signaled_at": datetime.now(timezone.utc).isoformat(),
        "result": result
    })

    # Remove from waiting
    if agent_id in barrier["state"]["waiting_agents"]:
        barrier["state"]["waiting_agents"].remove(agent_id)

    # Check if barrier should release
    signaled_count = len(barrier["state"]["signaled_agents"])
    required = barrier["condition"]["required_count"]

    released = False
    if barrier["condition"]["type"] == "all_complete":
        released = signaled_count >= required
    elif barrier["condition"]["type"] == "any_complete":
        released = signaled_count >= 1
    elif barrier["condition"]["type"] == "n_of_m":
        released = signaled_count >= required

    if released:
        barrier["status"] = "released"

        # Emit release event
        release_event_id = emit_event("system.barrier.release", {
            "barrier_id": barrier_id,
            "released_at": datetime.now(timezone.utc).isoformat(),
            "waiting_duration_ms": calculate_duration(barrier)
        })
        barrier["release_event_id"] = release_event_id

    # Save barrier state
    with open(barrier_path, "w") as f:
        yaml.dump(barrier, f)

    return released
```

### Waiting on a Barrier

```python
def wait_for_barrier(barrier_id: str, timeout_seconds: int = 600) -> str:
    """
    Wait for barrier to be released.
    Returns: "released" | "timeout" | "cancelled"
    """
    start_time = time.time()
    barrier_path = f".grid/events/barriers/{barrier_id}.yaml"

    while True:
        if time.time() - start_time > timeout_seconds:
            return "timeout"

        with open(barrier_path) as f:
            barrier = yaml.safe_load(f)

        if barrier["status"] == "released":
            return "released"
        elif barrier["status"] in ("timeout", "cancelled"):
            return barrier["status"]

        # Poll interval
        time.sleep(1)
```

---

## Stream Log Format

The `stream.log` file provides an append-only audit trail.

### Format

```
{timestamp}|{event_id}|{type}|{source}|{target}|{payload_json}
```

### Example Entries

```
2026-01-24T16:30:00.000Z|evt_550e8400|agent.spawned|mc|broadcast|{"agent_id":"executor-001","model":"opus"}
2026-01-24T16:30:01.000Z|evt_550e8401|task.block.assigned|mc|executor-001|{"block_id":"01","plan_path":".grid/phases/01/01-PLAN.md"}
2026-01-24T16:35:00.000Z|evt_550e8402|agent.heartbeat|executor-001|mc|{"status":"active","progress":25}
2026-01-24T16:45:00.000Z|evt_660e8400|task.block.complete|executor-001|mc|{"block_id":"01","commits":["abc123"]}
```

### Reading the Stream

```python
def read_stream(since: str = None) -> list:
    """Read events from stream log, optionally since a timestamp."""
    events = []

    with open(".grid/events/stream.log") as f:
        for line in f:
            parts = line.strip().split("|", 5)
            if len(parts) != 6:
                continue

            timestamp, event_id, event_type, source, target, payload = parts

            if since and timestamp <= since:
                continue

            events.append({
                "timestamp": timestamp,
                "id": event_id,
                "type": event_type,
                "source": source,
                "target": target if target != "broadcast" else None,
                "payload": json.loads(payload)
            })

    return events
```

---

## Common Event Examples

### Agent Spawned

```json
{
  "id": "evt_550e8400e29b",
  "type": "agent.spawned",
  "source": "mc",
  "timestamp": "2026-01-24T16:30:00.000Z",
  "target": null,
  "correlation_id": "mission_abc123",
  "causation_id": null,
  "schema_version": "1.0",
  "payload": {
    "agent_id": "executor-001",
    "agent_type": "executor",
    "spawned_by": "mc",
    "model": "opus",
    "purpose": "Execute block-01 foundation setup",
    "spawned_at": "2026-01-24T16:30:00.000Z",
    "ref": null
  },
  "metadata": {
    "session_id": "sess_20260124_163000_xyz",
    "phase": "01-foundation",
    "block": "01",
    "wave": 1
  }
}
```

### Agent Heartbeat

```json
{
  "id": "evt_550e8402a1b2",
  "type": "agent.heartbeat",
  "source": "executor-001",
  "timestamp": "2026-01-24T16:35:00.000Z",
  "target": "mc",
  "correlation_id": "mission_abc123",
  "causation_id": "evt_550e8400e29b",
  "schema_version": "1.0",
  "payload": {
    "agent_id": "executor-001",
    "status": "active",
    "context_usage_percent": 45.2,
    "current_action": "Writing POST handler for /api/auth",
    "heartbeat_at": "2026-01-24T16:35:00.000Z",
    "ref": null
  },
  "metadata": {
    "session_id": "sess_20260124_163000_xyz",
    "phase": "01-foundation",
    "block": "01",
    "wave": 1
  }
}
```

### Agent Complete

```json
{
  "id": "evt_660e8400c3d4",
  "type": "agent.complete",
  "source": "executor-001",
  "timestamp": "2026-01-24T16:45:00.000Z",
  "target": "mc",
  "correlation_id": "mission_abc123",
  "causation_id": "evt_550e8402a1b2",
  "schema_version": "1.0",
  "payload": {
    "agent_id": "executor-001",
    "completed_at": "2026-01-24T16:45:00.000Z",
    "result_status": "SUCCESS",
    "result_path": ".grid/phases/01-foundation/01-SUMMARY.md",
    "duration_seconds": 900,
    "tokens_used": 45000,
    "ref": ".grid/phases/01-foundation/01-SUMMARY.md"
  },
  "metadata": {
    "session_id": "sess_20260124_163000_xyz",
    "phase": "01-foundation",
    "block": "01",
    "wave": 1
  }
}
```

### Agent Error

```json
{
  "id": "evt_770e8500d4e5",
  "type": "agent.error",
  "source": "executor-002",
  "timestamp": "2026-01-24T16:50:00.000Z",
  "target": "mc",
  "correlation_id": "mission_abc123",
  "causation_id": "evt_550e8401b2c3",
  "schema_version": "1.0",
  "payload": {
    "agent_id": "executor-002",
    "error_at": "2026-01-24T16:50:00.000Z",
    "error_type": "api_error",
    "error_message": "Rate limit exceeded when calling external API",
    "recoverable": true,
    "ref": null
  },
  "metadata": {
    "session_id": "sess_20260124_163000_xyz",
    "phase": "01-foundation",
    "block": "02",
    "wave": 1
  }
}
```

### Block Complete

```json
{
  "id": "evt_660e8400c3d4",
  "type": "task.block.complete",
  "source": "executor-001",
  "timestamp": "2026-01-24T16:45:00.000Z",
  "target": "mc",
  "correlation_id": "mission_abc123",
  "causation_id": "evt_550e8402a1b2",
  "schema_version": "1.0",
  "payload": {
    "block_id": "01",
    "executor_id": "executor-001",
    "completed_at": "2026-01-24T16:45:00.000Z",
    "threads_completed": 3,
    "commits": ["abc123", "def456", "ghi789"],
    "summary_path": ".grid/phases/01-foundation/01-SUMMARY.md",
    "ref": ".grid/phases/01-foundation/01-SUMMARY.md"
  },
  "metadata": {
    "session_id": "sess_20260124_163000_xyz",
    "phase": "01-foundation",
    "block": "01",
    "wave": 1
  }
}
```

---

## Initialization

### Setting Up Event Bus

When Grid starts, initialize the event bus structure:

```bash
# Create directory structure
mkdir -p .grid/events/inbox
mkdir -p .grid/events/processed
mkdir -p .grid/events/subscriptions
mkdir -p .grid/events/barriers

# Initialize stream log
touch .grid/events/stream.log
```

### Python Initialization

```python
import os

def init_event_bus():
    """Initialize event bus directory structure."""
    dirs = [
        ".grid/events/inbox",
        ".grid/events/processed",
        ".grid/events/subscriptions",
        ".grid/events/barriers"
    ]

    for d in dirs:
        os.makedirs(d, exist_ok=True)

    # Create stream log if not exists
    stream_path = ".grid/events/stream.log"
    if not os.path.exists(stream_path):
        open(stream_path, "a").close()

    return True
```

---

## Cleanup and Archival

### Processing Events

After MC processes an event, move it to `processed/`:

```python
def archive_event(event_id: str, event_timestamp: str):
    """Move processed event to archive."""
    date = event_timestamp[:10]  # YYYY-MM-DD

    # Find event file
    inbox_path = ".grid/events/inbox/"
    for filename in os.listdir(inbox_path):
        if event_id in filename:
            source = os.path.join(inbox_path, filename)
            dest_dir = f".grid/events/processed/{date}/"
            os.makedirs(dest_dir, exist_ok=True)
            dest = os.path.join(dest_dir, filename)
            os.rename(source, dest)
            break
```

### Retention Policy

Default retention: 7 days

```python
def cleanup_old_events(retention_days: int = 7):
    """Remove events older than retention period."""
    cutoff = datetime.now() - timedelta(days=retention_days)
    cutoff_str = cutoff.strftime("%Y-%m-%d")

    processed_path = ".grid/events/processed/"
    for date_dir in os.listdir(processed_path):
        if date_dir < cutoff_str:
            shutil.rmtree(os.path.join(processed_path, date_dir))
```

---

## Configuration

### Event Bus Config

In `.grid/config.json`:

```json
{
  "event_bus": {
    "enabled": true,
    "implementation": "file",
    "inbox_poll_interval_ms": 500,
    "event_retention_days": 7,
    "max_inbox_size": 1000,
    "barrier_timeout_seconds": 600,
    "heartbeat_stale_threshold_minutes": 10,
    "stream_log_enabled": true,
    "archive_enabled": true
  }
}
```

---

## Summary

The Event Protocol enables:

1. **Decoupled Communication** - Agents emit/receive via files
2. **Topic-Based Routing** - Flexible subscription patterns
3. **Barrier Synchronization** - Wave/phase coordination
4. **Audit Trail** - Complete event history in stream.log
5. **Reference-Based Payloads** - Large data stays in Blackboard

All agents should emit events for visibility. MC processes events and coordinates responses.

---

*End of Event Protocol. End of Line.*