arborist.read: add SessionsView read seam for the sessions shard

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

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

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