From 4b47c5d435dbfb584f0a801052fb57f1821b7452 Mon Sep 17 00:00:00 2001 From: Robby Manihani Date: Fri, 3 Apr 2026 22:00:14 -0700 Subject: [PATCH] fix: wire TraceStore to event bus and pass to AgentExecutor (#187) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Closes #179 — traces were never persisted in the serve path because: 1. TraceStore.subscribe_to_bus() was never called in app.py 2. AgentExecutor was constructed without trace_store in serve.py and agent_manager_routes.py 3. The /v1/traces API returned raw dataclass fields instead of the frontend-expected format Fixes all three: wires bus subscription, passes trace_store to all executor instantiation sites, and adds _serialise_trace() for proper API response formatting. --- src/openjarvis/cli/serve.py | 15 +++++++++- src/openjarvis/server/agent_manager_routes.py | 4 +++ src/openjarvis/server/api_routes.py | 30 +++++++++++++++---- src/openjarvis/server/app.py | 6 +++- 4 files changed, 47 insertions(+), 8 deletions(-) diff --git a/src/openjarvis/cli/serve.py b/src/openjarvis/cli/serve.py index 998a0c68..3bda0864 100644 --- a/src/openjarvis/cli/serve.py +++ b/src/openjarvis/cli/serve.py @@ -344,7 +344,20 @@ def serve( from openjarvis.agents.executor import AgentExecutor from openjarvis.agents.scheduler import AgentScheduler - executor = AgentExecutor(manager=agent_manager, event_bus=bus) + _trace_store = None + try: + if config.traces.enabled: + from openjarvis.traces.store import TraceStore + + _trace_store = TraceStore(db_path=config.traces.db_path) + except Exception: + pass + + executor = AgentExecutor( + manager=agent_manager, + event_bus=bus, + trace_store=_trace_store, + ) from openjarvis.system import SystemBuilder system = SystemBuilder(config).build() diff --git a/src/openjarvis/server/agent_manager_routes.py b/src/openjarvis/server/agent_manager_routes.py index 5d5ac666..6bba86f5 100644 --- a/src/openjarvis/server/agent_manager_routes.py +++ b/src/openjarvis/server/agent_manager_routes.py @@ -1156,9 +1156,11 @@ def create_agent_manager_router( from openjarvis.agents.executor import AgentExecutor from openjarvis.core.events import get_event_bus + _ts = getattr(request.app.state, "trace_store", None) executor = AgentExecutor( manager=manager, event_bus=get_event_bus(), + trace_store=_ts, ) system = _make_lightweight_system( server_engine, @@ -1501,9 +1503,11 @@ def create_agent_manager_router( _srv_model, ) try: + _ts2 = getattr(request.app.state, "trace_store", None) executor = AgentExecutor( manager=manager, event_bus=get_event_bus(), + trace_store=_ts2, ) system = _make_lightweight_system( _srv_engine, diff --git a/src/openjarvis/server/api_routes.py b/src/openjarvis/server/api_routes.py index e2b95ff0..def4a804 100644 --- a/src/openjarvis/server/api_routes.py +++ b/src/openjarvis/server/api_routes.py @@ -196,17 +196,37 @@ async def memory_stats(request: Request): traces_router = APIRouter(prefix="/v1/traces", tags=["traces"]) +def _serialise_trace(trace) -> dict: + """Convert a Trace dataclass to a frontend-friendly dict.""" + import datetime + from dataclasses import asdict + + d = asdict(trace) + d["id"] = d.pop("trace_id", "") + started = d.pop("started_at", 0.0) + d["created_at"] = ( + datetime.datetime.fromtimestamp(started, tz=datetime.timezone.utc).isoformat() + if started + else None + ) + dur = d.pop("total_latency_seconds", 0.0) + d["duration_ms"] = round(dur * 1000) + for step in d.get("steps", []): + st = step.get("step_type") + if hasattr(st, "value"): + step["step_type"] = st.value + return d + + @traces_router.get("") async def list_traces(request: Request, limit: int = 20): """List recent traces.""" try: - from dataclasses import asdict - store = getattr(request.app.state, "trace_store", None) if store is None: return {"traces": []} traces = store.list_traces(limit=limit) - items = [asdict(t) for t in traces] + items = [_serialise_trace(t) for t in traces] return {"traces": items} except Exception as exc: return {"traces": [], "error": str(exc)} @@ -216,15 +236,13 @@ async def list_traces(request: Request, limit: int = 20): async def get_trace(trace_id: str, request: Request): """Get a specific trace by ID.""" try: - from dataclasses import asdict - store = getattr(request.app.state, "trace_store", None) if store is None: raise HTTPException(status_code=404, detail="Trace not found") trace = store.get(trace_id) if trace is None: raise HTTPException(status_code=404, detail="Trace not found") - return asdict(trace) + return _serialise_trace(trace) except HTTPException: raise except Exception as exc: diff --git a/src/openjarvis/server/app.py b/src/openjarvis/server/app.py index 26d9c6f4..423cdbfd 100644 --- a/src/openjarvis/server/app.py +++ b/src/openjarvis/server/app.py @@ -215,7 +215,11 @@ def create_app( cfg = config if config is not None else load_config() if cfg.traces.enabled: - app.state.trace_store = TraceStore(db_path=cfg.traces.db_path) + _trace_store = TraceStore(db_path=cfg.traces.db_path) + app.state.trace_store = _trace_store + _bus = getattr(app.state, "bus", None) + if _bus is not None: + _trace_store.subscribe_to_bus(_bus) except Exception: pass # traces are optional; don't block server startup