repo_description: Multi-Agent Harness for Production AI
repo_topics: agent agent-framework agent-skills anthropic automation autonomous-agents claude harness harness-engineering human-in-the-loop openai python self-hosted self-improving
code: """Tests for EventBus pub/sub event system.
  
  Validates subscription management, event publishing, filtering,
  concurrency handling, history operations, and convenience publishers.
  """
  
  from __future__ import annotations
  
  import asyncio
  from datetime import datetime
  
  import pytest
  
  from framework.host.event_bus import (
      AgentEvent,
      EventBus,
      EventType,
  )
  
  
  # ---------------------------------------------------------------------------
  # AgentEvent dataclass tests
  # ---------------------------------------------------------------------------
  class TestAgentEvent:
      """Tests for AgentEvent dataclass."""
  
      def test_minimal_construction(self):
          """Event can be created with just type and stream_id."""
          event = AgentEvent(type=EventType.EXECUTION_STARTED, stream_id="test_stream")
          assert event.type == EventType.EXECUTION_STARTED
          assert event.stream_id == "test_stream"
          assert event.node_id is None
          assert event.execution_id is None
          assert event.data == {}
          assert event.correlation_id is None
  
      def test_full_construction(self):
          """Event stores all provided fields."""
          event = AgentEvent(
              type=EventType.TOOL_CALL_COMPLETED,
              stream_id="stream_1",
              node_id="node_1",
              execution_id="exec_123",
              data={"result": "success"},
              correlation_id="corr_456",
          )
          assert event.type == EventType.TOOL_CALL_COMPLETED
          assert event.stream_id == "stream_1"
          assert event.node_id == "node_1"
          assert event.execution_id == "exec_123"
          assert event.data == {"result": "success"}
          assert event.correlation_id == "corr_456"
  
      def test_timestamp_auto_generated(self):
          """Timestamp is auto-generated if not provided."""
          before = datetime.now()
          event = AgentEvent(type=EventType.EXECUTION_STARTED, stream_id="test")
          after = datetime.now()
          assert before <= event.timestamp <= after
  
      def test_to_dict_serialization(self):
          """Event can be serialized to dictionary."""
          event = AgentEvent(
              type=EventType.EXECUTION_COMPLETED,
              stream_id="stream_1",
              node_id="node_1",
              execution_id="exec_1",
              data={"output": "result"},
              correlation_id="corr_1",
              colony_id="colony_1",
          )
          d = event.to_dict()
          assert d["type"] == "execution_completed"
          assert d["stream_id"] == "stream_1"
          assert d["colony_id"] == "colony_1"
  
      def test_to_dict_includes_run_id(self):
          """run_id is included in to_dict() when set."""
          event = AgentEvent(
              type=EventType.EXECUTION_STARTED,
              stream_id="s1",
              run_id="run-abc",
          )
          d = event.to_dict()
          assert d["run_id"] == "run-abc"
  
      def test_to_dict_omits_run_id_when_none(self):
          """run_id is omitted from to_dict() when None."""
          event = AgentEvent(
              type=EventType.EXECUTION_STARTED,
              stream_id="s1",
          )
          d = event.to_dict()
          assert "run_id" not in d
  
  
  # ---------------------------------------------------------------------------
  # Subscription management tests
  # ---------------------------------------------------------------------------
  class TestSubscriptionManagement:
      """Tests for subscribe/unsubscribe operations."""
  
      def test_subscribe_returns_id(self):
          """subscribe() returns a subscription ID."""
          bus = EventBus()
  
          async def handler(event: AgentEvent) -> None:
              pass
  
          sub_id = bus.subscribe(event_types=[EventType.EXECUTION_STARTED], handler=handler)
          assert sub_id.startswith("sub_")
  
      def test_subscribe_increments_id(self):
          """Each subscription gets a unique incremented ID."""
          bus = EventBus()
  
          async def handler(event: AgentEvent) -> None:
              pass
  
          id1 = bus.subscribe(event_types=[EventType.EXECUTION_STARTED], handler=handler)
          id2 = bus.subscribe(event_types=[EventType.EXECUTION_COMPLETED], handler=handler)
          id3 = bus.subscribe(event_types=[EventType.EXECUTION_FAILED], handler=handler)
  
          assert id1 == "sub_1"
          assert id2 == "sub_2"
          assert id3 == "sub_3"
  
      def test_unsubscribe_removes_subscription(self):
          """unsubscribe() removes the subscription."""
          bus = EventBus()
  
          async def handler(event: AgentEvent) -> None:
              pass
  
          sub_id = bus.subscribe(event_types=[EventType.EXECUTION_STARTED], handler=handler)
          assert sub_id in bus._subscriptions
  
          result = bus.unsubscribe(sub_id)
          assert result is True
          assert sub_id not in bus._subscriptions
  
      def test_unsubscribe_nonexistent_returns_false(self):
          """unsubscribe() returns False for non-existent subscription."""
          bus = EventBus()
          result = bus.unsubscribe("sub_nonexistent")
          assert result is False
  
      def test_multiple_subscriptions_same_event_type(self):
          """Multiple handlers can subscribe to the same event type."""
          bus = EventBus()
          received = []
  
          async def handler1(event: AgentEvent) -> None:
              received.append("handler1")
  
          async def handler2(event: AgentEvent) -> None:
              received.append("handler2")
  
          bus.subscribe(event_types=[EventType.EXECUTION_STARTED], handler=handler1)
          bus.subscribe(event_types=[EventType.EXECUTION_STARTED], handler=handler2)
  
          assert len(bus._subscriptions) == 2
  
  
  # ---------------------------------------------------------------------------
  # Event publishing tests
  # ---------------------------------------------------------------------------
  class TestEventPublishing:
      """Tests for publish() and event delivery."""
  
      @pytest.mark.asyncio
      async def test_publish_delivers_to_subscriber(self):
          """Published events are delivered to matching subscribers."""
          bus = EventBus()
          received_events = []
  
          async def handler(event: AgentEvent) -> None:
              received_events.append(event)
  
          bus.subscribe(event_types=[EventType.EXECUTION_STARTED], handler=handler)
  
          event = AgentEvent(type=EventType.EXECUTION_STARTED, stream_id="test")
          await bus.publish(event)
  
          assert len(received_events) == 1
          assert received_events[0] == event
  
      @pytest.mark.asyncio
      async def test_publish_to_multiple_subscribers(self):
          """Event is delivered to all matching subscribers."""
          bus = EventBus()
          received = []
  
          async def handler1(event: AgentEvent) -> None:
              received.append("h1")
  
          async def handler2(event: AgentEvent) -> None:
              received.append("h2")
  
          bus.subscribe(event_types=[EventType.EXECUTION_STARTED], handler=handler1)
          bus.subscribe(event_types=[EventType.EXECUTION_STARTED], handler=handler2)
  
          await bus.publish(AgentEvent(type=EventType.EXECUTION_STARTED, stream_id="test"))
  
          assert "h1" in received
          assert "h2" in received
  
      @pytest.mark.asyncio
      async def test_publish_non_matching_type_not_delivered(self):
          """Events with non-matching types are not delivered."""
          bus = EventBus()
          received = []
  
          async def handler(event: AgentEvent) -> None:
              received.append(event)
  
          bus.subscribe(event_types=[EventType.EXECUTION_STARTED], handler=handler)
  
          await bus.publish(AgentEvent(type=EventType.EXECUTION_COMPLETED, stream_id="test"))
  
          assert len(received) == 0
  
      @pytest.mark.asyncio
      async def test_publish_adds_to_history(self):
          """Published events are added to history."""
          bus = EventBus()
  
          event = AgentEvent(type=EventType.EXECUTION_STARTED, stream_id="test")
          await bus.publish(event)
  
          history = bus.get_history()
          assert len(history) == 1
          assert history[0] == event
  
  
  # ---------------------------------------------------------------------------
  # Filter tests
  # ---------------------------------------------------------------------------
  class TestEventFiltering:
      """Tests for subscription filters."""
  
      @pytest.mark.asyncio
      async def test_filter_by_stream(self):
          """filter_stream only receives events from that stream."""
          bus = EventBus()
          received = []
  
          async def handler(event: AgentEvent) -> None:
              received.append(event.stream_id)
  
          bus.subscribe(
              event_types=[EventType.EXECUTION_STARTED],
              handler=handler,
              filter_stream="stream_a",
          )
  
          await bus.publish(AgentEvent(type=EventType.EXECUTION_STARTED, stream_id="stream_a"))
          await bus.publish(AgentEvent(type=EventType.EXECUTION_STARTED, stream_id="stream_b"))
  
          assert received == ["stream_a"]
  
      @pytest.mark.asyncio
      async def test_filter_by_node(self):
          """filter_node only receives events from that node."""
          bus = EventBus()
          received = []
  
          async def handler(event: AgentEvent) -> None:
              received.append(event.node_id)
  
          bus.subscribe(
              event_types=[EventType.NODE_LOOP_STARTED],
              handler=handler,
              filter_node="node_x",
          )
  
          await bus.publish(AgentEvent(type=EventType.NODE_LOOP_STARTED, stream_id="s", node_id="node_x"))
          await bus.publish(AgentEvent(type=EventType.NODE_LOOP_STARTED, stream_id="s", node_id="node_y"))
  
          assert received == ["node_x"]
  
      @pytest.mark.asyncio
      async def test_filter_by_execution(self):
          """filter_execution only receives events from that execution."""
          bus = EventBus()
          received = []
  
          async def handler(event: AgentEvent) -> None:
              received.append(event.execution_id)
  
          bus.subscribe(
              event_types=[EventType.EXECUTION_COMPLETED],
              handler=handler,
              filter_execution="exec_1",
          )
  
          await bus.publish(AgentEvent(type=EventType.EXECUTION_COMPLETED, stream_id="s", execution_id="exec_1"))
          await bus.publish(AgentEvent(type=EventType.EXECUTION_COMPLETED, stream_id="s", execution_id="exec_2"))
  
          assert received == ["exec_1"]
  
      @pytest.mark.asyncio
      async def test_combined_filters(self):
          """Multiple filters are AND-ed together."""
          bus = EventBus()
          received = []
  
          async def handler(event: AgentEvent) -> None:
              received.append(True)
  
          bus.subscribe(
              event_types=[EventType.TOOL_CALL_COMPLETED],
              handler=handler,
              filter_stream="stream_1",
              filter_node="node_1",
          )
  
          # Matches both filters
          await bus.publish(
              AgentEvent(
                  type=EventType.TOOL_CALL_COMPLETED,
                  stream_id="stream_1",
                  node_id="node_1",
              )
          )
          # Matches stream but not node
          await bus.publish(
              AgentEvent(
                  type=EventType.TOOL_CALL_COMPLETED,
                  stream_id="stream_1",
                  node_id="node_2",
              )
          )
          # Matches node but not stream
          await bus.publish(
              AgentEvent(
                  type=EventType.TOOL_CALL_COMPLETED,
                  stream_id="stream_2",
                  node_id="node_1",
              )
          )
  
          assert len(received) == 1
  
      @pytest.mark.asyncio
      async def test_filter_by_colony(self):
          """filter_colony only receives events from that colony."""
          bus = EventBus()
          received = []
  
          async def handler(event: AgentEvent) -> None:
              received.append(event.colony_id)
  
          bus.subscribe(
              event_types=[EventType.EXECUTION_STARTED],
              handler=handler,
              filter_colony="colony_a",
          )
  
          await bus.publish(AgentEvent(type=EventType.EXECUTION_STARTED, stream_id="s", colony_id="colony_a"))
          await bus.publish(AgentEvent(type=EventType.EXECUTION_STARTED, stream_id="s", colony_id="colony_b"))
  
          assert received == ["colony_a"]
  
  
  # ---------------------------------------------------------------------------
  # Concurrency tests
  # ---------------------------------------------------------------------------
  class TestConcurrency:
      """Tests for concurrent handler execution."""
  
      @pytest.mark.asyncio
      async def test_handler_error_doesnt_crash_others(self):
          """One handler's error doesn't prevent other handlers from running."""
          bus = EventBus()
          results = []
  
          async def failing_handler(event: AgentEvent) -> None:
              raise ValueError("Handler error!")
  
          async def working_handler(event: AgentEvent) -> None:
              results.append("success")
  
          bus.subscribe(event_types=[EventType.EXECUTION_STARTED], handler=failing_handler)
          bus.subscribe(event_types=[EventType.EXECUTION_STARTED], handler=working_handler)
  
          await bus.publish(AgentEvent(type=EventType.EXECUTION_STARTED, stream_id="test"))
  
          assert "success" in results
  
      @pytest.mark.asyncio
      async def test_max_concurrent_handlers_respected(self):
          """Semaphore limits concurrent handler executions."""
          bus = EventBus(max_concurrent_handlers=2)
          concurrent_count = 0
          max_concurrent = 0
  
          async def slow_handler(event: AgentEvent) -> None:
              nonlocal concurrent_count, max_concurrent
              concurrent_count += 1
              max_concurrent = max(max_concurrent, concurrent_count)
              await asyncio.sleep(0.1)
              concurrent_count -= 1
  
          # Subscribe 5 handlers
          for _ in range(5):
              bus.subscribe(event_types=[EventType.EXECUTION_STARTED], handler=slow_handler)
  
          await bus.publish(AgentEvent(type=EventType.EXECUTION_STARTED, stream_id="test"))
  
          # Max concurrent should be limited to 2
          assert max_concurrent <= 2
  
  
  # ---------------------------------------------------------------------------
  # History and query tests
  # ---------------------------------------------------------------------------
  class TestHistoryAndQueries:
      """Tests for get_history() and get_stats()."""
  
      @pytest.mark.asyncio
      async def test_history_returns_events_most_recent_first(self):
          """get_history() returns events in reverse chronological order."""
          bus = EventBus()
  
          await bus.publish(AgentEvent(type=EventType.EXECUTION_STARTED, stream_id="s1"))
          await bus.publish(AgentEvent(type=EventType.EXECUTION_COMPLETED, stream_id="s2"))
          await bus.publish(AgentEvent(type=EventType.EXECUTION_FAILED, stream_id="s3"))
  
          history = bus.get_history()
          assert history[0].stream_id == "s3"  # Most recent
          assert history[1].stream_id == "s2"
          assert history[2].stream_id == "s1"  # Oldest
  
      @pytest.mark.asyncio
      async def test_history_filter_by_event_type(self):
          """get_history() can filter by event type."""
          bus = EventBus()
  
          await bus.publish(AgentEvent(type=EventType.EXECUTION_STARTED, stream_id="s"))
          await bus.publish(AgentEvent(type=EventType.EXECUTION_COMPLETED, stream_id="s"))
          await bus.publish(AgentEvent(type=EventType.EXECUTION_STARTED, stream_id="s"))
  
          history = bus.get_history(event_type=EventType.EXECUTION_STARTED)
          assert len(history) == 2
          assert all(e.type == EventType.EXECUTION_STARTED for e in history)
  
      @pytest.mark.asyncio
      async def test_history_filter_by_stream_id(self):
          """get_history() can filter by stream_id."""
          bus = EventBus()
  
          await bus.publish(AgentEvent(type=EventType.EXECUTION_STARTED, stream_id="stream_a"))
          await bus.publish(AgentEvent(type=EventType.EXECUTION_STARTED, stream_id="stream_b"))
          await bus.publish(AgentEvent(type=EventType.EXECUTION_STARTED, stream_id="stream_a"))
  
          history = bus.get_history(stream_id="stream_a")
          assert len(history) == 2
          assert all(e.stream_id == "stream_a" for e in history)
  
      @pytest.mark.asyncio
      async def test_history_limit(self):
          """get_history() respects limit parameter."""
          bus = EventBus()
  
          for i in range(10):
              await bus.publish(AgentEvent(type=EventType.EXECUTION_STARTED, stream_id=f"s{i}"))
  
          history = bus.get_history(limit=3)
          assert len(history) == 3
  
      @pytest.mark.asyncio
      async def test_max_history_enforced(self):
          """EventBus enforces max_history limit."""
          bus = EventBus(max_history=5)
  
          for i in range(10):
              await bus.publish(AgentEvent(type=EventType.EXECUTION_STARTED, stream_id=f"s{i}"))
  
          assert len(bus._event_history) == 5
          # Should have the 5 most recent
          assert bus._event_history[-1].stream_id == "s9"
          assert bus._event_history[0].stream_id == "s5"
  
      @pytest.mark.asyncio
      async def test_get_stats(self):
          """get_stats() returns accurate statistics."""
          bus = EventBus()
  
          async def handler(event: AgentEvent) -> None:
              pass
  
          bus.subscribe(event_types=[EventType.EXECUTION_STARTED], handler=handler)
          bus.subscribe(event_types=[EventType.EXECUTION_COMPLETED], handler=handler)
  
          await bus.publish(AgentEvent(type=EventType.EXECUTION_STARTED, stream_id="s"))
          await bus.publish(AgentEvent(type=EventType.EXECUTION_STARTED, stream_id="s"))
          await bus.publish(AgentEvent(type=EventType.EXECUTION_COMPLETED, stream_id="s"))
  
          stats = bus.get_stats()
          assert stats["total_events"] == 3
          assert stats["subscriptions"] == 2
          assert stats["events_by_type"]["execution_started"] == 2
          assert stats["events_by_type"]["execution_completed"] == 1
  
  
  # ---------------------------------------------------------------------------
  # Wait operations tests
  # ---------------------------------------------------------------------------
  class TestWaitOperations:
      """Tests for wait_for() async waiting."""
  
      @pytest.mark.asyncio
      async def test_wait_for_receives_event(self):
          """wait_for() returns when matching event is published."""
          bus = EventBus()
  
          async def publish_later():
              await asyncio.sleep(0.05)
              await bus.publish(
                  AgentEvent(
                      type=EventType.EXECUTION_COMPLETED,
                      stream_id="test",
                      execution_id="exec_1",
                  )
              )
  
          asyncio.create_task(publish_later())
  
          event = await bus.wait_for(
              event_type=EventType.EXECUTION_COMPLETED,
              timeout=1.0,
          )
  
          assert event is not None
          assert event.type == EventType.EXECUTION_COMPLETED
  
      @pytest.mark.asyncio
      async def test_wait_for_timeout_returns_none(self):
          """wait_for() returns None on timeout."""
          bus = EventBus()
  
          event = await bus.wait_for(
              event_type=EventType.EXECUTION_COMPLETED,
              timeout=0.05,
          )
  
          assert event is None
  
      @pytest.mark.asyncio
      async def test_wait_for_with_filters(self):
          """wait_for() respects filters."""
          bus = EventBus()
  
          async def publish_events():
              await asyncio.sleep(0.02)
              # This one shouldn't match
              await bus.publish(
                  AgentEvent(
                      type=EventType.EXECUTION_COMPLETED,
                      stream_id="wrong_stream",
                  )
              )
              await asyncio.sleep(0.02)
              # This one should match
              await bus.publish(
                  AgentEvent(
                      type=EventType.EXECUTION_COMPLETED,
                      stream_id="correct_stream",
                  )
              )
  
          asyncio.create_task(publish_events())
  
          event = await bus.wait_for(
              event_type=EventType.EXECUTION_COMPLETED,
              stream_id="correct_stream",
              timeout=1.0,
          )
  
          assert event is not None
          assert event.stream_id == "correct_stream"
  
      @pytest.mark.asyncio
      async def test_wait_for_cleans_up_subscription(self):
          """wait_for() removes its subscription after completion."""
          bus = EventBus()
  
          initial_count = len(bus._subscriptions)
  
          async def publish_later():
              await asyncio.sleep(0.02)
              await bus.publish(AgentEvent(type=EventType.EXECUTION_COMPLETED, stream_id="s"))
  
          asyncio.create_task(publish_later())
  
          await bus.wait_for(event_type=EventType.EXECUTION_COMPLETED, timeout=1.0)
  
          assert len(bus._subscriptions) == initial_count
  
  
  # ---------------------------------------------------------------------------
  # Convenience publisher tests
  # ---------------------------------------------------------------------------
  class TestConveniencePublishers:
      """Tests for emit_* convenience methods."""
  
      @pytest.mark.asyncio
      async def test_emit_execution_started(self):
          """emit_execution_started publishes correct event."""
          bus = EventBus()
          received = []
  
          async def handler(event: AgentEvent) -> None:
              received.append(event)
  
          bus.subscribe(event_types=[EventType.EXECUTION_STARTED], handler=handler)
  
          await bus.emit_execution_started(
              stream_id="test_stream",
              execution_id="exec_1",
              input_data={"key": "value"},
              correlation_id="corr_1",
          )
  
          assert len(received) == 1
          assert received[0].type == EventType.EXECUTION_STARTED
          assert received[0].stream_id == "test_stream"
          assert received[0].execution_id == "exec_1"
          assert received[0].data == {"input": {"key": "value"}}
          assert received[0].correlation_id == "corr_1"
  
      @pytest.mark.asyncio
      async def test_emit_execution_completed(self):
          """emit_execution_completed publishes correct event."""
          bus = EventBus()
          received = []
  
          async def handler(event: AgentEvent) -> None:
              received.append(event)
  
          bus.subscribe(event_types=[EventType.EXECUTION_COMPLETED], handler=handler)
  
          await bus.emit_execution_completed(
              stream_id="s",
              execution_id="e",
              output={"result": "success"},
          )
  
          assert len(received) == 1
          assert received[0].type == EventType.EXECUTION_COMPLETED
          assert received[0].data == {"output": {"result": "success"}}
  
      @pytest.mark.asyncio
      async def test_emit_execution_failed(self):
          """emit_execution_failed publishes correct event."""
          bus = EventBus()
          received = []
  
          async def handler(event: AgentEvent) -> None:
              received.append(event)
  
          bus.subscribe(event_types=[EventType.EXECUTION_FAILED], handler=handler)
  
          await bus.emit_execution_failed(
              stream_id="s",
              execution_id="e",
              error="Something went wrong",
          )
  
          assert len(received) == 1
          assert received[0].type == EventType.EXECUTION_FAILED
          assert received[0].data == {"error": "Something went wrong"}
  
      @pytest.mark.asyncio
      async def test_emit_tool_call_started(self):
          """emit_tool_call_started publishes correct event."""
          bus = EventBus()
          received = []
  
          async def handler(event: AgentEvent) -> None:
              received.append(event)
  
          bus.subscribe(event_types=[EventType.TOOL_CALL_STARTED], handler=handler)
  
          await bus.emit_tool_call_started(
              stream_id="s",
              node_id="n",
              tool_use_id="tool_1",
              tool_name="web_search",
              tool_input={"query": "test"},
          )
  
          assert len(received) == 1
          assert received[0].data["tool_name"] == "web_search"
          assert received[0].data["tool_input"] == {"query": "test"}
  
      @pytest.mark.asyncio
      async def test_emit_tool_call_completed(self):
          """emit_tool_call_completed publishes correct event."""
          bus = EventBus()
          received = []
  
          async def handler(event: AgentEvent) -> None:
              received.append(event)
  
          bus.subscribe(event_types=[EventType.TOOL_CALL_COMPLETED], handler=handler)
  
          await bus.emit_tool_call_completed(
              stream_id="s",
              node_id="n",
              tool_use_id="tool_1",
              tool_name="web_search",
              result="search results",
              is_error=False,
          )
  
          assert len(received) == 1
          assert received[0].data["result"] == "search results"
          assert received[0].data["is_error"] is False
  
      @pytest.mark.asyncio
      async def test_emit_webhook_received(self):
          """emit_webhook_received publishes correct event."""
          bus = EventBus()
          received = []
  
          async def handler(event: AgentEvent) -> None:
              received.append(event)
  
          bus.subscribe(event_types=[EventType.WEBHOOK_RECEIVED], handler=handler)
  
          await bus.emit_webhook_received(
              source_id="webhook_1",
              path="/api/webhook",
              method="POST",
              headers={"Content-Type": "application/json"},
              payload={"data": "test"},
              query_params={"token": "abc"},
          )
  
          assert len(received) == 1
          assert received[0].data["path"] == "/api/webhook"
          assert received[0].data["method"] == "POST"
          assert received[0].data["payload"] == {"data": "test"}
  
      @pytest.mark.asyncio
      async def test_emit_tool_doom_loop(self):
          """emit_tool_doom_loop publishes correct event."""
          bus = EventBus()
          received = []
  
          async def handler(event: AgentEvent) -> None:
              received.append(event)
  
          bus.subscribe(event_types=[EventType.NODE_TOOL_DOOM_LOOP], handler=handler)
  
          await bus.emit_tool_doom_loop(
              stream_id="test_stream",
              node_id="node_1",
              description="Tool called same endpoint 5 times",
              execution_id="exec_1",
          )
  
          assert len(received) == 1
          assert received[0].type == EventType.NODE_TOOL_DOOM_LOOP
          assert received[0].stream_id == "test_stream"
          assert received[0].node_id == "node_1"
          assert received[0].data["description"] == "Tool called same endpoint 5 times"
  
      @pytest.mark.asyncio
      async def test_emit_escalation_requested(self):
          """emit_escalation_requested publishes correct event."""
          bus = EventBus()
          received = []
  
          async def handler(event: AgentEvent) -> None:
              received.append(event)
  
          bus.subscribe(event_types=[EventType.ESCALATION_REQUESTED], handler=handler)
  
          await bus.emit_escalation_requested(
              stream_id="test_stream",
              node_id="node_1",
              reason="Need human intervention",
              context="Complex decision required",
              execution_id="exec_1",
          )
  
          assert len(received) == 1
          assert received[0].type == EventType.ESCALATION_REQUESTED
          assert received[0].stream_id == "test_stream"
          assert received[0].node_id == "node_1"
          assert received[0].data["reason"] == "Need human intervention"
          assert received[0].data["context"] == "Complex decision required"
  
      @pytest.mark.asyncio
      async def test_emit_llm_turn_complete(self):
          """emit_llm_turn_complete publishes correct event."""
          bus = EventBus()
          received = []
  
          async def handler(event: AgentEvent) -> None:
              received.append(event)
  
          bus.subscribe(event_types=[EventType.LLM_TURN_COMPLETE], handler=handler)
  
          await bus.emit_llm_turn_complete(
              stream_id="test_stream",
              node_id="node_1",
              stop_reason="end_turn",
              model="claude-sonnet-4-20250514",
              input_tokens=100,
              output_tokens=50,
              execution_id="exec_1",
              iteration=3,
          )
  
          assert len(received) == 1
          assert received[0].type == EventType.LLM_TURN_COMPLETE
          assert received[0].data["stop_reason"] == "end_turn"
          assert received[0].data["model"] == "claude-sonnet-4-20250514"
          assert received[0].data["input_tokens"] == 100
          assert received[0].data["output_tokens"] == 50
          assert received[0].data["iteration"] == 3
  
      @pytest.mark.asyncio
      async def test_emit_node_action_plan(self):
          """emit_node_action_plan publishes correct event."""
          bus = EventBus()
          received = []
  
          async def handler(event: AgentEvent) -> None:
              received.append(event)
  
          bus.subscribe(event_types=[EventType.NODE_ACTION_PLAN], handler=handler)
  
          await bus.emit_node_action_plan(
              stream_id="test_stream",
              node_id="node_1",
              plan="1. Search for data\n2. Analyze results\n3. Generate report",
              execution_id="exec_1",
          )
  
          assert len(received) == 1
          assert received[0].type == EventType.NODE_ACTION_PLAN
          assert received[0].data["plan"] == "1. Search for data\n2. Analyze results\n3. Generate report"
  
      @pytest.mark.asyncio
      async def test_emit_subagent_report(self):
          """emit_subagent_report publishes correct event."""
          bus = EventBus()
          received = []
  
          async def handler(event: AgentEvent) -> None:
              received.append(event)
  
          bus.subscribe(event_types=[EventType.SUBAGENT_REPORT], handler=handler)
  
          await bus.emit_subagent_report(
              stream_id="test_stream",
              node_id="queen",
              subagent_id="worker-1",
              message="Task 50% complete",
              data={"progress": 0.5},
          )
  
          assert len(received) == 1
          assert received[0].type == EventType.SUBAGENT_REPORT
          assert received[0].data["subagent_id"] == "worker-1"
          assert received[0].data["message"] == "Task 50% complete"
          assert received[0].data["data"]["progress"] == 0.5
  
  
  # ---------------------------------------------------------------------------
  # EventType enum tests
  # ---------------------------------------------------------------------------
  class TestEventType:
      """Tests for EventType enum."""
  
      def test_all_event_types_are_strings(self):
          """All EventType values are strings."""
          for event_type in EventType:
              assert isinstance(event_type.value, str)
  
      def test_event_types_are_unique(self):
          """All EventType values are unique."""
          values = [e.value for e in EventType]
          assert len(values) == len(set(values))
  
      def test_key_event_types_exist(self):
          """Key event types are defined."""
          assert EventType.EXECUTION_STARTED
          assert EventType.EXECUTION_COMPLETED
          assert EventType.EXECUTION_FAILED
          assert EventType.EXECUTION_PAUSED
          assert EventType.EXECUTION_RESUMED
          assert EventType.TOOL_CALL_STARTED
          assert EventType.TOOL_CALL_COMPLETED
          assert EventType.WEBHOOK_RECEIVED
          assert EventType.NODE_TOOL_DOOM_LOOP
          assert EventType.ESCALATION_REQUESTED
          assert EventType.LLM_TURN_COMPLETE
          assert EventType.NODE_ACTION_PLAN
          assert EventType.WORKER_COLONY_LOADED
          assert EventType.CREDENTIALS_REQUIRED
          assert EventType.EXECUTION_RESURRECTED
          assert EventType.QUEEN_PHASE_CHANGED
          assert EventType.SUBAGENT_REPORT
          assert EventType.TRIGGER_AVAILABLE
          assert EventType.TRIGGER_FIRED
  