simple event bus impl

AGENTS.md

@@ -9,11 +9,16 @@ yes: assert dict['field'] == dict['field'] unless we know better
 
 more later
 
-# when you move through code, add docstrings / comment weird logic
+# when you move through code, add docstrings / comment weird/complicated logic
 
 # use available MCP tools/skills
 
-- serena mcp: symbol discovery
+- serena mcp: symbol discovery. Has some of LSP stuff
 - context7: docs
 
-- skills: outside of workspace, request commands
+- skills: outside of workspace, request commands. Use when appropiate
+
+# docs/ pair with `superpowers` and other skills
+
+superpowers, a Codex-cli plugin, is a set of skills. check paths given by Codex
+if not found, might be stale hash

docs/superpowers/plans/2026-05-12-unified-mcp-surface.md

@@ -314,13 +314,13 @@ config_reloaded
 - Modify: unified server files from Phase 3.
 - Test: `tests/wf_mcp/test_events.py`
 
-- [ ] Introduce an in-process event bus abstraction.
-- [ ] Keep the existing stored `McpEvent` list as one subscriber/sink.
+- [x] Introduce an in-process event bus abstraction.
+- [x] Keep the existing stored `McpEvent` list as one subscriber/sink.
 - [ ] Add event kinds for workflow artifacts and deployments:
-  - `workflow_artifact_saved`
-  - `workflow_deployment_saved`
-  - `workflow_run_started`
-  - `workflow_run_completed`
+  - [x] `workflow_artifact_saved`
+  - [x] `workflow_deployment_saved`
+  - [x] `workflow_run_started`
+  - [x] `workflow_run_completed`
   - `workflow_run_failed`
 - [ ] Add event kinds for capability changes:
   - `source_enabled`

src/wf_mcp/broker/events.py

@@ -1,33 +1,3 @@
-from __future__ import annotations
+from wf_mcp.events import McpEvent, make_event
 
-import time
-from dataclasses import dataclass, field
-from typing import Any
-
-
-@dataclass(slots=True)
-class McpEvent:
-    kind: str
-    timestamp_epoch_ms: int
-    connection_id: str | None = None
-    capability_id: str | None = None
-    workflow_name: str | None = None
-    payload: dict[str, Any] = field(default_factory=dict)
-
-
-def make_event(
-    kind: str,
-    *,
-    connection_id: str | None = None,
-    capability_id: str | None = None,
-    workflow_name: str | None = None,
-    payload: dict[str, Any] | None = None,
-) -> McpEvent:
-    return McpEvent(
-        kind=kind,
-        timestamp_epoch_ms=int(time.time() * 1000),
-        connection_id=connection_id,
-        capability_id=capability_id,
-        workflow_name=workflow_name,
-        payload=payload or {},
-    )
+__all__ = ["McpEvent", "make_event"]

src/wf_mcp/broker/service/core.py

@@ -18,6 +18,7 @@ from wf_artifacts import (
 from wf_core import NodeUse, Workflow, execute_workflow_async
 
 from ...connections import ConnectionRegistry, parse_connection_id, qualify_node_name
+from ...events import EventBus, McpEvent, make_event
 from ...models import (
     AuthRecord,
     CatalogNodeEntry,
@@ -34,7 +35,6 @@ from ...storage import Store
 from ...workflow.wrappers import _model_from_schema
 from ..catalog import CombinedCatalog, snapshot_from_specs
 from ..discovery import discover_connection_capabilities, specs_from_discovered_tools
-from ..events import McpEvent, make_event
 from ..admin_capabilities import admin_source
 from .adapters import require_adapter
 from .builtins import builtin_sources
@@ -61,7 +61,7 @@ class WfMcpService:
     connections: ConnectionRegistry = field(default_factory=ConnectionRegistry)
     adapters: dict[str, BackendAdapter] = field(default_factory=dict)
     capability_sources: dict[str, CapabilitySource] = field(default_factory=dict)
-    events: list[McpEvent] = field(default_factory=list)
+    event_bus: EventBus = field(default_factory=EventBus)
     include_builtin_specs: bool = True
     artifact_store: WorkflowArtifactStore | None = None
 
@@ -590,7 +590,7 @@ class WfMcpService:
         return run
 
     def list_events(self) -> list[McpEvent]:
-        return list(self.events)
+        return self.event_bus.list_events()
 
     def register_capability_source(self, source: CapabilitySource) -> None:
         """Register a capability source as canonical service state."""
@@ -669,4 +669,4 @@ class WfMcpService:
         return get_qualified_spec(self.capability_sources, qualified_name)
 
     def _record_event(self, event: McpEvent) -> None:
-        self.events.append(event)
+        self.event_bus.publish(event)

src/wf_mcp/events/__init__.py

@@ -0,0 +1,10 @@
+from .bus import EventBus, EventSubscriber, InMemoryEventSink
+from .models import McpEvent, make_event
+
+__all__ = [
+    "EventBus",
+    "EventSubscriber",
+    "InMemoryEventSink",
+    "McpEvent",
+    "make_event",
+]

src/wf_mcp/events/bus.py

@@ -0,0 +1,44 @@
+from __future__ import annotations
+
+from collections.abc import Callable
+
+from .models import McpEvent
+
+EventSubscriber = Callable[[McpEvent], None]
+
+
+class InMemoryEventSink:
+    """Append-only event sink used by current broker history APIs."""
+
+    def __init__(self) -> None:
+        self._events: list[McpEvent] = []
+
+    def __call__(self, event: McpEvent) -> None:
+        self._events.append(event)
+
+    def list_events(self) -> list[McpEvent]:
+        """Return a defensive copy of recorded events."""
+        return list(self._events)
+
+
+class EventBus:
+    """Synchronous in-process fanout for broker-local events.
+
+    This is intentionally not an MCP notification system. Protocol projections
+    can subscribe later without changing service code that emits events.
+    """
+
+    def __init__(self, history: InMemoryEventSink | None = None) -> None:
+        self._history = history or InMemoryEventSink()
+        self._subscribers: list[EventSubscriber] = [self._history]
+
+    def subscribe(self, subscriber: EventSubscriber) -> None:
+        self._subscribers.append(subscriber)
+
+    def publish(self, event: McpEvent) -> None:
+        for subscriber in self._subscribers:
+            subscriber(event)
+
+    def list_events(self) -> list[McpEvent]:
+        """Expose the default history sink for compatibility tools/resources."""
+        return self._history.list_events()

src/wf_mcp/events/models.py

@@ -0,0 +1,36 @@
+from __future__ import annotations
+
+import time
+from dataclasses import dataclass, field
+from typing import Any
+
+
+@dataclass(slots=True)
+class McpEvent:
+    """Broker-local event record before protocol-specific projection."""
+
+    kind: str
+    timestamp_epoch_ms: int
+    connection_id: str | None = None
+    capability_id: str | None = None
+    workflow_name: str | None = None
+    payload: dict[str, Any] = field(default_factory=dict)
+
+
+def make_event(
+    kind: str,
+    *,
+    connection_id: str | None = None,
+    capability_id: str | None = None,
+    workflow_name: str | None = None,
+    payload: dict[str, Any] | None = None,
+) -> McpEvent:
+    """Create a timestamped event with optional routing metadata."""
+    return McpEvent(
+        kind=kind,
+        timestamp_epoch_ms=int(time.time() * 1000),
+        connection_id=connection_id,
+        capability_id=capability_id,
+        workflow_name=workflow_name,
+        payload=payload or {},
+    )

src/wf_mcp/workflow_surface/handlers.py

@@ -15,6 +15,7 @@ from wf_artifacts import (
     validate_deployment_dependencies,
 )
 
+from ..events import make_event
 from ..models import RawWorkflowPlan
 
 if TYPE_CHECKING:
@@ -43,6 +44,16 @@ class WorkflowSurfaceHandlers:
             raise KeyError("workflow artifact store is not configured")
         workflow_artifact = WorkflowArtifact.model_validate(artifact)
         self.service.artifact_store.save_artifact(workflow_artifact)
+        self.service._record_event(
+            make_event(
+                "workflow_artifact_saved",
+                capability_id=_artifact_capability_id(workflow_artifact),
+                payload={
+                    "artifact_id": workflow_artifact.id,
+                    "version": workflow_artifact.version,
+                },
+            )
+        )
         return {
             "artifact_id": workflow_artifact.id,
             "version": workflow_artifact.version,
@@ -77,6 +88,17 @@ class WorkflowSurfaceHandlers:
             created_from_catalog_version=created_from_catalog_version,
         )
         self.service.artifact_store.save_artifact(workflow_artifact)
+        self.service._record_event(
+            make_event(
+                "workflow_artifact_saved",
+                capability_id=_artifact_capability_id(workflow_artifact),
+                payload={
+                    "artifact_id": workflow_artifact.id,
+                    "version": workflow_artifact.version,
+                    "created_from_plan": True,
+                },
+            )
+        )
         return {
             "artifact_id": workflow_artifact.id,
             "version": workflow_artifact.version,
@@ -106,6 +128,17 @@ class WorkflowSurfaceHandlers:
             raise KeyError("workflow artifact store is not configured")
         workflow_deployment = WorkflowDeployment.model_validate(deployment)
         self.service.artifact_store.save_deployment(workflow_deployment)
+        self.service._record_event(
+            make_event(
+                "workflow_deployment_saved",
+                capability_id=f"deployment.{workflow_deployment.id}",
+                payload={
+                    "deployment_id": workflow_deployment.id,
+                    "artifact_id": workflow_deployment.artifact_id,
+                    "artifact_version": workflow_deployment.artifact_version,
+                },
+            )
+        )
         return {
             "deployment_id": workflow_deployment.id,
             "artifact_id": workflow_deployment.artifact_id,
@@ -201,6 +234,11 @@ def _available_sources(service: WfMcpService) -> list[AvailableSource]:
     return sources
 
 
+def _artifact_capability_id(artifact: WorkflowArtifact) -> str:
+    """Use the same stable name shape as workflow artifact catalog entries."""
+    return f"workflow.{artifact.id}.v{artifact.version}"
+
+
 def _raw_plan_from_artifact(artifact: WorkflowArtifact) -> RawWorkflowPlan:
     """Validate the stored plan shape expected by the broker workflow runner."""
     return RawWorkflowPlan(

tests/wf_mcp/test_events.py

@@ -0,0 +1,37 @@
+from __future__ import annotations
+
+from wf_mcp.broker import WfMcpService
+from wf_mcp.events import EventBus, InMemoryEventSink, make_event
+from wf_mcp.models import ConnectionConfig
+from wf_mcp.storage import FileStore
+
+from .test_support import local_temp_root
+
+
+def test_event_bus_fans_out_to_subscribers() -> None:
+    sink = InMemoryEventSink()
+    seen_kinds: list[str] = []
+    bus = EventBus(sink)
+    bus.subscribe(lambda event: seen_kinds.append(event.kind))
+
+    bus.publish(make_event("catalog_changed", connection_id="demo.personal"))
+
+    assert [event.kind for event in sink.list_events()] == ["catalog_changed"]
+    assert sink.list_events()[0].connection_id == "demo.personal"
+    assert seen_kinds == ["catalog_changed"]
+
+
+def test_service_records_events_through_event_bus() -> None:
+    sink = InMemoryEventSink()
+    bus = EventBus(sink)
+    service = WfMcpService(
+        store=FileStore(local_temp_root() / "event_bus_service_store"),
+        event_bus=bus,
+    )
+
+    service.register_connection(
+        ConnectionConfig(id="demo.personal", server="demo", account="personal")
+    )
+
+    assert service.list_events()[0].kind == "connection_registered"
+    assert sink.list_events()[0] is service.list_events()[0]

tests/wf_mcp/test_workflow_surface.py

@@ -52,6 +52,35 @@ def test_workflow_surface_validates_deployment_dependencies() -> None:
     assert payload["diagnostics"][0]["code"] == "source_missing"
 
 
+def test_workflow_surface_records_artifact_and_deployment_save_events() -> None:
+    artifact_store = FileWorkflowArtifactStore(local_temp_root() / "surface_events")
+    handlers = _handlers(artifact_store)
+
+    artifact_payload = asyncio.run(
+        handlers.save_artifact(_echo_artifact().model_dump(mode="json"))
+    )
+    deployment_payload = asyncio.run(
+        handlers.save_deployment(
+            WorkflowDeployment(
+                id="echo.personal",
+                artifact_id="echo",
+                artifact_version=1,
+                bindings={"demo": "demo.personal"},
+            ).model_dump(mode="json")
+        )
+    )
+
+    events = handlers.service.list_events()
+    assert artifact_payload["saved"] is True
+    assert deployment_payload["saved"] is True
+    assert [event.kind for event in events] == [
+        "workflow_artifact_saved",
+        "workflow_deployment_saved",
+    ]
+    assert events[0].capability_id == "workflow.echo.v1"
+    assert events[1].capability_id == "deployment.echo.personal"
+
+
 def test_workflow_surface_runs_non_interrupting_deployment() -> None:
     artifact_store = FileWorkflowArtifactStore(local_temp_root() / "surface_run")
     artifact_store.save_artifact(_echo_artifact())