diff --git a/hermes_cli/web_server.py b/hermes_cli/web_server.py index 176cb82e4f6f..b17dab4cb6b5 100644 --- a/hermes_cli/web_server.py +++ b/hermes_cli/web_server.py @@ -9936,12 +9936,15 @@ async def bulk_delete_sessions_endpoint(body: BulkDeleteSessions): status_code=400, detail="ids must contain at most 500 entries", ) - db = _open_session_db_for_profile(body.profile) - try: - deleted = db.delete_sessions(body.ids) - return {"ok": True, "deleted": deleted} - finally: - db.close() + def _delete() -> int: + db = _open_session_db_for_profile(body.profile) + try: + return db.delete_sessions(body.ids) + finally: + db.close() + + deleted = await asyncio.to_thread(_delete) + return {"ok": True, "deleted": deleted} @app.post("/api/sessions/import") @@ -9978,11 +9981,14 @@ async def count_empty_sessions_endpoint(profile: Optional[str] = None): UI hides the affordance so users aren't presented with a button that does nothing. Cheap, single-COUNT query. """ - db = _open_session_db_for_profile(profile) - try: - return {"count": db.count_empty_sessions()} - finally: - db.close() + def _count() -> int: + db = _open_session_db_for_profile(profile) + try: + return db.count_empty_sessions() + finally: + db.close() + + return {"count": await asyncio.to_thread(_count)} @app.delete("/api/sessions/empty") @@ -10005,12 +10011,15 @@ async def delete_empty_sessions_endpoint(profile: Optional[str] = None): prune-on-startup pass. Matching that pre-existing trade-off keeps the two delete endpoints' DB-vs-disk behaviour consistent. """ - db = _open_session_db_for_profile(profile) - try: - deleted = db.delete_empty_sessions() - return {"ok": True, "deleted": deleted} - finally: - db.close() + def _delete() -> int: + db = _open_session_db_for_profile(profile) + try: + return db.delete_empty_sessions() + finally: + db.close() + + deleted = await asyncio.to_thread(_delete) + return {"ok": True, "deleted": deleted} @app.get("/api/sessions/stats") @@ -10080,19 +10089,22 @@ async def get_session_latest_descendant( session_id: str, profile: Optional[str] = None, ): - db = _open_session_db_for_profile(profile) - try: - latest, path = _session_latest_descendant(session_id, db) - if not latest: - raise HTTPException(status_code=404, detail="Session not found") - return { - "requested_session_id": path[0] if path else session_id, - "session_id": latest, - "path": path, - "changed": bool(path and latest != path[0]), - } - finally: - db.close() + def _lookup(): + db = _open_session_db_for_profile(profile) + try: + return _session_latest_descendant(session_id, db) + finally: + db.close() + + latest, path = await asyncio.to_thread(_lookup) + if not latest: + raise HTTPException(status_code=404, detail="Session not found") + return { + "requested_session_id": path[0] if path else session_id, + "session_id": latest, + "path": path, + "changed": bool(path and latest != path[0]), + } @app.get("/api/sessions/{session_id}/messages") async def get_session_messages( @@ -10101,26 +10113,32 @@ async def get_session_messages( limit: Optional[int] = None, offset: int = 0, ): - db = _open_session_db_for_profile(profile) - try: - sid = db.resolve_session_id(session_id) - if not sid: - raise HTTPException(status_code=404, detail="Session not found") - sid = db.resolve_resume_session_id(sid) - # Clamp limit to prevent abuse (max 500 per page) - _limit = min(limit, 500) if limit is not None else None - messages = db.get_messages(sid, limit=_limit, offset=offset) - return { - "session_id": sid, - "messages": messages, - "pagination": { - "limit": _limit, - "offset": offset, - "returned": len(messages), - }, - } - finally: - db.close() + def _read(): + db = _open_session_db_for_profile(profile) + try: + sid = db.resolve_session_id(session_id) + if not sid: + return None + sid = db.resolve_resume_session_id(sid) + # Clamp limit to prevent abuse (max 500 per page) + _limit = min(limit, 500) if limit is not None else None + return sid, _limit, db.get_messages(sid, limit=_limit, offset=offset) + finally: + db.close() + + result = await asyncio.to_thread(_read) + if result is None: + raise HTTPException(status_code=404, detail="Session not found") + sid, _limit, messages = result + return { + "session_id": sid, + "messages": messages, + "pagination": { + "limit": _limit, + "offset": offset, + "returned": len(messages), + }, + } @app.delete("/api/sessions/{session_id}") @@ -10128,24 +10146,27 @@ async def delete_session_endpoint(session_id: str, profile: Optional[str] = None # ``profile`` deletes a session belonging to another (local) profile by # opening its state.db directly. Remote profiles never reach here — the # desktop routes their DELETE to the remote backend. Omit for current/default. - db = _open_session_db_for_profile(profile) - try: - # Resolve exact ids / unique prefixes like every other session endpoint - # (detail, messages, rename, export all do). A session that no longer - # exists is an idempotent success: DELETE's contract is "ensure it's - # gone", and the desktop optimistically removes the row then RESTORES it - # on any error — so a 404 on an already-absent row resurrected a ghost - # row and surfaced "session not found". /goal + auto-compression churn - # leaves transient empty rows (reaped by empty-session hygiene) that - # race the sidebar snapshot, which is exactly when this fired. Mirrors - # the bulk-delete endpoint, which already treats ghost ids as success. - sid = db.resolve_session_id(session_id) - if not sid: - return {"ok": True, "already_absent": True} - db.delete_session(sid) - return {"ok": True} - finally: - db.close() + def _delete(): + db = _open_session_db_for_profile(profile) + try: + # Resolve exact ids / unique prefixes like every other session endpoint + # (detail, messages, rename, export all do). A session that no longer + # exists is an idempotent success: DELETE's contract is "ensure it's + # gone", and the desktop optimistically removes the row then RESTORES it + # on any error — so a 404 on an already-absent row resurrected a ghost + # row and surfaced "session not found". /goal + auto-compression churn + # leaves transient empty rows (reaped by empty-session hygiene) that + # race the sidebar snapshot, which is exactly when this fired. Mirrors + # the bulk-delete endpoint, which already treats ghost ids as success. + sid = db.resolve_session_id(session_id) + if not sid: + return {"ok": True, "already_absent": True} + db.delete_session(sid) + return {"ok": True} + finally: + db.close() + + return await asyncio.to_thread(_delete) class SessionRename(BaseModel): @@ -10193,17 +10214,18 @@ async def rename_session_endpoint(session_id: str, body: SessionRename): @app.get("/api/sessions/{session_id}/export") async def export_session_endpoint(session_id: str, profile: Optional[str] = None): """Export a single session (metadata + messages) as JSON.""" - db = _open_session_db_for_profile(profile) - try: - sid = db.resolve_session_id(session_id) - if not sid: - raise HTTPException(status_code=404, detail="Session not found") - data = db.export_session(sid) - if data is None: - raise HTTPException(status_code=404, detail="Session not found") - return data - finally: - db.close() + def _export(): + db = _open_session_db_for_profile(profile) + try: + sid = db.resolve_session_id(session_id) + return db.export_session(sid) if sid else None + finally: + db.close() + + data = await asyncio.to_thread(_export) + if data is None: + raise HTTPException(status_code=404, detail="Session not found") + return data class SessionPrune(BaseModel): @@ -10234,8 +10256,7 @@ class SessionPrune(BaseModel): dry_run: bool = False -@app.post("/api/sessions/prune") -async def prune_sessions_endpoint(body: SessionPrune): +def _prune_sessions(body: SessionPrune): """Delete ended sessions matching filters (mirrors `hermes sessions prune`).""" has_window = ( body.started_before is not None or body.started_after is not None @@ -10317,6 +10338,12 @@ async def prune_sessions_endpoint(body: SessionPrune): db.close() +@app.post("/api/sessions/prune") +async def prune_sessions_endpoint(body: SessionPrune): + """Delete ended sessions matching filters without blocking the event loop.""" + return await asyncio.to_thread(_prune_sessions, body) + + # --------------------------------------------------------------------------- # Log viewer endpoint # --------------------------------------------------------------------------- @@ -14481,8 +14508,7 @@ def _aux_task_summary(aux_rows: List[Dict[str, Any]]) -> List[Dict[str, Any]]: return result -@app.get("/api/analytics/usage") -async def get_usage_analytics(days: int = 30, profile: Optional[str] = None): +def _get_usage_analytics(days: int = 30, profile: Optional[str] = None): from agent.insights import InsightsEngine db = _open_session_db_for_profile(profile) @@ -14563,8 +14589,12 @@ async def get_usage_analytics(days: int = 30, profile: Optional[str] = None): db.close() -@app.get("/api/analytics/models") -async def get_models_analytics(days: int = 30, profile: Optional[str] = None): +@app.get("/api/analytics/usage") +async def get_usage_analytics(days: int = 30, profile: Optional[str] = None): + return await asyncio.to_thread(_get_usage_analytics, days, profile) + + +def _get_models_analytics(days: int = 30, profile: Optional[str] = None): """Rich per-model analytics for the Models dashboard page. Returns token/cost/session breakdown per model plus capability metadata @@ -14740,6 +14770,12 @@ async def get_models_analytics(days: int = 30, profile: Optional[str] = None): db.close() +@app.get("/api/analytics/models") +async def get_models_analytics(days: int = 30, profile: Optional[str] = None): + """Return model analytics without blocking the serving event loop.""" + return await asyncio.to_thread(_get_models_analytics, days, profile) + + # --------------------------------------------------------------------------- # /api/pty — PTY-over-WebSocket bridge for the dashboard "Chat" tab. # diff --git a/tests/test_web_server_sessiondb_eventloop.py b/tests/test_web_server_sessiondb_eventloop.py new file mode 100644 index 000000000000..7b91af5a8887 --- /dev/null +++ b/tests/test_web_server_sessiondb_eventloop.py @@ -0,0 +1,96 @@ +import ast +import asyncio +import threading +from pathlib import Path + +from hermes_cli import web_server + + +TARGET_HANDLERS = { + "bulk_delete_sessions_endpoint", + "count_empty_sessions_endpoint", + "delete_empty_sessions_endpoint", + "get_session_latest_descendant", + "get_session_messages", + "delete_session_endpoint", + "export_session_endpoint", + "prune_sessions_endpoint", + "get_usage_analytics", + "get_models_analytics", +} + + +def _call_name(call: ast.Call) -> str | None: + if isinstance(call.func, ast.Name): + return call.func.id + if isinstance(call.func, ast.Attribute): + return call.func.attr + return None + + +def test_sessiondb_handlers_open_connections_inside_executor_helpers(): + tree = ast.parse(Path(web_server.__file__).read_text(encoding="utf-8")) + handlers = { + node.name: node + for node in tree.body + if isinstance(node, ast.AsyncFunctionDef) and node.name in TARGET_HANDLERS + } + top_level_helpers = { + node.name: node for node in tree.body if isinstance(node, ast.FunctionDef) + } + assert handlers.keys() == TARGET_HANDLERS + + for name, handler in handlers.items(): + helpers = { + **top_level_helpers, + **{ + node.name: node + for node in handler.body + if isinstance(node, ast.FunctionDef) + }, + } + offloaded = { + arg.id + for node in ast.walk(handler) + if isinstance(node, ast.Call) + and _call_name(node) == "to_thread" + for arg in node.args[:1] + if isinstance(arg, ast.Name) + } + db_open_owners = { + helper_name + for helper_name, helper in helpers.items() + if helper_name in offloaded + and any( + isinstance(node, ast.Call) + and _call_name(node) == "_open_session_db_for_profile" + for node in ast.walk(helper) + ) + } + assert db_open_owners, f"{name} does not offload SessionDB open + work" + + +def test_bulk_delete_sessiondb_work_runs_off_event_loop(monkeypatch): + loop_thread = threading.get_ident() + db_threads: list[int] = [] + + class _DB: + def delete_sessions(self, ids): + db_threads.append(threading.get_ident()) + assert ids == ["one", "two"] + return 2 + + def close(self): + db_threads.append(threading.get_ident()) + + monkeypatch.setattr(web_server, "_open_session_db_for_profile", lambda profile=None: _DB()) + + result = asyncio.run( + web_server.bulk_delete_sessions_endpoint( + web_server.BulkDeleteSessions(ids=["one", "two"]) + ) + ) + + assert result == {"ok": True, "deleted": 2} + assert db_threads + assert all(thread_id != loop_thread for thread_id in db_threads)