"""Home + meta endpoints + dedicated proof/audit visualizer pages.""" from __future__ import annotations import os from pyramid.response import FileResponse, Response from pyramid.view import view_config from .. import GIT_COMMIT, SERVICE_NAME HERE = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) def _shard_stats(request) -> dict: """Header pill counts. Single seam call into arborist.read.""" shards = request.shards if shards is None: return {"docs": 0, "audit_events": 0, "shard_open": False, "shard_count": 0} counts = shards.counts() return { "docs": counts.documents, "audit_events": counts.audit_events, "shard_open": True, "shard_count": counts.shard_count, } def _common_ctx(request) -> dict: return { "shards": request.arborist_shards, "shard_stats": _shard_stats(request), "git_commit": request.git_commit, "reveal_private": request.reveal_private_leaves, } @view_config(route_name="home", renderer="dashboard.jinja2") def home(request): return { "title": "Arborist VIZ — Merkle Command Center", "dashboard_id": "default", **_common_ctx(request), } @view_config(route_name="dashboard_page", renderer="dashboard.jinja2") def dashboard_page(request): return { "title": "Arborist VIZ — {0}".format(request.matchdict["dashboard_id"]), "dashboard_id": request.matchdict["dashboard_id"], **_common_ctx(request), } @view_config(route_name="proof_page") def proof_page(request): """Single-root visualizer. Dispatches by hash kind: - corpus document_root / leaf / merkle interior → proof.jinja2 (the 3D Merkle lattice + inclusion proof + leaf-list view) - session node_hash / subtree_hash → sessions_tree.jinja2 (the conversation-tree view, scrolled to the matching node) - providence cache_key → cache_key.jinja2 (every session node referencing this cache hit) All hashes drop through ONE URL — `/proof/{hash}` — per the ticket's "never mint new kind-named URLs" discipline. """ from pyramid.renderers import render_to_response root_hash = request.matchdict["root_hash"].strip().lower() leaf_param = request.GET.get("leaf") cache_key = request.GET.get("cache_key") # Sessions dispatch — check before falling through to the corpus # proof renderer. A session_root or session node_hash here means # the hash IS the answer; the corpus proof page can't help. sessions = request.sessions if sessions is not None and sessions.is_open: # session cache_key? show all referencing nodes. ck_hits = sessions.find_by_cache_key(root_hash) if ck_hits: ctx = _common_ctx(request) ctx["title"] = "Arborist VIZ — cache_key {0}".format( root_hash[:12] ) ctx["cache_key"] = root_hash ctx["hits"] = ck_hits ctx["sessions_stats"] = _sessions_stats_for_ctx(request) return render_to_response( "cache_key.jinja2", ctx, request=request ) # session node hash (node_hash or subtree_hash)? node = _find_session_node_by_hash(sessions, root_hash) if node is not None: return _render_session_in_proof( request, node.sid, focus_bates=node.bates, ) # session_root? (= subtree_hash of a session's root node) sid_for_root = _find_sid_by_session_root(sessions, root_hash) if sid_for_root is not None: return _render_session_in_proof(request, sid_for_root) # Default: corpus root. Hand off to proof.jinja2. ctx = { "title": "Arborist VIZ — root {0}".format(root_hash[:12]), "root_hash": root_hash, "leaf_index": leaf_param, "cache_key": cache_key, **_common_ctx(request), } return render_to_response("proof.jinja2", ctx, request=request) def _sessions_stats_for_ctx(request) -> dict: s = request.sessions if s is None or not s.is_open: return {"n_sessions": 0, "n_nodes": 0, "n_audit_events": 0, "open": False} c = s.counts() return { "n_sessions": c.n_sessions, "n_nodes": c.n_nodes, "n_audit_events": c.n_audit_events, "open": True, "path": c.shard_path, } def _find_session_node_by_hash(sessions, h: str): """Lookup by node_hash OR subtree_hash. Returns first match or None.""" try: r = sessions._conn.execute( # type: ignore[attr-defined] "SELECT * FROM nodes WHERE node_hash=? OR subtree_hash=? " "LIMIT 1", (h, h), ).fetchone() except Exception: return None if not r: return None from arborist.read import _row_to_session_node return _row_to_session_node(r) def _find_sid_by_session_root(sessions, h: str): """A session_root is the synthetic root node's subtree_hash — so look for a seq=0 node with that subtree_hash.""" try: r = sessions._conn.execute( # type: ignore[attr-defined] "SELECT sid FROM nodes WHERE seq=0 AND subtree_hash=? LIMIT 1", (h,), ).fetchone() except Exception: return None return r["sid"] if r else None def _render_session_in_proof(request, sid, focus_bates=None): """Render the session tree inside the unified /proof/{hash} dispatch. Re-uses sessions_tree.jinja2 so the visualization is consistent whether you arrived via /proof/{session_root}, /proof/{node_hash}, or directly via /sessions/{sid}.""" from pyramid.renderers import render_to_response sessions = request.sessions meta = sessions.get_session(sid) if meta is None: return _common_ctx(request) nodes = sessions.all_nodes_in_session(sid) children: dict[str, list] = {} for n in nodes: if n.parent_bates: children.setdefault(n.parent_bates, []).append(n) branches = sessions.branches_in_session(sid) ctx = _common_ctx(request) ctx["title"] = "Arborist VIZ — session {0}".format(sid) ctx["session"] = meta ctx["nodes"] = nodes ctx["children"] = children ctx["branches"] = set(branches) ctx["focus_bates"] = focus_bates ctx["sessions_stats"] = _sessions_stats_for_ctx(request) return render_to_response("sessions_tree.jinja2", ctx, request=request) @view_config(route_name="audit_page", renderer="audit.jinja2") def audit_page(request): event_hash = request.matchdict["event_hash"] return { "title": "Arborist VIZ — audit {0}".format(event_hash[:12]), "event_hash": event_hash, **_common_ctx(request), } @view_config(route_name="version") def version(request): return Response( json_body={ "service": SERVICE_NAME, "git_commit": GIT_COMMIT, "phase": "0-5+8", } ) @view_config(route_name="favicon") def favicon(request): path = os.path.join(HERE, "static", "favicon.ico") if os.path.exists(path): return FileResponse(path, request=request) return Response(status=404) @view_config(route_name="robots") def robots(request): return Response(body="User-agent: *\nDisallow:\n", content_type="text/plain")