fix(server): wire WS event bridge to the bus channels actually publish to (#673)

* fix(server): wire WS event bridge to the bus channels actually publish to

include_all_routes() built the /v1/agents/events WebSocket router from
get_event_bus() — a global singleton nothing in `jarvis serve` publishes
to. The real bus lives at app.state.bus (set in server/app.py and handed
to every channel/agent). Events published on it silently never reached
any connected browser client, since the WS router was subscribed to a
different, disconnected EventBus instance entirely.

* fix(server): keep managed agent events on app bus

---------

Co-authored-by: Ari <ari.silva@paipe.co>
Co-authored-by: Elliot Slusky <elliot@slusky.com>

src/openjarvis/server/agent_manager_routes.py

@@ -34,6 +34,15 @@ _MEMORY_BACKEND_LOCK_SETUP = threading.Lock()
 _MCP_LOCK_SETUP = threading.Lock()
 
 
+def _get_runtime_event_bus(runtime: Any = None) -> Any:
+    """Return the server-owned event bus, falling back outside app runtimes."""
+
+    from openjarvis.core.events import get_event_bus
+
+    bus = getattr(runtime, "bus", None)
+    return bus if bus is not None else get_event_bus()
+
+
 def _start_managed_worker(app_state: Any, target: Any, *, name: str) -> Any:
     """Start and track a managed-agent worker for orderly app shutdown."""
 
@@ -301,14 +310,13 @@ def _make_lightweight_system(
         # Wrap with InstrumentedEngine so agent ticks are recorded
         # in telemetry (FLOPs, energy, cost savings).
         try:
-            from openjarvis.core.events import get_event_bus
             from openjarvis.telemetry.instrumented_engine import (
                 InstrumentedEngine,
             )
 
             plain_engine = InstrumentedEngine(
                 plain_engine,
-                get_event_bus(),
+                _get_runtime_event_bus(runtime),
             )
         except Exception:
             pass  # telemetry is optional
@@ -1644,26 +1652,27 @@ def create_agent_manager_router(
 
         # Re-use the server's engine + model so we don't pick a
         # random model from Ollama's list.
-        server_engine = getattr(request.app.state, "engine", None)
-        server_model = getattr(request.app.state, "model", "")
-        server_config = getattr(request.app.state, "config", None)
+        app_state = request.app.state
+        server_engine = getattr(app_state, "engine", None)
+        server_model = getattr(app_state, "model", "")
+        server_config = getattr(app_state, "config", None)
+        server_bus = _get_runtime_event_bus(app_state)
 
         def _run_tick():
             try:
                 from openjarvis.agents.executor import AgentExecutor
-                from openjarvis.core.events import get_event_bus
 
-                _ts = getattr(request.app.state, "trace_store", None)
+                _ts = getattr(app_state, "trace_store", None)
                 executor = AgentExecutor(
                     manager=manager,
-                    event_bus=get_event_bus(),
+                    event_bus=server_bus,
                     trace_store=_ts,
                 )
                 system = _make_lightweight_system(
                     server_engine,
                     server_model,
                     server_config,
-                    request.app.state,
+                    app_state,
                 )
                 executor.set_system(system)
                 # The route handler above already called start_tick() to
@@ -1690,7 +1699,7 @@ def create_agent_manager_router(
 
         try:
             _start_managed_worker(
-                request.app.state,
+                app_state,
                 _run_tick,
                 name=f"managed-agent-run-{agent_id}",
             )
@@ -1997,11 +2006,12 @@ def create_agent_manager_router(
             import time as _time
 
             from openjarvis.agents.executor import AgentExecutor
-            from openjarvis.core.events import get_event_bus
 
-            _srv_engine = getattr(request.app.state, "engine", None)
-            _srv_model = getattr(request.app.state, "model", "")
-            _srv_config = getattr(request.app.state, "config", None)
+            _app_state = request.app.state
+            _srv_engine = getattr(_app_state, "engine", None)
+            _srv_model = getattr(_app_state, "model", "")
+            _srv_config = getattr(_app_state, "config", None)
+            _srv_bus = _get_runtime_event_bus(_app_state)
 
             def _immediate_tick():
                 _start = _time.time()
@@ -2011,17 +2021,17 @@ def create_agent_manager_router(
                     _srv_model,
                 )
                 try:
-                    _ts2 = getattr(request.app.state, "trace_store", None)
+                    _ts2 = getattr(_app_state, "trace_store", None)
                     executor = AgentExecutor(
                         manager=manager,
-                        event_bus=get_event_bus(),
+                        event_bus=_srv_bus,
                         trace_store=_ts2,
                     )
                     system = _make_lightweight_system(
                         _srv_engine,
                         _srv_model,
                         _srv_config,
-                        request.app.state,
+                        _app_state,
                     )
                     executor.set_system(system)
                     logger.info(
@@ -2055,7 +2065,7 @@ def create_agent_manager_router(
 
             try:
                 _start_managed_worker(
-                    request.app.state,
+                    _app_state,
                     _immediate_tick,
                     name=f"managed-agent-immediate-{agent_id}",
                 )
@@ -2106,12 +2116,12 @@ def create_agent_manager_router(
         return {"learning_log": manager.list_learning_log(agent_id)}
 
     @agents_router.post("/{agent_id}/learning/run")
-    def trigger_learning(agent_id: str):
+    def trigger_learning(agent_id: str, request: Request):
         if not manager.get_agent(agent_id):
             raise HTTPException(status_code=404, detail="Agent not found")
-        from openjarvis.core.events import EventType, get_event_bus
+        from openjarvis.core.events import EventType
 
-        bus = get_event_bus()
+        bus = _get_runtime_event_bus(request.app.state)
         bus.publish(EventType.AGENT_LEARNING_STARTED, {"agent_id": agent_id})
         return {"status": "triggered"}
 

src/openjarvis/server/api_routes.py

@@ -1087,12 +1087,16 @@ def include_all_routes(app) -> None:
     except ImportError:
         pass
 
-    # WebSocket bridge for real-time agent events
+    # WebSocket bridge for real-time agent events. Must subscribe on the
+    # same EventBus instance channels/agents actually publish to
+    # (app.state.bus, set in server/app.py) — the get_event_bus() global
+    # singleton is a *different* bus that nothing in `jarvis serve` ever
+    # publishes to, so events silently never reached this endpoint.
     try:
         from openjarvis.core.events import get_event_bus
         from openjarvis.server.ws_bridge import create_ws_router
 
-        ws_router = create_ws_router(get_event_bus())
+        ws_router = create_ws_router(getattr(app.state, "bus", None) or get_event_bus())
         app.include_router(ws_router)
     except Exception:
         logger.debug("WebSocket bridge not available", exc_info=True)

tests/server/test_agent_manager_routes.py

@@ -605,6 +605,39 @@ class TestLightweightSystemEngineResolution:
         )
         assert captured["key"] == "llamacpp"
 
+    def test_instrumented_engine_uses_runtime_event_bus(self, monkeypatch):
+        pytest.importorskip("fastapi")
+        from openjarvis.core.events import EventBus
+        from openjarvis.server import agent_manager_routes as amr
+        from openjarvis.telemetry import instrumented_engine
+
+        resolved_engine = MagicMock()
+        wrapped_engine = MagicMock()
+        runtime_bus = EventBus()
+        runtime = SimpleNamespace(
+            bus=runtime_bus,
+            memory_backend=object(),
+            channel_backend=None,
+            channel_bridge=None,
+            knowledge_db_path=None,
+        )
+        monkeypatch.setattr(
+            "openjarvis.engine._discovery.get_engine",
+            MagicMock(return_value=("resolved", resolved_engine)),
+        )
+        instrumented = MagicMock(return_value=wrapped_engine)
+        monkeypatch.setattr(instrumented_engine, "InstrumentedEngine", instrumented)
+
+        system = amr._make_lightweight_system(
+            engine=MagicMock(),
+            model="m",
+            config=self._cfg("vllm", "ollama"),
+            runtime=runtime,
+        )
+
+        instrumented.assert_called_once_with(resolved_engine, runtime_bus)
+        assert system.engine is wrapped_engine
+
     def test_caches_tool_memory_backend_when_prompt_context_is_disabled(
         self,
         monkeypatch,

tests/server/test_ws_bridge.py

@@ -3,8 +3,10 @@
 from __future__ import annotations
 
 import asyncio
+import threading
 import time
 from types import SimpleNamespace
+from unittest.mock import MagicMock, patch
 
 import pytest
 
@@ -161,3 +163,91 @@ class TestWSBridge:
             ]
 
         asyncio.run(exercise())
+
+
+class TestIncludeAllRoutesBusWiring:
+    """Regression: the WS endpoint must subscribe on the same EventBus that
+    channels/agents actually publish to (app.state.bus), not the unrelated
+    get_event_bus() global singleton — publishing on the latter used to
+    silently never reach any connected browser client."""
+
+    def test_uses_app_state_bus_not_global_singleton(self):
+        from openjarvis.core.events import reset_event_bus
+        from openjarvis.server.api_routes import include_all_routes
+
+        reset_event_bus()  # isolate from other tests' global singleton state
+        app = FastAPI()
+        real_bus = EventBus()
+        app.state.bus = real_bus
+        include_all_routes(app)
+
+        client = TestClient(app)
+        with client.websocket_connect("/v1/agents/events") as ws:
+            real_bus.publish(EventType.AGENT_TICK_START, {"agent_id": "test-123"})
+            time.sleep(0.05)
+            data = ws.receive_json()
+            assert data["data"]["agent_id"] == "test-123"
+
+    @pytest.mark.parametrize(
+        ("path", "payload"),
+        [
+            ("/v1/managed-agents/test-123/run", None),
+            (
+                "/v1/managed-agents/test-123/messages",
+                {"content": "run now", "mode": "immediate", "stream": False},
+            ),
+        ],
+    )
+    def test_managed_agent_run_paths_publish_to_app_bus(self, path, payload):
+        from openjarvis.agents.executor import AgentExecutor
+        from openjarvis.core.events import reset_event_bus
+        from openjarvis.server.api_routes import include_all_routes
+
+        reset_event_bus()
+        app_bus = EventBus()
+        manager = MagicMock()
+        manager.get_agent.return_value = {
+            "id": "test-123",
+            "name": "test",
+            "status": "idle",
+            "config": {},
+        }
+        manager.send_message.return_value = {
+            "id": "message-123",
+            "agent_id": "test-123",
+            "content": "run now",
+            "mode": "immediate",
+        }
+
+        app = FastAPI()
+        app.state.bus = app_bus
+        app.state.agent_manager = manager
+        include_all_routes(app)
+
+        executed = threading.Event()
+        observed_buses = []
+
+        def publish_tick(executor, agent_id, **_kwargs):
+            observed_buses.append(executor._bus)
+            executor._bus.publish(EventType.AGENT_TICK_START, {"agent_id": agent_id})
+            executed.set()
+
+        client = TestClient(app)
+        with (
+            patch.object(AgentExecutor, "execute_tick", publish_tick),
+            patch(
+                "openjarvis.server.agent_manager_routes._make_lightweight_system",
+                return_value=MagicMock(),
+            ) as make_system,
+            client.websocket_connect("/v1/agents/events?agent_id=test-123") as ws,
+        ):
+            request_kwargs = {"json": payload} if payload is not None else {}
+            response = client.post(path, **request_kwargs)
+            assert response.status_code == 200
+            assert executed.wait(timeout=1)
+            assert observed_buses == [app_bus]
+            data = ws.receive_json()
+
+        assert data["type"] == "agent_tick_start"
+        assert data["data"]["agent_id"] == "test-123"
+        make_system.assert_called_once()