arborist-viz/arborist_viz/views/sessions.py

"""Sessions — Merkle conversation forest dashboard.

Read-only view of the sessions shard (one file per user, at
``~/.arborist/sessions.db`` by default; overridable via env or .ini).
Three surfaces:

  - /sessions                  list all sessions
  - /sessions/{sid}            one session's tree
  - /sessions/find?q=...       FTS5 search across every session
  - /api/sessions              JSON list
  - /api/sessions/{sid}        JSON tree
  - /api/sessions/find?q=...   JSON search

All schema knowledge lives behind :mod:`arborist.read.SessionsView`.
"""
from __future__ import annotations

from pyramid.httpexceptions import HTTPNotFound
from pyramid.view import view_config

from ..views.home import _common_ctx


def _sessions_ctx(request) -> dict:
    """Common template context with sessions stats."""
    sessions = request.sessions
    if sessions is None or not sessions.is_open:
        s_stats = {"n_sessions": 0, "n_nodes": 0,
                   "n_audit_events": 0, "open": False}
    else:
        c = sessions.counts()
        s_stats = {
            "n_sessions": c.n_sessions,
            "n_nodes": c.n_nodes,
            "n_audit_events": c.n_audit_events,
            "open": True,
            "path": c.shard_path,
        }
    ctx = _common_ctx(request)
    ctx["sessions_stats"] = s_stats
    return ctx


@view_config(route_name="sessions_list", renderer="sessions_list.jinja2")
def sessions_list(request):
    sessions = request.sessions
    rows = sessions.list_sessions() if sessions else []
    ctx = _sessions_ctx(request)
    ctx["title"] = "Sessions — Arborist VIZ"
    ctx["rows"] = rows
    return ctx


@view_config(route_name="sessions_tree", renderer="sessions_tree.jinja2")
def sessions_tree(request):
    sessions = request.sessions
    if sessions is None:
        raise HTTPNotFound("no sessions shard configured")
    sid = request.matchdict["sid"]
    meta = sessions.get_session(sid)
    if meta is None:
        raise HTTPNotFound(f"no such session: {sid}")
    nodes = sessions.all_nodes_in_session(sid)
    # Materialize a children-map per Bates so the template can render
    # the tree without N round-trips into SessionsView.
    children: dict[str, list] = {}
    for n in nodes:
        if n.parent_bates:
            children.setdefault(n.parent_bates, []).append(n)
    # Also include cross-session children whose parent_bates points
    # into this session.
    for parent_bates in list(children.keys()):
        pass  # already populated by the loop above
    branches = sessions.branches_in_session(sid)
    ctx = _sessions_ctx(request)
    ctx["title"] = f"session {sid} — Arborist VIZ"
    ctx["session"] = meta
    ctx["nodes"] = nodes
    ctx["children"] = children
    ctx["branches"] = set(branches)
    return ctx


@view_config(route_name="sessions_find", renderer="sessions_find.jinja2")
def sessions_find(request):
    sessions = request.sessions
    q = (request.params.get("q") or "").strip()
    try:
        limit = max(1, min(100, int(request.params.get("limit") or "20")))
    except ValueError:
        limit = 20
    hits = sessions.find(q, limit=limit) if (sessions and q) else []
    ctx = _sessions_ctx(request)
    ctx["title"] = f"search — Arborist VIZ" + (f" — {q}" if q else "")
    ctx["query"] = q
    ctx["limit"] = limit
    ctx["hits"] = hits
    return ctx


# ---- JSON APIs ------------------------------------------------------


def _node_to_json(n) -> dict:
    return {
        "bates": n.bates, "sid": n.sid, "seq": n.seq,
        "parent_bates": n.parent_bates,
        "question": n.question,
        "answer_text": n.answer_text,
        "cache_key": n.cache_key,
        "audit_mode": n.audit_mode,
        "cited_titles": n.cited_titles,
        "n_cited_sources": n.n_cited_sources,
        "node_hash": n.node_hash,
        "subtree_hash": n.subtree_hash,
        "created_at": n.created_at,
        "label": n.label,
    }


@view_config(route_name="api_sessions_list", renderer="json")
def api_sessions_list(request):
    sessions = request.sessions
    if sessions is None:
        return {"sessions": [], "open": False}
    return {
        "sessions": [
            {
                "sid": s.sid, "root_bates": s.root_bates,
                "current_bates": s.current_bates,
                "label": s.label, "n_nodes": s.n_nodes,
                "session_root": s.session_root,
                "created_at": s.created_at,
                "updated_at": s.updated_at,
            }
            for s in sessions.list_sessions()
        ],
        "open": True,
    }


@view_config(route_name="api_sessions_tree", renderer="json")
def api_sessions_tree(request):
    sessions = request.sessions
    if sessions is None:
        raise HTTPNotFound("no sessions shard")
    sid = request.matchdict["sid"]
    meta = sessions.get_session(sid)
    if meta is None:
        raise HTTPNotFound(f"no such session: {sid}")
    nodes = sessions.all_nodes_in_session(sid)
    return {
        "sid": sid,
        "session_root": meta.session_root,
        "current_bates": meta.current_bates,
        "branches": sessions.branches_in_session(sid),
        "nodes": [_node_to_json(n) for n in nodes],
    }


@view_config(route_name="api_sessions_find", renderer="json")
def api_sessions_find(request):
    sessions = request.sessions
    q = (request.params.get("q") or "").strip()
    try:
        limit = max(1, min(100, int(request.params.get("limit") or "20")))
    except ValueError:
        limit = 20
    hits = sessions.find(q, limit=limit) if (sessions and q) else []
    return {
        "query": q, "limit": limit,
        "hits": [_node_to_json(n) for n in hits],
    }