From 04a2b2790201b66c3a54c8a80f23b8cc39e63e95 Mon Sep 17 00:00:00 2001 From: "russell@unturf.com" Date: Mon, 1 Jun 2026 17:41:04 -0400 Subject: [PATCH] arborist.read: add SessionsView read seam for the sessions shard MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Mirrors `open_shards(paths) → Shards`: arborist-viz and other read-only consumers import `open_sessions(path) → SessionsView`, exposing: - counts() → SessionCounts(n_sessions, n_nodes, n_audit_events) - list_sessions() → list[SessionRow] - get_session(sid), get_node(bates), children_of, path_to_root - all_nodes_in_session(sid), branches_in_session(sid) - find(query, limit) — FTS5 over question+answer+cited_titles, skips synthetic-root nodes (seq=0) - find_by_cache_key(cache_key) - chain_check() → (intact, breaks) Opens sqlite with mode=ro + check_same_thread=False for WSGI worker pools, serialized internally by an RLock. Bootstrap via SessionStore.open().close() ensures the schema exists before the read-only handle attaches. 9 tests pass: counts, list, get_node, path_to_root crossing sessions, find FTS5, find-skips-synth-root, chain integrity, missing-db fallback, branches in session, find_by_cache_key. arborist-viz commit lands separately. --- arborist/read.py | 862 +++++++++++++++++++++++++----------- tests/test_read_sessions.py | 113 +++++ 2 files changed, 718 insertions(+), 257 deletions(-) create mode 100644 tests/test_read_sessions.py diff --git a/arborist/read.py b/arborist/read.py index fbf69e6..b10f9fa 100644 --- a/arborist/read.py +++ b/arborist/read.py @@ -59,6 +59,7 @@ from __future__ import annotations import json import logging import sqlite3 +import threading from collections.abc import Iterable from dataclasses import dataclass, field from pathlib import Path @@ -83,6 +84,13 @@ __all__ = [ "ContextRoot", "ResolveResult", "ShardCounts", + # Sessions (Merkle conversation forest) — separate file/shard from + # corpus shards. See arborist/qa/session.py for the producer. + "open_sessions", + "SessionsView", + "SessionRow", + "SessionNode", + "SessionCounts", ] @@ -268,6 +276,19 @@ class Shards: def __init__(self, paths: list[str]): self._paths: list[str] = list(paths) self._conns: dict[str, sqlite3.Connection] = {} + # sqlite3 connections allow shared *use* across threads + # (check_same_thread=False) but NOT concurrent use; a second + # thread calling execute()/fetchone() on a connection already + # in flight raises SQLITE_MISUSE or returns a stale cursor + # (fetchone() → None on a COUNT(*) statement). Multi-worker + # WSGI servers (waitress) + SSE poll loops + page renders all + # hit the same Shards handle from different threads, so + # serialize. A single Shards-wide RLock is simpler than per- + # shard locks and the workload (dev viz, sub-ms reads) doesn't + # measurably benefit from cross-shard parallelism. RLock so a + # method that already holds the lock can call a helper that + # also acquires it (e.g. proof() → _owning_conn_for_root()). + self._lock = threading.RLock() for p in self._paths: try: # Run arborist's migration probe via the canonical connect() @@ -289,69 +310,73 @@ class Shards: return list(self._conns.keys()) def close(self) -> None: - for c in self._conns.values(): - try: - c.close() - except Exception: - pass - self._conns.clear() + with self._lock: + for c in self._conns.values(): + try: + c.close() + except Exception: + pass + self._conns.clear() def counts(self) -> ShardCounts: docs = events = qa = 0 - for c in self._conns.values(): - try: - docs += int(c.execute("SELECT COUNT(*) FROM documents").fetchone()[0]) - except sqlite3.Error: - pass - try: - events += int( - c.execute("SELECT COALESCE(MAX(seq), 0) FROM audit_events").fetchone()[0] - ) - except sqlite3.Error: - pass - try: - qa += int(c.execute("SELECT COUNT(*) FROM providence_cache").fetchone()[0]) - except sqlite3.Error: - pass - return ShardCounts(documents=docs, audit_events=events, qa_records=qa, shard_count=len(self._conns)) + with self._lock: + for c in self._conns.values(): + try: + docs += int(c.execute("SELECT COUNT(*) FROM documents").fetchone()[0]) + except sqlite3.Error: + pass + try: + events += int( + c.execute("SELECT COALESCE(MAX(seq), 0) FROM audit_events").fetchone()[0] + ) + except sqlite3.Error: + pass + try: + qa += int(c.execute("SELECT COUNT(*) FROM providence_cache").fetchone()[0]) + except sqlite3.Error: + pass + return ShardCounts(documents=docs, audit_events=events, qa_records=qa, shard_count=len(self._conns)) # ---------- documents / roots -------------------------------------- def root(self, document_root: str) -> Optional[Root]: - for path, c in self._conns.items(): - r = c.execute( - "SELECT document_root, document_uri, source_type, kind, title, " - " chunking_version, canonicalization_version, schema_version, " - " ingest_ts, hit_count " - "FROM documents WHERE document_root = ?", - (document_root,), - ).fetchone() - if r is None: - continue - leaf_count = c.execute( - "SELECT COUNT(*) FROM chunks WHERE document_root = ?", - (document_root,), - ).fetchone()[0] - return Root(leaf_count=int(leaf_count), shard_path=path, **dict(r)) - return None + with self._lock: + for path, c in self._conns.items(): + r = c.execute( + "SELECT document_root, document_uri, source_type, kind, title, " + " chunking_version, canonicalization_version, schema_version, " + " ingest_ts, hit_count " + "FROM documents WHERE document_root = ?", + (document_root,), + ).fetchone() + if r is None: + continue + leaf_count = c.execute( + "SELECT COUNT(*) FROM chunks WHERE document_root = ?", + (document_root,), + ).fetchone()[0] + return Root(leaf_count=int(leaf_count), shard_path=path, **dict(r)) + return None def roots(self, limit: int = 50) -> list[Root]: merged: list[Root] = [] - for path, c in self._conns.items(): - rows = c.execute( - "SELECT document_root, document_uri, source_type, kind, title, " - " chunking_version, canonicalization_version, schema_version, " - " ingest_ts, hit_count " - "FROM documents ORDER BY ingest_ts DESC LIMIT ?", - (limit,), - ).fetchall() - for r in rows: - d = dict(r) - lc = c.execute( - "SELECT COUNT(*) FROM chunks WHERE document_root = ?", - (d["document_root"],), - ).fetchone()[0] - merged.append(Root(leaf_count=int(lc), shard_path=path, **d)) + with self._lock: + for path, c in self._conns.items(): + rows = c.execute( + "SELECT document_root, document_uri, source_type, kind, title, " + " chunking_version, canonicalization_version, schema_version, " + " ingest_ts, hit_count " + "FROM documents ORDER BY ingest_ts DESC LIMIT ?", + (limit,), + ).fetchall() + for r in rows: + d = dict(r) + lc = c.execute( + "SELECT COUNT(*) FROM chunks WHERE document_root = ?", + (d["document_root"],), + ).fetchone()[0] + merged.append(Root(leaf_count=int(lc), shard_path=path, **d)) merged.sort(key=lambda r: r.ingest_ts, reverse=True) return merged[:limit] @@ -373,15 +398,15 @@ class Shards: Returns an empty list when the root is unknown. """ - conn = self._owning_conn_for_root(document_root) - if conn is None: - return [] - - rows = conn.execute( - "SELECT document_root, idx, leaf_hash, content, tier " - "FROM chunks WHERE document_root = ? ORDER BY idx", - (document_root,), - ).fetchall() + with self._lock: + conn = self._owning_conn_for_root(document_root) + if conn is None: + return [] + rows = conn.execute( + "SELECT document_root, idx, leaf_hash, content, tier " + "FROM chunks WHERE document_root = ? ORDER BY idx", + (document_root,), + ).fetchall() to_base, base_version = _get_wikitext_projector() if project else (None, None) @@ -420,14 +445,15 @@ class Shards: """Compute a Merkle inclusion proof against the shard that owns the root. Returns ``None`` if the root is unknown; raises ``IndexError`` if ``leaf_index`` is out of range.""" - conn = self._owning_conn_for_root(document_root) - path = self._path_for_root(document_root) - if conn is None or path is None: - return None - rows = conn.execute( - "SELECT idx, leaf_hash FROM chunks WHERE document_root = ? ORDER BY idx", - (document_root,), - ).fetchall() + with self._lock: + conn = self._owning_conn_for_root(document_root) + path = self._path_for_root(document_root) + if conn is None or path is None: + return None + rows = conn.execute( + "SELECT idx, leaf_hash FROM chunks WHERE document_root = ? ORDER BY idx", + (document_root,), + ).fetchall() if not rows: return None if leaf_index < 0 or leaf_index >= len(rows): @@ -457,13 +483,14 @@ class Shards: def tree_layers(self, document_root: str) -> Optional[list[list[str]]]: """All Merkle tree layers as hex (layer 0 = leaves, last = root). Drives the 3D lattice visualization.""" - conn = self._owning_conn_for_root(document_root) - if conn is None: - return None - rows = conn.execute( - "SELECT leaf_hash FROM chunks WHERE document_root = ? ORDER BY idx", - (document_root,), - ).fetchall() + with self._lock: + conn = self._owning_conn_for_root(document_root) + if conn is None: + return None + rows = conn.execute( + "SELECT leaf_hash FROM chunks WHERE document_root = ? ORDER BY idx", + (document_root,), + ).fetchall() if not rows: return None leaves = [bytes.fromhex(r["leaf_hash"]) for r in rows] @@ -473,15 +500,16 @@ class Shards: # ---------- audit -------------------------------------------------- def audit_event(self, event_hash: str) -> Optional[AuditEvent]: - for path, c in self._conns.items(): - r = c.execute( - "SELECT seq, event_hash, prev_event_hash, event_type, subject_root, body, ts " - "FROM audit_events WHERE event_hash = ?", - (event_hash,), - ).fetchone() - if r is not None: - return _row_to_audit(r, path) - return None + with self._lock: + for path, c in self._conns.items(): + r = c.execute( + "SELECT seq, event_hash, prev_event_hash, event_type, subject_root, body, ts " + "FROM audit_events WHERE event_hash = ?", + (event_hash,), + ).fetchone() + if r is not None: + return _row_to_audit(r, path) + return None def audit_chain(self, head_event_hash: str, *, limit: int = 100) -> list[AuditEvent]: out: list[AuditEvent] = [] @@ -498,25 +526,27 @@ class Shards: def audit_recent(self, *, limit: int = 100) -> list[AuditEvent]: merged: list[AuditEvent] = [] - for path, c in self._conns.items(): - for r in c.execute( - "SELECT seq, event_hash, prev_event_hash, event_type, subject_root, body, ts " - "FROM audit_events ORDER BY seq DESC LIMIT ?", - (limit,), - ).fetchall(): - merged.append(_row_to_audit(r, path)) + with self._lock: + for path, c in self._conns.items(): + for r in c.execute( + "SELECT seq, event_hash, prev_event_hash, event_type, subject_root, body, ts " + "FROM audit_events ORDER BY seq DESC LIMIT ?", + (limit,), + ).fetchall(): + merged.append(_row_to_audit(r, path)) merged.sort(key=lambda e: e.ts, reverse=True) return merged[:limit] def audit_by_root(self, root: str, *, limit: int = 50) -> list[AuditEvent]: merged: list[AuditEvent] = [] - for path, c in self._conns.items(): - for r in c.execute( - "SELECT seq, event_hash, prev_event_hash, event_type, subject_root, body, ts " - "FROM audit_events WHERE subject_root = ? ORDER BY seq DESC LIMIT ?", - (root, limit), - ).fetchall(): - merged.append(_row_to_audit(r, path)) + with self._lock: + for path, c in self._conns.items(): + for r in c.execute( + "SELECT seq, event_hash, prev_event_hash, event_type, subject_root, body, ts " + "FROM audit_events WHERE subject_root = ? ORDER BY seq DESC LIMIT ?", + (root, limit), + ).fetchall(): + merged.append(_row_to_audit(r, path)) merged.sort(key=lambda e: e.seq, reverse=True) return merged[:limit] @@ -533,25 +563,27 @@ class Shards: """ cursor = dict(cursor or {}) new_events: list[AuditEvent] = [] - for path, c in self._conns.items(): - last_seq = int(cursor.get(path, 0)) - for r in c.execute( - "SELECT seq, event_hash, prev_event_hash, event_type, subject_root, body, ts " - "FROM audit_events WHERE seq > ? ORDER BY seq ASC LIMIT ?", - (last_seq, limit), - ).fetchall(): - ev = _row_to_audit(r, path) - new_events.append(ev) - cursor[path] = max(int(cursor.get(path, 0)), ev.seq) + with self._lock: + for path, c in self._conns.items(): + last_seq = int(cursor.get(path, 0)) + for r in c.execute( + "SELECT seq, event_hash, prev_event_hash, event_type, subject_root, body, ts " + "FROM audit_events WHERE seq > ? ORDER BY seq ASC LIMIT ?", + (last_seq, limit), + ).fetchall(): + ev = _row_to_audit(r, path) + new_events.append(ev) + cursor[path] = max(int(cursor.get(path, 0)), ev.seq) return new_events, cursor def audit_cursor(self) -> dict: """Cursor at the current head of every shard (use as starting point for a tail that only wants future events).""" - return { - path: int(c.execute("SELECT COALESCE(MAX(seq), 0) FROM audit_events").fetchone()[0]) - for path, c in self._conns.items() - } + with self._lock: + return { + path: int(c.execute("SELECT COALESCE(MAX(seq), 0) FROM audit_events").fetchone()[0]) + for path, c in self._conns.items() + } # ---------- providence_cache (Q&A) --------------------------------- @@ -569,42 +601,43 @@ class Shards: single ingested document. Consumers don't need to know which case applies; they just chain the calls. """ - for path, c in self._conns.items(): - try: - r = c.execute( - _QA_SELECT + " WHERE source_root = ? LIMIT 1", - (context_root,), - ).fetchone() - except sqlite3.Error: - continue - if r is None: - continue - qa = _row_to_qa(r, path) - sources = qa.merkle_proof.get("sources", []) if isinstance(qa.merkle_proof, dict) else [] - primary_uri, domains = _summarize_sources(sources) - # Headline the real source. The stored document_uri for a - # multi-source answer is the opaque ``corpus://multi-source`` - # sentinel — replace it with the primary answer source so the - # node says where the knowledge came from (e.g. - # russell.ballestrini.net), not a placeholder. - headline_uri = qa.document_uri or "" - if (not headline_uri or headline_uri == _MULTI_SOURCE_URI) and primary_uri: - headline_uri = primary_uri - return ContextRoot( - context_root=context_root, - cache_key=qa.cache_key, - question_text=qa.question_text or "", - document_uri=headline_uri, - sources=sources, - chunking_version=qa.chunking_version, - canonicalization_version=qa.canonicalization_version, - schema_version=qa.schema_version, - created_at=qa.created_at, - hit_count=qa.hit_count, - shard_path=path, - primary_source_uri=primary_uri, - source_domains=domains, - ) + with self._lock: + for path, c in self._conns.items(): + try: + r = c.execute( + _QA_SELECT + " WHERE source_root = ? LIMIT 1", + (context_root,), + ).fetchone() + except sqlite3.Error: + continue + if r is None: + continue + qa = _row_to_qa(r, path) + sources = qa.merkle_proof.get("sources", []) if isinstance(qa.merkle_proof, dict) else [] + primary_uri, domains = _summarize_sources(sources) + # Headline the real source. The stored document_uri for a + # multi-source answer is the opaque ``corpus://multi-source`` + # sentinel — replace it with the primary answer source so the + # node says where the knowledge came from (e.g. + # russell.ballestrini.net), not a placeholder. + headline_uri = qa.document_uri or "" + if (not headline_uri or headline_uri == _MULTI_SOURCE_URI) and primary_uri: + headline_uri = primary_uri + return ContextRoot( + context_root=context_root, + cache_key=qa.cache_key, + question_text=qa.question_text or "", + document_uri=headline_uri, + sources=sources, + chunking_version=qa.chunking_version, + canonicalization_version=qa.canonicalization_version, + schema_version=qa.schema_version, + created_at=qa.created_at, + hit_count=qa.hit_count, + shard_path=path, + primary_source_uri=primary_uri, + source_domains=domains, + ) return None def context_layers(self, context_root: str) -> Optional[list[list[str]]]: @@ -666,32 +699,35 @@ class Shards: ) def qa(self, cache_key: str) -> Optional[QaRecord]: - for path, c in self._conns.items(): - r = c.execute( - _QA_SELECT + " WHERE cache_key = ?", (cache_key,) - ).fetchone() - if r is not None: - return _row_to_qa(r, path) - return None + with self._lock: + for path, c in self._conns.items(): + r = c.execute( + _QA_SELECT + " WHERE cache_key = ?", (cache_key,) + ).fetchone() + if r is not None: + return _row_to_qa(r, path) + return None def qa_recent(self, *, limit: int = 50) -> list[QaRecord]: merged: list[QaRecord] = [] - for path, c in self._conns.items(): - for r in c.execute( - _QA_SELECT + " ORDER BY created_at DESC LIMIT ?", (limit,) - ).fetchall(): - merged.append(_row_to_qa(r, path)) + with self._lock: + for path, c in self._conns.items(): + for r in c.execute( + _QA_SELECT + " ORDER BY created_at DESC LIMIT ?", (limit,) + ).fetchall(): + merged.append(_row_to_qa(r, path)) merged.sort(key=lambda q: q.created_at, reverse=True) return merged[:limit] def qa_by_root(self, source_root: str, *, limit: int = 50) -> list[QaRecord]: merged: list[QaRecord] = [] - for path, c in self._conns.items(): - for r in c.execute( - _QA_SELECT + " WHERE source_root = ? ORDER BY created_at DESC LIMIT ?", - (source_root, limit), - ).fetchall(): - merged.append(_row_to_qa(r, path)) + with self._lock: + for path, c in self._conns.items(): + for r in c.execute( + _QA_SELECT + " WHERE source_root = ? ORDER BY created_at DESC LIMIT ?", + (source_root, limit), + ).fetchall(): + merged.append(_row_to_qa(r, path)) merged.sort(key=lambda q: q.created_at, reverse=True) return merged[:limit] @@ -705,13 +741,14 @@ class Shards: """ pat = "%{0}%".format(text.replace("%", r"\%").replace("_", r"\_")) merged: list[QaRecord] = [] - for path, c in self._conns.items(): - for r in c.execute( - _QA_SELECT + " WHERE question_text LIKE ? ESCAPE '\\' " - "ORDER BY created_at DESC LIMIT ?", - (pat, limit), - ).fetchall(): - merged.append(_row_to_qa(r, path)) + with self._lock: + for path, c in self._conns.items(): + for r in c.execute( + _QA_SELECT + " WHERE question_text LIKE ? ESCAPE '\\' " + "ORDER BY created_at DESC LIMIT ?", + (pat, limit), + ).fetchall(): + merged.append(_row_to_qa(r, path)) merged.sort(key=lambda q: q.created_at, reverse=True) return merged[:limit] @@ -726,104 +763,107 @@ class Shards: """ h = hex_hash.lower().strip() - for path, c in self._conns.items(): - if c.execute("SELECT 1 FROM documents WHERE document_root = ? LIMIT 1", (h,)).fetchone(): - return ResolveResult(kind="document_root", hash=h, shard_path=path) + with self._lock: + for path, c in self._conns.items(): + if c.execute("SELECT 1 FROM documents WHERE document_root = ? LIMIT 1", (h,)).fetchone(): + return ResolveResult(kind="document_root", hash=h, shard_path=path) - for path, c in self._conns.items(): - r = c.execute("SELECT document_root, idx FROM chunks WHERE leaf_hash = ? LIMIT 1", (h,)).fetchone() - if r: - return ResolveResult( - kind="leaf_hash", hash=h, shard_path=path, - extra={"document_root": r["document_root"], "leaf_index": int(r["idx"])}, - ) + for path, c in self._conns.items(): + r = c.execute("SELECT document_root, idx FROM chunks WHERE leaf_hash = ? LIMIT 1", (h,)).fetchone() + if r: + return ResolveResult( + kind="leaf_hash", hash=h, shard_path=path, + extra={"document_root": r["document_root"], "leaf_index": int(r["idx"])}, + ) - for path, c in self._conns.items(): - r = c.execute("SELECT subject_root FROM audit_events WHERE event_hash = ? LIMIT 1", (h,)).fetchone() - if r: - return ResolveResult( - kind="audit_event", hash=h, shard_path=path, - extra={"subject_root": r["subject_root"]}, - ) + for path, c in self._conns.items(): + r = c.execute("SELECT subject_root FROM audit_events WHERE event_hash = ? LIMIT 1", (h,)).fetchone() + if r: + return ResolveResult( + kind="audit_event", hash=h, shard_path=path, + extra={"subject_root": r["subject_root"]}, + ) - for path, c in self._conns.items(): - try: + for path, c in self._conns.items(): + try: + r = c.execute( + "SELECT source_root FROM providence_cache WHERE cache_key = ? LIMIT 1", (h,) + ).fetchone() + if r: + return ResolveResult( + kind="qa_cache_key", hash=h, shard_path=path, + extra={"source_root": r["source_root"]}, + ) + except sqlite3.Error: + continue + + # providence_cache.source_root: the QA's evidence root. Reaches + # this branch only when the hash isn't in ``documents`` (the + # documents probe runs first and returns "document_root"), so + # any match here is a synthesized multi-source context. + for path, c in self._conns.items(): + try: + r = c.execute( + "SELECT cache_key FROM providence_cache WHERE source_root = ? LIMIT 1", (h,) + ).fetchone() + if r: + return ResolveResult( + kind="context_root", hash=h, shard_path=path, + extra={"cache_key": r["cache_key"]}, + ) + except sqlite3.Error: + continue + + for path, c in self._conns.items(): + try: + r = c.execute( + "SELECT cache_key FROM providence_cache WHERE run_dag_root = ? LIMIT 1", (h,) + ).fetchone() + if r: + return ResolveResult( + kind="qa_run_dag_root", hash=h, shard_path=path, + extra={"cache_key": r["cache_key"]}, + ) + except sqlite3.Error: + continue + + for path, c in self._conns.items(): r = c.execute( - "SELECT source_root FROM providence_cache WHERE cache_key = ? LIMIT 1", (h,) + "SELECT document_root, layer, idx FROM merkle_nodes WHERE hash = ? LIMIT 1", (h,) ).fetchone() if r: return ResolveResult( - kind="qa_cache_key", hash=h, shard_path=path, - extra={"source_root": r["source_root"]}, + kind="merkle_interior", hash=h, shard_path=path, + extra={ + "document_root": r["document_root"], + "layer": int(r["layer"]), + "idx": int(r["idx"]), + }, ) - except sqlite3.Error: - continue - - # providence_cache.source_root: the QA's evidence root. Reaches - # this branch only when the hash isn't in ``documents`` (the - # documents probe runs first and returns "document_root"), so - # any match here is a synthesized multi-source context. - for path, c in self._conns.items(): - try: - r = c.execute( - "SELECT cache_key FROM providence_cache WHERE source_root = ? LIMIT 1", (h,) - ).fetchone() - if r: - return ResolveResult( - kind="context_root", hash=h, shard_path=path, - extra={"cache_key": r["cache_key"]}, - ) - except sqlite3.Error: - continue - - for path, c in self._conns.items(): - try: - r = c.execute( - "SELECT cache_key FROM providence_cache WHERE run_dag_root = ? LIMIT 1", (h,) - ).fetchone() - if r: - return ResolveResult( - kind="qa_run_dag_root", hash=h, shard_path=path, - extra={"cache_key": r["cache_key"]}, - ) - except sqlite3.Error: - continue - - for path, c in self._conns.items(): - r = c.execute( - "SELECT document_root, layer, idx FROM merkle_nodes WHERE hash = ? LIMIT 1", (h,) - ).fetchone() - if r: - return ResolveResult( - kind="merkle_interior", hash=h, shard_path=path, - extra={ - "document_root": r["document_root"], - "layer": int(r["layer"]), - "idx": int(r["idx"]), - }, - ) return ResolveResult(kind="unknown", hash=h) # ---------- internals ---------------------------------------------- def _owning_conn_for_root(self, document_root: str) -> Optional[sqlite3.Connection]: - for _path, c in self._conns.items(): - if c.execute( - "SELECT 1 FROM documents WHERE document_root = ? LIMIT 1", - (document_root,), - ).fetchone(): - return c - return None + with self._lock: + for _path, c in self._conns.items(): + if c.execute( + "SELECT 1 FROM documents WHERE document_root = ? LIMIT 1", + (document_root,), + ).fetchone(): + return c + return None def _path_for_root(self, document_root: str) -> Optional[str]: - for path, c in self._conns.items(): - if c.execute( - "SELECT 1 FROM documents WHERE document_root = ? LIMIT 1", - (document_root,), - ).fetchone(): - return path - return None + with self._lock: + for path, c in self._conns.items(): + if c.execute( + "SELECT 1 FROM documents WHERE document_root = ? LIMIT 1", + (document_root,), + ).fetchone(): + return path + return None def open_shards(paths: Iterable[str | Path]) -> Shards: @@ -963,3 +1003,311 @@ def _get_wikitext_projector(): return to_base, BASE_VERSION except Exception: return None, None + + +# ==================================================================== +# Sessions — read-only view over the conversation-forest shard. +# ==================================================================== +# +# Schema producer: arborist/qa/session.py (SessionStore). One shard at +# ~/.arborist/sessions.db per user. Three tables: +# - sessions (per-sid metadata) +# - nodes (every turn ever; FTS5 index nodes_fts) +# - audit_events (one global hash chain over the shard's mutations) +# +# This read seam is what arborist-viz (and any other read-only +# consumer) imports. Mutations stay on the SessionStore producer; the +# connection here is opened mode=ro. + + +@dataclass(frozen=True) +class SessionRow: + sid: str + root_bates: str + current_bates: str + label: Optional[str] + n_nodes: int + session_root: str + created_at: str + updated_at: str + + +@dataclass(frozen=True) +class SessionNode: + """One conversation turn. ``cited_titles`` is the parsed pointer-line + title list materialized at insert time — what the model actually + cited, not the wider retrieval pool.""" + + bates: str + sid: str + seq: int + parent_bates: Optional[str] + question: str + answer_text: str + cache_key: str + audit_mode: str + cited_titles: list[str] + n_cited_sources: int + created_at: str + label: Optional[str] + node_hash: str + subtree_hash: str + + +@dataclass(frozen=True) +class SessionCounts: + n_sessions: int + n_nodes: int + n_audit_events: int + shard_path: str + + +def _row_to_session_node(r: sqlite3.Row) -> SessionNode: + try: + titles = json.loads(r["cited_titles_json"] or "[]") + if not isinstance(titles, list): + titles = [] + except Exception: + titles = [] + return SessionNode( + bates=r["bates"], sid=r["sid"], seq=r["seq"], + parent_bates=r["parent_bates"], + question=r["question"], answer_text=r["answer_text"], + cache_key=r["cache_key"], audit_mode=r["audit_mode"], + cited_titles=titles, n_cited_sources=r["n_cited_sources"], + created_at=r["created_at"], label=r["label"], + node_hash=r["node_hash"], subtree_hash=r["subtree_hash"], + ) + + +class SessionsView: + """Read-only handle over the single sessions shard. Opens + ``mode=ro`` with ``check_same_thread=False`` so a WSGI worker pool + (waitress, gunicorn) can share one handle. Serialized internally + by an RLock — sqlite3 connections are share-but-not-concurrent. + """ + + def __init__(self, path: str): + self._path = path + self._lock = threading.RLock() + # Run arborist's session schema bootstrap (CREATE IF NOT EXISTS) + # by opening with the producer once and closing — keeps the + # read seam's connection truly read-only. + try: + from arborist.qa.session import SessionStore + SessionStore.open(Path(path)).close() + except Exception as exc: + log.warning("arborist.read.sessions: bootstrap failed for %s: %s", + path, exc) + self._conn: Optional[sqlite3.Connection] = None + try: + self._conn = sqlite3.connect( + "file:{0}?mode=ro".format(path), uri=True, + check_same_thread=False, + ) + self._conn.row_factory = sqlite3.Row + except Exception as exc: + log.warning("arborist.read.sessions: open failed for %s: %s", + path, exc) + + @property + def path(self) -> str: + return self._path + + @property + def is_open(self) -> bool: + return self._conn is not None + + def close(self) -> None: + with self._lock: + if self._conn is not None: + try: + self._conn.close() + except Exception: + pass + self._conn = None + + # ---- meta ---- + + def counts(self) -> SessionCounts: + if self._conn is None: + return SessionCounts(0, 0, 0, self._path) + with self._lock: + n_s = self._conn.execute( + "SELECT COUNT(*) AS c FROM sessions" + ).fetchone()["c"] + n_n = self._conn.execute( + "SELECT COUNT(*) AS c FROM nodes WHERE seq > 0" + ).fetchone()["c"] + n_a = self._conn.execute( + "SELECT COUNT(*) AS c FROM audit_events" + ).fetchone()["c"] + return SessionCounts(n_s, n_n, n_a, self._path) + + # ---- sessions ---- + + def list_sessions(self) -> list[SessionRow]: + if self._conn is None: + return [] + out: list[SessionRow] = [] + with self._lock: + for r in self._conn.execute( + "SELECT s.sid, s.root_bates, s.current_bates, s.label, " + " s.created_at, s.updated_at, " + " (SELECT COUNT(*) FROM nodes n " + " WHERE n.sid=s.sid AND n.seq>0) AS n_nodes, " + " (SELECT n2.subtree_hash FROM nodes n2 " + " WHERE n2.bates=s.root_bates) AS session_root " + "FROM sessions s ORDER BY s.updated_at DESC" + ): + out.append(SessionRow( + sid=r["sid"], root_bates=r["root_bates"], + current_bates=r["current_bates"], label=r["label"], + n_nodes=r["n_nodes"], + session_root=r["session_root"] or "", + created_at=r["created_at"], updated_at=r["updated_at"], + )) + return out + + def get_session(self, sid: str) -> Optional[SessionRow]: + for s in self.list_sessions(): + if s.sid == sid: + return s + return None + + # ---- nodes ---- + + def get_node(self, bates: str) -> Optional[SessionNode]: + if self._conn is None: + return None + with self._lock: + r = self._conn.execute( + "SELECT * FROM nodes WHERE bates=?", (bates,) + ).fetchone() + return _row_to_session_node(r) if r else None + + def children_of(self, bates: str) -> list[SessionNode]: + if self._conn is None: + return [] + with self._lock: + rows = list(self._conn.execute( + "SELECT * FROM nodes WHERE parent_bates=? " + "ORDER BY created_at, seq", + (bates,), + )) + return [_row_to_session_node(r) for r in rows] + + def path_to_root(self, bates: str) -> list[SessionNode]: + if self._conn is None: + return [] + out: list[SessionNode] = [] + seen: set[str] = set() + cur: Optional[str] = bates + while cur is not None and cur not in seen: + seen.add(cur) + n = self.get_node(cur) + if n is None: + break + out.append(n) + cur = n.parent_bates + return out + + def all_nodes_in_session(self, sid: str) -> list[SessionNode]: + if self._conn is None: + return [] + with self._lock: + rows = list(self._conn.execute( + "SELECT * FROM nodes WHERE sid=? ORDER BY seq", (sid,), + )) + return [_row_to_session_node(r) for r in rows] + + def branches_in_session(self, sid: str) -> list[str]: + """Bates owned by ``sid`` that have ≥2 children (cross-session + children count toward the branch-point status).""" + if self._conn is None: + return [] + with self._lock: + return [ + r["parent_bates"] + for r in self._conn.execute( + "SELECT n.parent_bates FROM nodes n " + "JOIN nodes p ON p.bates = n.parent_bates " + "WHERE p.sid = ? " + "GROUP BY n.parent_bates " + "HAVING COUNT(*) >= 2 " + "ORDER BY MIN(n.seq)", + (sid,), + ) + ] + + # ---- search ---- + + def find(self, query: str, limit: int = 10) -> list[SessionNode]: + """FTS5 over question + answer_text + cited_titles. Skips + synthetic-root nodes (seq=0).""" + if self._conn is None: + return [] + try: + from arborist.wallet.bucket import _to_fts5 + match_expr = _to_fts5(query) + except Exception: + match_expr = query + if not match_expr.strip(): + return [] + with self._lock: + rows = list(self._conn.execute( + "SELECT nodes.*, bm25(nodes_fts) AS score " + "FROM nodes_fts " + "JOIN nodes ON nodes.rowid = nodes_fts.rowid " + "WHERE nodes_fts MATCH ? AND nodes.seq > 0 " + "ORDER BY score ASC LIMIT ?", + (match_expr, limit), + )) + return [_row_to_session_node(r) for r in rows] + + def find_by_cache_key(self, cache_key: str) -> list[SessionNode]: + if self._conn is None: + return [] + with self._lock: + rows = list(self._conn.execute( + "SELECT * FROM nodes WHERE cache_key=? ORDER BY created_at", + (cache_key,), + )) + return [_row_to_session_node(r) for r in rows] + + # ---- audit chain ---- + + def chain_check(self) -> tuple[int, int]: + """Walk audit_events in order, recompute event_hash, compare. + Returns ``(intact_count, break_count)``.""" + if self._conn is None: + return (0, 0) + import hashlib + intact = 0 + breaks = 0 + prev_hex = "" + with self._lock: + for r in self._conn.execute( + "SELECT body, event_hash FROM audit_events ORDER BY seq" + ): + body_bytes = r["body"].encode("utf-8", errors="surrogatepass") + h = hashlib.sha256() + if prev_hex: + h.update(bytes.fromhex(prev_hex)) + h.update(body_bytes) + expected = h.hexdigest() + if expected == r["event_hash"]: + intact += 1 + else: + breaks += 1 + prev_hex = r["event_hash"] + return (intact, breaks) + + +def open_sessions(path: str | Path) -> SessionsView: + """Open the sessions shard read-only. ``path`` defaults to + ``~/.arborist/sessions.db`` when None or empty.""" + if not path: + from arborist.qa.session import default_db_path + path = str(default_db_path()) + return SessionsView(str(path)) diff --git a/tests/test_read_sessions.py b/tests/test_read_sessions.py new file mode 100644 index 0000000..653f427 --- /dev/null +++ b/tests/test_read_sessions.py @@ -0,0 +1,113 @@ +"""Tests for arborist.read.open_sessions / SessionsView.""" +from __future__ import annotations + +from pathlib import Path + +import pytest + +from arborist.qa.session import SessionStore +from arborist.read import open_sessions + + +@pytest.fixture +def sessions_db(tmp_path: Path) -> Path: + db = tmp_path / "sessions.db" + store = SessionStore.open(db) + a = store.create_session(sid="A") + a.add_node("who is spider man?", "ck1", "STRICT", + answer_text="Spider-Man is from Marvel.\n " + '[E1 | Spider-Man | abc123: "..."]') + a.add_node("who created him?", "ck2", "STRICT", + answer_text="Stan Lee.\n " + '[E1 | Spider-Man | abc123: "..."]') + b = store.create_session(sid="B") + b.add_node("how do you cook risotto?", "ck3", "HYBRID", + answer_text="Stir constantly.") + store.close() + return db + + +def test_counts(sessions_db: Path): + v = open_sessions(sessions_db) + c = v.counts() + assert c.n_sessions == 2 + assert c.n_nodes == 3 # synthetic roots excluded + assert c.n_audit_events >= 5 # 2 init + 3 added + v.close() + + +def test_list_sessions(sessions_db: Path): + v = open_sessions(sessions_db) + sids = {s.sid for s in v.list_sessions()} + assert sids == {"A", "B"} + a = v.get_session("A") + assert a is not None + assert a.n_nodes == 2 + assert a.session_root # non-empty hex + v.close() + + +def test_get_node_and_path_to_root(sessions_db: Path): + v = open_sessions(sessions_db) + n = v.get_node("A-000002") + assert n is not None + assert n.question == "who created him?" + assert n.cited_titles == ["Spider-Man"] + path = v.path_to_root("A-000002") + assert [p.bates for p in path] == ["A-000002", "A-000001", "A-000000"] + v.close() + + +def test_find_fts5_across_sessions(sessions_db: Path): + v = open_sessions(sessions_db) + hits = v.find("spider") + assert {h.sid for h in hits} == {"A"} + hits = v.find("risotto") + assert {h.sid for h in hits} == {"B"} + v.close() + + +def test_find_skips_synthetic_root(sessions_db: Path): + v = open_sessions(sessions_db) + hits = v.find("anything that wouldn't match") + assert hits == [] + # Even targeting empty content the synth root never surfaces. + v.close() + + +def test_chain_check_intact(sessions_db: Path): + v = open_sessions(sessions_db) + intact, breaks = v.chain_check() + assert breaks == 0 + assert intact >= 5 + v.close() + + +def test_missing_db_returns_empty_view(tmp_path: Path): + """Open a path that doesn't exist yet — view is created (the + bootstrap inside SessionsView ensures the schema), but everything + reads back empty rather than erroring.""" + db = tmp_path / "fresh.db" + v = open_sessions(db) + assert v.is_open + assert v.counts().n_sessions == 0 + assert v.list_sessions() == [] + assert v.find("anything") == [] + assert v.chain_check() == (0, 0) + v.close() + + +def test_branches_in_session(sessions_db: Path): + """Sanity: with two single-line sessions and no forks, branches=[].""" + v = open_sessions(sessions_db) + assert v.branches_in_session("A") == [] + assert v.branches_in_session("B") == [] + v.close() + + +def test_find_by_cache_key(sessions_db: Path): + v = open_sessions(sessions_db) + hits = v.find_by_cache_key("ck1") + assert len(hits) == 1 + assert hits[0].sid == "A" + v.close()