# Grid Event Bus Specification Version: 1.0 Status: Draft Author: Grid Architecture Team --- ## Overview The Event Bus is the central nervous system of The Grid's multi-agent architecture. It enables decoupled communication between Master Control and Programs (subagents), providing pub/sub messaging, synchronization barriers, and event sourcing for audit trails. ### Design Principles 1. **Decoupled Communication** - Programs don't need to know about each other 2. **File-Based First** - Simple implementation that works immediately 3. **Reference-Based Payloads** - Events carry references to Blackboard, not data copies 4. **Ordered Delivery** - Events within a topic maintain FIFO ordering 5. **Barrier Support** - First-class synchronization primitives for wave execution 6. **Audit Trail** - All events persisted for debugging and replay --- ## Topic Hierarchy ### Topic Structure Topics use a hierarchical dot-notation with wildcards for flexible subscription. ``` {namespace}.{category}.{action} Examples: system.mission.start phase.01-foundation.start task.block-01.assigned agent.executor-001.spawned ``` ### Core Topics #### System Events (`system.*`) Grid-level lifecycle and control events. ```yaml system.*: system.mission.start: description: "Mission begins" payload: cluster: string # Mission name mode: string # AUTOPILOT | GUIDED | HANDS_ON session_id: string # Unique session identifier started_at: iso8601 system.mission.complete: description: "Mission finished successfully" payload: cluster: string session_id: string completed_at: iso8601 duration_seconds: number blocks_completed: number commits_made: number system.mission.failed: description: "Mission terminated with failure" payload: cluster: string session_id: string failed_at: iso8601 reason: string last_position: phase: number block: number wave: number thread: number system.barrier.wait: description: "Barrier synchronization requested" payload: barrier_id: string # Unique barrier identifier barrier_type: string # wave | phase | checkpoint waiting_agents: [string] required_count: number system.barrier.release: description: "Barrier condition met, proceed" payload: barrier_id: string released_at: iso8601 waiting_duration_ms: number system.checkpoint.created: description: "Checkpoint file written" payload: checkpoint_path: string reason: string # human_verify | decision | failure | session_death position: object system.checkpoint.resumed: description: "Resumed from checkpoint" payload: checkpoint_path: string resumed_at: iso8601 system.context.warning: description: "Context budget warning" payload: usage_percent: number threshold: string # WARNING | COMPRESS | EMERGENCY | OVERFLOW agent_id: string ``` #### Phase Events (`phase.*`) Phase lifecycle events. ```yaml phase.*: phase.{phase_id}.start: description: "Phase execution begins" payload: phase_id: string # e.g., "01-foundation" phase_name: string phase_number: number total_phases: number total_blocks: number total_waves: number phase.{phase_id}.complete: description: "Phase finished successfully" payload: phase_id: string completed_at: iso8601 blocks_completed: number duration_seconds: number phase.{phase_id}.failed: description: "Phase failed" payload: phase_id: string failed_at: iso8601 failed_block: string reason: string ``` #### Task Events (`task.*`) Block and wave-level task coordination. ```yaml task.*: task.wave.start: description: "Wave execution begins" payload: wave_number: number phase_id: string blocks_in_wave: [string] parallel_count: number task.wave.complete: description: "Wave finished" payload: wave_number: number phase_id: string blocks_completed: [string] duration_seconds: number task.block.assigned: description: "Block assigned to executor" payload: block_id: string plan_path: string # Reference to Blackboard assigned_to: string # Agent ID wave: number task.block.started: description: "Block execution started" payload: block_id: string executor_id: string started_at: iso8601 threads_total: number task.block.progress: description: "Block execution progress update" payload: block_id: string executor_id: string thread_current: number thread_total: number percent_complete: number last_action: string task.block.complete: description: "Block finished successfully" payload: block_id: string executor_id: string completed_at: iso8601 threads_completed: number commits: [string] summary_path: string # Reference to Blackboard task.block.failed: description: "Block execution failed" payload: block_id: string executor_id: string failed_at: iso8601 thread_failed: number reason: string failure_report: string # Reference to Blackboard task.block.checkpoint: description: "Block hit checkpoint" payload: block_id: string executor_id: string checkpoint_type: string # human_verify | decision | human_action checkpoint_path: string thread_blocked: number ``` #### Agent Events (`agent.*`) Individual agent lifecycle and health. ```yaml agent.*: agent.spawned: description: "New agent spawned" payload: agent_id: string # e.g., "executor-001" agent_type: string # executor | planner | recognizer | etc. spawned_by: string # Parent agent ID (usually "mc") model: string # opus | sonnet | haiku purpose: string # Brief description spawned_at: iso8601 agent.heartbeat: description: "Agent health check" payload: agent_id: string status: string # active | idle | blocked context_usage_percent: number current_action: string heartbeat_at: iso8601 agent.complete: description: "Agent finished successfully" payload: agent_id: string completed_at: iso8601 result_status: string # SUCCESS | CHECKPOINT | VERIFIED | CLEAR result_path: string # Reference to output duration_seconds: number tokens_used: number agent.error: description: "Agent encountered error" payload: agent_id: string error_at: iso8601 error_type: string # timeout | crash | context_overflow | api_error error_message: string recoverable: boolean agent.stale: description: "Agent heartbeat timeout (staleness detected)" payload: agent_id: string last_heartbeat_at: iso8601 stale_duration_minutes: number last_known_action: string ``` #### Verification Events (`verify.*`) Recognizer verification events. ```yaml verify.*: verify.started: description: "Verification patrol begins" payload: verifier_id: string target_block: string plan_path: string started_at: iso8601 verify.complete: description: "Verification finished" payload: verifier_id: string target_block: string status: string # CLEAR | GAPS_FOUND | PARTIAL | CRITICAL_ANOMALY confidence_score: number recommendation: string # auto_approve | human_verify report_path: string verify.gap_found: description: "Gap detected during verification" payload: verifier_id: string target_block: string gap_type: string # truth_failed | artifact_failed | key_link_not_wired gap_item: string severity: string # blocker | warning fix_required: string ``` #### Refinement Events (`refine.*`) Refinement swarm events. ```yaml refine.*: refine.swarm.started: description: "Refinement swarm begins" payload: swarm_id: string agents_spawned: [string] project_context: object refine.visual.complete: description: "Visual inspection finished" payload: inspector_id: string screenshots_taken: number issues_found: number report_path: string refine.e2e.complete: description: "E2E testing finished" payload: exerciser_id: string tests_run: number tests_passed: number tests_failed: number report_path: string refine.persona.complete: description: "Persona simulation finished" payload: persona_id: string persona_name: string would_return: boolean would_recommend: boolean report_path: string refine.swarm.complete: description: "Refinement swarm finished" payload: swarm_id: string refinement_plan_path: string total_issues: number p0_issues: number ``` --- ## Event Schema ### Base Event Structure Every event conforms to this schema: ```yaml event: # Required fields id: string # UUID v4, unique event identifier type: string # Topic path (e.g., "agent.spawned") source: string # Agent ID that emitted event (e.g., "mc", "executor-001") timestamp: iso8601 # When event occurred # Routing target: string | null # Specific agent ID, "broadcast", or null for topic subscribers correlation_id: string # Links related events (e.g., same mission) causation_id: string # ID of event that caused this one (chain of causality) # Versioning schema_version: "1.0" # Event schema version # Payload payload: # Event-specific data ref: string | null # Blackboard path for large data (reference, not copy) # ... additional fields per event type # Metadata metadata: session_id: string # Current Grid session phase: string | null # Current phase if applicable block: string | null # Current block if applicable wave: number | null # Current wave if applicable ``` ### Example Events **Agent Spawned:** ```json { "id": "evt_550e8400-e29b-41d4-a716-446655440000", "type": "agent.spawned", "source": "mc", "timestamp": "2026-01-24T16:30:00.000Z", "target": null, "correlation_id": "mission_abc123", "causation_id": "evt_previous_event_id", "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 } } ``` **Block Complete:** ```json { "id": "evt_660e8400-e29b-41d4-a716-446655440001", "type": "task.block.complete", "source": "executor-001", "timestamp": "2026-01-24T16:45:00.000Z", "target": "mc", "correlation_id": "mission_abc123", "causation_id": "evt_550e8400-e29b-41d4-a716-446655440000", "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"], "ref": ".grid/phases/01-foundation/01-SUMMARY.md" }, "metadata": { "session_id": "sess_20260124_163000_xyz", "phase": "01-foundation", "block": "01", "wave": 1 } } ``` --- ## File-Based Implementation ### Directory Structure ``` .grid/ ├── events/ │ ├── inbox/ # Pending events to process │ │ ├── {timestamp}_{event_id}.json │ │ └── ... │ ├── processed/ # Processed events (archive) │ │ ├── {date}/ │ │ │ ├── {timestamp}_{event_id}.json │ │ │ └── ... │ │ └── ... │ ├── subscriptions/ # Subscription registrations │ │ ├── {agent_id}.yaml │ │ └── ... │ ├── barriers/ # Active barrier state │ │ ├── {barrier_id}.yaml │ │ └── ... │ └── stream.log # Append-only event log (audit trail) ``` ### Event File Format Individual event files in `inbox/` and `processed/`: **Filename:** `{iso_timestamp}_{event_id}.json` - Timestamp enables ordering - Event ID ensures uniqueness **Example:** `2026-01-24T163000.000Z_evt_550e8400.json` ### Event Stream Log Append-only log for audit trail in `events/stream.log`: ``` 2026-01-24T16:30:00.000Z|evt_550e8400|agent.spawned|mc|executor-001|{"agent_id":"executor-001",...} 2026-01-24T16:30:01.000Z|evt_550e8401|task.block.assigned|mc|executor-001|{"block_id":"01",...} 2026-01-24T16:45:00.000Z|evt_660e8400|task.block.complete|executor-001|mc|{"block_id":"01",...} ``` Format: `{timestamp}|{event_id}|{type}|{source}|{target}|{payload_json}` --- ## Subscription Model ### Subscription Registration Agents register subscriptions on spawn: **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" # Only events for this block # Wildcard subscription - topic: "system.*" # All system events # Category wildcard - topic: "verify.*.complete" # All events (monitoring) - topic: "*" filter: correlation_id: "mission_abc123" # Only this mission ``` ### Wildcard Patterns | Pattern | Matches | |---------|---------| | `system.*` | `system.mission.start`, `system.barrier.wait`, etc. | | `*.complete` | `task.block.complete`, `agent.complete`, etc. | | `task.block.*` | `task.block.assigned`, `task.block.complete`, etc. | | `agent.{id}.*` | Events for specific agent | | `*` | All events (use with filter) | ### Event Delivery **Polling Model (File-Based):** ```python def poll_events(agent_id: str) -> list[Event]: """Poll for events matching agent's subscriptions.""" # Load subscriptions subs = load_subscriptions(agent_id) # Scan inbox for matching events inbox_path = ".grid/events/inbox/" events = [] for event_file in sorted(os.listdir(inbox_path)): event = load_event(inbox_path + event_file) if matches_subscriptions(event, subs): events.append(event) # Move to agent's processing queue or mark as delivered mark_delivered(event, agent_id) return events def matches_subscriptions(event: Event, subs: list) -> bool: """Check if event matches any subscription.""" for sub in subs: if topic_matches(event.type, sub.topic): if filter_matches(event, sub.get("filter", {})): return True return False def topic_matches(event_type: str, pattern: str) -> bool: """Match event type against subscription pattern with wildcards.""" if pattern == "*": return True pattern_parts = pattern.split(".") event_parts = event_type.split(".") for p, e in zip(pattern_parts, event_parts): if p == "*": continue if p != e: return False # Handle trailing wildcard if pattern.endswith(".*"): return event_type.startswith(pattern[:-2]) return len(pattern_parts) == len(event_parts) ``` ### Delivery Guarantees | Guarantee | Implementation | |-----------|----------------| | **At-least-once** | Events persist until acknowledged | | **Ordering** | Timestamp-based filename sorting within topic | | **Filtering** | Subscription filters applied before delivery | | **Durability** | File-based storage survives process restart | --- ## Barrier Synchronization ### Barrier Types | Type | Trigger | Use Case | |------|---------|----------| | `wave` | All executors in wave complete | Wave-level parallelism | | `phase` | All blocks in phase complete | Phase transitions | | `checkpoint` | User response received | Human-in-the-loop | | `verification` | Recognizer returns CLEAR | Quality gate | | `custom` | N of M agents signal | Flexible coordination | ### Barrier State File **File:** `.grid/events/barriers/{barrier_id}.yaml` ```yaml # .grid/events/barriers/wave_1_barrier.yaml barrier_id: "wave_1_barrier" 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 # For n_of_m type timeout_seconds: 600 # Optional timeout state: signaled_agents: - agent_id: "executor-001" signaled_at: "2026-01-24T16:40:00.000Z" result: "SUCCESS" - agent_id: "executor-002" signaled_at: "2026-01-24T16:42:00.000Z" result: "SUCCESS" waiting_agents: ["executor-003"] release_event_id: null # Populated when released ``` ### Barrier Protocol **Creating a Barrier:** ```python def create_barrier( barrier_id: str, barrier_type: str, required_agents: list[str], timeout_seconds: int = 600 ) -> None: """Create a new synchronization barrier.""" barrier = { "barrier_id": barrier_id, "barrier_type": barrier_type, "created_at": now_iso(), "status": "waiting", "condition": { "type": "all_complete", "required_agents": required_agents, "required_count": len(required_agents), "timeout_seconds": timeout_seconds }, "state": { "signaled_agents": [], "waiting_agents": required_agents.copy() }, "release_event_id": None } write_yaml(f".grid/events/barriers/{barrier_id}.yaml", barrier) # Emit barrier.wait event emit_event("system.barrier.wait", { "barrier_id": barrier_id, "barrier_type": barrier_type, "waiting_agents": required_agents, "required_count": len(required_agents) }) ``` **Signaling Barrier:** ```python def signal_barrier(barrier_id: str, agent_id: str, result: str) -> bool: """Agent signals barrier completion. Returns True if barrier released.""" barrier = load_barrier(barrier_id) if barrier["status"] != "waiting": return False # Already released or cancelled # Record signal barrier["state"]["signaled_agents"].append({ "agent_id": agent_id, "signaled_at": now_iso(), "result": result }) barrier["state"]["waiting_agents"].remove(agent_id) # Check release condition if check_barrier_condition(barrier): release_barrier(barrier) return True save_barrier(barrier) return False def check_barrier_condition(barrier: dict) -> bool: """Check if barrier condition is met.""" condition = barrier["condition"] signaled = len(barrier["state"]["signaled_agents"]) required = condition["required_count"] if condition["type"] == "all_complete": return signaled >= required elif condition["type"] == "any_complete": return signaled >= 1 elif condition["type"] == "n_of_m": return signaled >= required return False def release_barrier(barrier: dict) -> None: """Release barrier and notify waiting agents.""" barrier["status"] = "released" release_event = emit_event("system.barrier.release", { "barrier_id": barrier["barrier_id"], "released_at": now_iso(), "waiting_duration_ms": calculate_duration(barrier) }) barrier["release_event_id"] = release_event.id save_barrier(barrier) ``` ### Implicit Phase Barriers Phase transitions create implicit barriers: ```python def handle_wave_completion(wave_number: int, phase_id: str) -> None: """Handle wave completion - implicit barrier check.""" barrier_id = f"wave_{wave_number}_{phase_id}" barrier = load_barrier(barrier_id) if barrier and barrier["status"] == "released": # Wave complete - check if phase complete if is_last_wave_in_phase(wave_number, phase_id): emit_event(f"phase.{phase_id}.complete", { "phase_id": phase_id, "completed_at": now_iso() }) # Release phase barrier if exists phase_barrier_id = f"phase_{phase_id}" if barrier_exists(phase_barrier_id): signal_barrier(phase_barrier_id, "wave_executor", "SUCCESS") else: # Start next wave start_wave(wave_number + 1, phase_id) ``` --- ## Event Emission API ### Core Emit Function ```python import uuid import json from datetime import datetime, timezone def emit_event( event_type: str, payload: dict, source: str = None, target: str = None, correlation_id: str = None, causation_id: str = None, ref: str = None ) -> Event: """ Emit an event to the Event Bus. Args: event_type: Topic path (e.g., "agent.spawned") payload: Event-specific data source: Emitting agent ID (auto-detected if None) target: Target agent ID, "broadcast", or None correlation_id: Links related events causation_id: ID of causing event ref: Blackboard path for large data Returns: The emitted Event object """ event_id = f"evt_{uuid.uuid4().hex[:12]}" timestamp = datetime.now(timezone.utc).isoformat() # Get current context context = get_current_context() event = { "id": event_id, "type": event_type, "source": source or context.agent_id, "timestamp": timestamp, "target": target, "correlation_id": correlation_id or context.correlation_id, "causation_id": causation_id, "schema_version": "1.0", "payload": { **payload, "ref": ref }, "metadata": { "session_id": context.session_id, "phase": context.phase, "block": context.block, "wave": context.wave } } # Write to inbox filename = f"{timestamp.replace(':', '')}_{event_id}.json" inbox_path = f".grid/events/inbox/{filename}" write_json(inbox_path, event) # Append to stream log append_to_stream(event) return Event(**event) def append_to_stream(event: dict) -> None: """Append event to audit stream log.""" stream_path = ".grid/events/stream.log" line = "|".join([ event["timestamp"], event["id"], event["type"], event["source"], event["target"] or "broadcast", json.dumps(event["payload"], separators=(',', ':')) ]) with open(stream_path, "a") as f: f.write(line + "\n") ``` ### Convenience Functions ```python # Agent lifecycle def emit_agent_spawned(agent_id: str, agent_type: str, model: str, purpose: str) -> Event: return emit_event("agent.spawned", { "agent_id": agent_id, "agent_type": agent_type, "spawned_by": get_current_agent(), "model": model, "purpose": purpose, "spawned_at": now_iso() }) def emit_agent_complete(result_status: str, result_path: str = None) -> Event: return emit_event("agent.complete", { "agent_id": get_current_agent(), "completed_at": now_iso(), "result_status": result_status, "result_path": result_path }) def emit_agent_heartbeat(status: str, current_action: str, context_usage: float) -> Event: return emit_event("agent.heartbeat", { "agent_id": get_current_agent(), "status": status, "current_action": current_action, "context_usage_percent": context_usage, "heartbeat_at": now_iso() }) # Task lifecycle def emit_block_started(block_id: str, threads_total: int) -> Event: return emit_event("task.block.started", { "block_id": block_id, "executor_id": get_current_agent(), "started_at": now_iso(), "threads_total": threads_total }) def emit_block_progress(block_id: str, thread_current: int, thread_total: int, action: str) -> Event: return emit_event("task.block.progress", { "block_id": block_id, "executor_id": get_current_agent(), "thread_current": thread_current, "thread_total": thread_total, "percent_complete": int(thread_current / thread_total * 100), "last_action": action }) def emit_block_complete(block_id: str, threads: int, commits: list, summary_path: str) -> Event: return emit_event("task.block.complete", { "block_id": block_id, "executor_id": get_current_agent(), "completed_at": now_iso(), "threads_completed": threads, "commits": commits, "ref": summary_path }, target="mc") # Verification def emit_verify_complete(block_id: str, status: str, confidence: float, report_path: str) -> Event: return emit_event("verify.complete", { "verifier_id": get_current_agent(), "target_block": block_id, "status": status, "confidence_score": confidence, "recommendation": "auto_approve" if confidence >= 0.85 else "human_verify", "ref": report_path }, target="mc") ``` --- ## Event Consumption Patterns ### MC Event Processing Loop Master Control's main event processing loop: ```python async def mc_event_loop(): """MC's main event processing loop.""" while mission_active(): # Poll for events events = poll_events("mc") for event in events: try: await handle_event(event) except Exception as e: log_error(f"Event handling failed: {event.id}", e) # Check for stale agents (heartbeat timeout) check_staleness() # Check barrier timeouts check_barrier_timeouts() # Small delay between polls await asyncio.sleep(0.5) async def handle_event(event: Event): """Route event to appropriate handler.""" handlers = { "agent.spawned": handle_agent_spawned, "agent.complete": handle_agent_complete, "agent.error": handle_agent_error, "agent.stale": handle_agent_stale, "task.block.complete": handle_block_complete, "task.block.failed": handle_block_failed, "task.block.checkpoint": handle_block_checkpoint, "verify.complete": handle_verification_complete, "system.barrier.release": handle_barrier_release, } # Find matching handler for pattern, handler in handlers.items(): if topic_matches(event.type, pattern): await handler(event) break # Archive event archive_event(event) async def handle_block_complete(event: Event): """Handle block completion event.""" block_id = event.payload["block_id"] commits = event.payload["commits"] summary_path = event.payload.get("ref") # Update state update_state({ "last_completed_block": block_id, "total_commits": state.total_commits + len(commits) }) # Display progress display_progress(f"Block {block_id} complete ({len(commits)} commits)") # Check if wave is complete wave_barrier_id = get_current_wave_barrier() if signal_barrier(wave_barrier_id, event.source, "SUCCESS"): # Wave complete - auto-spawn Recognizer await spawn_recognizer_for_wave() async def handle_verification_complete(event: Event): """Handle verification result.""" status = event.payload["status"] confidence = event.payload["confidence_score"] recommendation = event.payload["recommendation"] if status == "CLEAR" and recommendation == "auto_approve": if get_mode() == "AUTOPILOT": # Auto-approve and continue log(f"Auto-approved (confidence: {confidence:.2f})") await proceed_to_next_wave() else: # Show to user await present_verification_result(event) elif status == "GAPS_FOUND": # Spawn Planner for gap closure gaps_report = load_file(event.payload.get("ref")) await spawn_planner_gaps(gaps_report) else: # Present to user await present_verification_result(event) ``` ### Executor Event Emission Pattern Executor emits events during execution: ```python def execute_block(block_id: str, plan: dict): """Execute a block with event emission.""" threads = plan["threads"] total_threads = len(threads) commits = [] # Emit block started emit_block_started(block_id, total_threads) for i, thread in enumerate(threads, 1): # Emit progress emit_block_progress( block_id, thread_current=i, thread_total=total_threads, action=f"Starting thread: {thread['name']}" ) # Emit heartbeat emit_agent_heartbeat( status="active", current_action=f"Executing thread {i}: {thread['name']}", context_usage=estimate_context_usage() ) try: # Execute thread commit = execute_thread(thread) commits.append(commit) except CheckpointRequired as cp: # Emit checkpoint event emit_event("task.block.checkpoint", { "block_id": block_id, "executor_id": get_current_agent(), "checkpoint_type": cp.type, "checkpoint_path": cp.path, "thread_blocked": i }, target="mc") # Wait for checkpoint resolution wait_for_checkpoint_resolution() except Exception as e: # Emit failure emit_event("task.block.failed", { "block_id": block_id, "executor_id": get_current_agent(), "failed_at": now_iso(), "thread_failed": i, "reason": str(e) }, target="mc") raise # Emit completion summary_path = write_summary(block_id, commits) emit_block_complete(block_id, total_threads, commits, summary_path) emit_agent_complete("SUCCESS", summary_path) ``` --- ## Event Ordering Guarantees ### Ordering Rules 1. **Within Topic:** Events maintain FIFO order by timestamp 2. **Across Topics:** No global ordering guarantee 3. **From Same Source:** Events from same agent are ordered 4. **Causation Chain:** `causation_id` links enable reconstruction ### Ordering Implementation ```python def poll_events(agent_id: str) -> list[Event]: """Poll events with ordering guarantees.""" # Get matching events from inbox events = get_matching_events(agent_id) # Sort by timestamp (FIFO within topic) events_by_topic = defaultdict(list) for event in events: events_by_topic[event.type].append(event) ordered_events = [] for topic, topic_events in events_by_topic.items(): # Sort by timestamp within topic topic_events.sort(key=lambda e: e.timestamp) ordered_events.extend(topic_events) # Final sort by timestamp for interleaving ordered_events.sort(key=lambda e: e.timestamp) return ordered_events ``` ### Causation Chain Reconstruction ```python def reconstruct_causation_chain(event_id: str) -> list[Event]: """Reconstruct chain of events leading to this one.""" chain = [] current_id = event_id while current_id: event = load_event_by_id(current_id) if event: chain.append(event) current_id = event.causation_id else: break # Reverse to get chronological order chain.reverse() return chain ``` --- ## Integration with Existing Grid Components ### MC Integration MC uses Event Bus for all agent coordination: ```python # In mc.md workflow # When spawning executor executor_id = generate_agent_id("executor") emit_agent_spawned(executor_id, "executor", model, f"Execute {block_id}") # Create wave barrier create_barrier( f"wave_{wave_number}_{phase_id}", "wave", executors_in_wave ) # Spawn executors (they emit events during execution) Task(prompt=..., ...) # MC event loop handles responses via events ``` ### Scratchpad Integration Scratchpad entries can trigger events: ```python def write_scratchpad_entry(category: str, topic: str, content: str, relevance: str): """Write scratchpad entry and emit event if high relevance.""" # Write to scratchpad entry = format_entry(category, topic, content, relevance) append_to_scratchpad(entry) # Emit event for HIGH relevance discoveries if relevance == "HIGH": emit_event("agent.discovery", { "agent_id": get_current_agent(), "category": category, "topic": topic, "relevance": relevance, "ref": ".grid/SCRATCHPAD.md" }) ``` ### Checkpoint Integration Checkpoints emit events for MC: ```python def write_checkpoint(reason: str, position: dict, **kwargs): """Write checkpoint file and emit event.""" checkpoint_path = ".grid/CHECKPOINT.md" # Write checkpoint file write_checkpoint_file(checkpoint_path, reason, position, **kwargs) # Emit event emit_event("system.checkpoint.created", { "checkpoint_path": checkpoint_path, "reason": reason, "position": position }, target="mc") ``` ### STATE.md Synchronization State updates can be event-driven: ```python def update_state_from_events(): """Synchronize STATE.md from event stream.""" events = replay_events_since(state.last_event_processed) for event in events: if event.type == "task.block.complete": state.blocks_complete += 1 elif event.type == "agent.spawned": state.agents_spawned += 1 elif event.type.startswith("phase.") and event.type.endswith(".complete"): state.phases_complete += 1 state.last_event_processed = events[-1].id if events else state.last_event_processed save_state() ``` --- ## Future Considerations ### In-Memory Event Bus (v2) For real-time performance in long-running sessions: ```python # Redis-based implementation class RedisEventBus: def __init__(self, redis_client): self.redis = redis_client async def emit(self, event: Event): # Publish to topic channel await self.redis.publish(event.type, event.json()) # Store in stream for persistence await self.redis.xadd("grid:events", event.dict()) async def subscribe(self, patterns: list[str]): pubsub = self.redis.pubsub() for pattern in patterns: await pubsub.psubscribe(pattern.replace("*", "*")) return pubsub ``` ### Event Replay for Debugging ```python def replay_mission(session_id: str, speed: float = 1.0): """Replay a mission from event stream for debugging.""" events = load_events_for_session(session_id) prev_timestamp = None for event in events: # Calculate delay if prev_timestamp: delay = (event.timestamp - prev_timestamp).total_seconds() / speed time.sleep(delay) # Display event display_event(event) prev_timestamp = event.timestamp ``` ### Metrics and Monitoring ```python def collect_event_metrics(session_id: str) -> dict: """Collect metrics from event stream.""" events = load_events_for_session(session_id) return { "total_events": len(events), "events_by_type": Counter(e.type for e in events), "events_by_source": Counter(e.source for e in events), "agents_spawned": sum(1 for e in events if e.type == "agent.spawned"), "blocks_completed": sum(1 for e in events if e.type == "task.block.complete"), "barriers_released": sum(1 for e in events if e.type == "system.barrier.release"), "checkpoints_hit": sum(1 for e in events if e.type == "system.checkpoint.created"), "total_duration": calculate_duration(events[0], events[-1]) if events else 0 } ``` --- ## Configuration ### Event Bus Configuration **File:** `.grid/config.json` (event_bus section) ```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 } } ``` ### Environment Variables ```bash # Enable/disable event bus GRID_EVENT_BUS_ENABLED=true # Event retention GRID_EVENT_RETENTION_DAYS=7 # Barrier defaults GRID_BARRIER_TIMEOUT_SECONDS=600 ``` --- ## Summary The Grid Event Bus provides: 1. **Decoupled Communication** - Agents communicate via events, not direct calls 2. **Topic-Based Routing** - Flexible pub/sub with wildcards 3. **Synchronization Barriers** - First-class support for parallel execution coordination 4. **File-Based Durability** - Events persist across sessions for replay and audit 5. **Reference-Based Payloads** - Large data stays in Blackboard, events carry references 6. **Ordering Guarantees** - FIFO within topics, causation chains for reconstruction The file-based implementation works immediately without external dependencies. Future versions can migrate to in-memory or Redis-based implementations for real-time performance while maintaining the same API. --- *End of Event Bus Specification. End of Line.*