Offload blocking SessionDB calls off event loop

Fixes on record

Merged to main. A stamp counts the issues a fix is linked to.

• #65893 fix(dashboard): offload blocking SessionDB handlers (supersedes #63082)

  merged to main

  files: hermes_cli/web_server.py, tests/test_web_server_sessiondb_eventloop.py

  Show the diff

  • 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
     
     
    [redacted:email]("/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()
     
     
    [redacted:email]("/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
     
     
    [redacted:email]("/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()
     
     
    [redacted:email]("/api/analytics/models")
    -async def get_models_analytics(days: int = 30, profile: Optional[str] = None):
    [redacted:email]("/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()
     
     
    [redacted:email]("/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.
     #

  • 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)

• #117041 Dashboard session search/stats/detail no longer block the event loop on a large state.db (#60747)

  merged to main

  files: hermes_cli/web_routers/sessions.py, tests/hermes_cli/test_web_server_sessiondb_eventloop.py