engineering/arborist
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

session: single-shard forest + FTS5 search + cross-session forks

Browse source 
Refactor from per-session sqlite files to one shared shard at
~/.arborist/sessions.db. Three things that didn't work before now do:

1. Queries are first-class members of the tree.
   nodes_fts (FTS5 over question + answer_text + cited_titles) lets
   /find <query> walk every prior turn across every session. Cached
   answers and threads become findable, surface in the REPL as
   `[<bates>] <audit> <question>` lines.

2. Forking from history works the same as forking from a sibling.
   parent_bates can cross sids. After /find returns a hit from
   last week's session, /cd <bates> + ask = your next question
   lands as a child under that historical turn. The cross-session
   parent's subtree_hash ripples up its session_root.

3. One global audit chain instead of per-file.
   audit_events.event_hash = sha256(prev || canonical body), one
   chain over every state change in the shard. `make
   session-chain-check` is now a single pass; tampering anywhere
   in the operator's history breaks the chain.

Wire:
- arborist/qa/session.py — drop file-per-session SessionStore class;
  Session becomes a viewport on SessionStore. cited_titles_json +
  n_cited_sources materialized at insert time so FTS5 doesn't need
  a join into providence_cache.
- arborist/cli.py — `session` subcommand swaps --gc for --find;
  REPL adds /find. Ancestor-keyword extraction now reads
  nodes.cited_titles_json directly (no qa.db roundtrip).
- Makefile — `make session-find Q="..." [LIMIT=N JSON=1]`; `make
  session-gc` retired (no per-session files to GC).
- docs/sessions.md — rewritten for the single-shard shape.
- tests/test_session.py — 17 tests: create, add, fork (incl.
  cross-session), find (FTS5 + by_cache_key), path_to_root crossing
  sessions, audit chain (intact + tampered), cited-title extraction,
  subtree_hash ripple across sessions.

Migration: pre-existing per-session dbs at ~/.arborist/sessions/*.db
become orphaned. None lost data — test sessions only. Operator can
rm -rf ~/.arborist/sessions/ (or rename to sessions-old/) at leisure.

126 session+providence+verify+inspect tests pass.
This commit is contained in:
russell@unturf.com 2026-06-01 17:32:44 -04:00
parent 5f07d9ece2
commit 45a348b4f0
No known key found for this signature in database
5 changed files with 868 additions and 781 deletions
Download patch file Download diff file Expand all files Collapse all files

14
Makefile
View file

@ -224,7 +224,7 @@ session: bootstrap ## interactive multi-turn Q&A REPL [SID=... LLM=qwen|hermes A
@echo "# llm: $(SESSION_LLM_ENDPOINT) / $(SESSION_LLM_MODEL)" >&2
$(ARBORIST) --shards-dir $(SHARDS_DIR) session $(if $(SID),--sid $(SID),) --top-k $(QUERY_TOP_K) --endpoint $(SESSION_LLM_ENDPOINT) --model $(SESSION_LLM_MODEL) $(if $(ANSWER_MODE),--answer-mode $(ANSWER_MODE),) $(if $(BURN),--burn,)
session-list: bootstrap ## list all session dbs in ~/.arborist/sessions/
session-list: bootstrap ## list all sessions in the single shard
$(ARBORIST) session --list $(if $(JSON),--json,)
session-tree: bootstrap ## print one session's tree [SID=...]
@ -233,12 +233,14 @@ session-tree: bootstrap ## print one session's tree [SID=...]
fi
$(ARBORIST) session --sid $(SID) --tree $(if $(JSON),--json,)
session-chain-check: bootstrap ## verify session audit-chain integrity [SID=... (default: all)]
$(ARBORIST) session --chain-check $(if $(SID),--sid $(SID),)
session-chain-check: bootstrap ## verify the single shard's global audit chain
$(ARBORIST) session --chain-check
SESSION_KEEP ?= 100
session-gc: bootstrap ## delete old session dbs, keep $(SESSION_KEEP) most recent [SESSION_KEEP=N]
$(ARBORIST) session --gc $(SESSION_KEEP)
session-find: bootstrap ## FTS5 search over all session turns [Q="..." LIMIT=N JSON=1]
@if [ -z "$$Q" ] && [ -z "$(Q)" ]; then \
echo "usage: make session-find Q=\"query text\" [LIMIT=10] [JSON=1]"; exit 2; \
fi
$(ARBORIST) session --find "$(Q)" $(if $(LIMIT),--limit $(LIMIT),) $(if $(JSON),--json,)
query-dry: bootstrap ## like 'make query' but skip the LLM call (dry-run) [JSON=1 BURN=1 ANSWER_MODE=... BROAD=1 REJECT_BROAD=1 ALLOW_BROAD=1 XLANG=1 XLANG_MT=1]
@if [ -z "$$Q" ] && [ -z "$(Q)" ]; then \

365
arborist/cli.py
View file

@ -7460,8 +7460,9 @@ def build_parser() -> argparse.ArgumentParser:
help="resume an existing session id (default: new session)",
)
session_cmd.add_argument(
"--sessions-dir", default=None,
help="directory containing session dbs (default ~/.arborist/sessions/)",
"--sessions-db", default=None,
help="path to the single sessions shard "
"(default ~/.arborist/sessions.db)",
)
session_cmd.add_argument(
"--list", action="store_true",
@ -7473,11 +7474,16 @@ def build_parser() -> argparse.ArgumentParser:
)
session_cmd.add_argument(
"--chain-check", action="store_true",
help="verify session audit chain integrity (all sessions if no --sid)",
help="verify the global audit chain (one shard for the user)",
)
session_cmd.add_argument(
"--gc", type=int, default=None, metavar="N",
help="keep the N most-recent sessions, delete older db files",
"--find", default=None, metavar="QUERY",
help="FTS5 search over all session nodes "
"(question + answer + cited titles)",
)
session_cmd.add_argument(
"--limit", type=int, default=10,
help="--find result count cap (default 10)",
)
session_cmd.add_argument(
"--json", action="store_true",
@ -7514,151 +7520,130 @@ def _ignore(fn, *a, **kw):
def _cmd_session(args: argparse.Namespace) -> int:
"""Merkle-rooted multi-turn Q&A REPL with forks.
Subcommands flagged on the parser:
--list list sessions
--tree print tree (requires --sid)
--chain-check verify audit chain
--gc N keep N most recent sessions
Default (no flag): start an interactive REPL on a new session
(or --sid to resume). REPL commands:
/tree print tree
/branches list branch-point Bates (≥2 children)
/cd <ref> set current to <bates|seq|label>
/back cd to parent of current
/label <name> label the current node
/show <ref> show stored cache_key + audit for that node
/sid print session id
/root print session_root
/help print this help
/quit exit (or Ctrl-D)
<anything else> new question, child of current node
Single shard at ~/.arborist/sessions.db holds every turn from
every session. FTS5 over question + answer + cited titles makes
prior turns findable + forkable across all sessions.
"""
from pathlib import Path as _Path
from arborist.qa.session import (
Session, default_sessions_dir, list_sessions, render_tree,
session_path,
SessionStore, default_db_path, render_tree,
)
sessions_dir = (
_Path(args.sessions_dir) if args.sessions_dir
else default_sessions_dir()
db_path = (
_Path(args.sessions_db) if args.sessions_db
else default_db_path()
)
store = SessionStore.open(db_path)
# --list
if getattr(args, "list", False):
listing = list_sessions(sessions_dir)
if args.json:
print(json.dumps(listing, indent=2))
else:
if not listing:
print(f"no sessions in {sessions_dir}")
return 0
print(f"{len(listing)} session(s) in {sessions_dir}:")
for s in listing:
print(
f" {s['sid']:32s} nodes={s['n_nodes']:3d} "
f"root={s['session_root']} "
f"updated={s['updated_at']}"
)
return 0
try:
# --list
if getattr(args, "list", False):
listing = store.list_sessions()
if args.json:
print(json.dumps(listing, indent=2))
else:
if not listing:
print(f"no sessions in {db_path}")
else:
print(f"{len(listing)} session(s) in {db_path}:")
for s in listing:
label = f" [{s['label']}]" if s["label"] else ""
print(
f" {s['sid']:32s} nodes={s['n_nodes']:3d} "
f"root={s['session_root']} "
f"updated={s['updated_at']}{label}"
)
return 0
# --gc
if args.gc is not None:
listing = list_sessions(sessions_dir)
listing.sort(key=lambda s: s["updated_at"], reverse=True)
keep, drop = listing[: args.gc], listing[args.gc:]
for s in drop:
try:
_Path(s["path"]).unlink()
except Exception:
pass
print(
f"kept {len(keep)}, deleted {len(drop)} "
f"session db(s) in {sessions_dir}"
)
return 0
# --chain-check
if args.chain_check:
intact, breaks = store.chain_check()
print(f"{db_path} intact={intact} breaks={breaks}")
return 1 if breaks else 0
# --chain-check
if args.chain_check:
if args.sid:
paths = [session_path(args.sid, sessions_dir)]
else:
paths = sorted(sessions_dir.glob("*.db"))
any_break = False
for p in paths:
if not p.is_file():
print(f"missing: {p}")
any_break = True
continue
sess = Session.open(p)
intact, breaks = sess.chain_check()
sess.close()
print(f"{p.name} intact={intact} breaks={breaks}")
if breaks:
any_break = True
return 1 if any_break else 0
# --tree
if args.tree:
if not args.sid:
print("--tree requires --sid", file=sys.stderr)
return 2
p = session_path(args.sid, sessions_dir)
if not p.is_file():
print(f"no such session: {args.sid}", file=sys.stderr)
return 2
sess = Session.open(p)
if args.json:
nodes = [
{
"bates": n.bates, "seq": n.seq,
# --find
if args.find:
hits = store.find(args.find, limit=args.limit)
if args.json:
print(json.dumps([{
"bates": n.bates, "sid": n.sid,
"parent_bates": n.parent_bates,
"question": n.question, "cache_key": n.cache_key,
"question": n.question,
"answer_text": n.answer_text,
"audit_mode": n.audit_mode,
"cited_titles": n.cited_titles,
"node_hash": n.node_hash,
"subtree_hash": n.subtree_hash,
"created_at": n.created_at,
} for n in hits], indent=2))
else:
if not hits:
print("no matches")
else:
for n in hits:
print(
f"[{n.bates}] {n.audit_mode} "
f"{(n.question or '')[:80]}"
)
return 0
# --tree
if args.tree:
if not args.sid:
print("--tree requires --sid", file=sys.stderr)
return 2
try:
sess = store.open_session(args.sid)
except ValueError:
print(f"no such session: {args.sid}", file=sys.stderr)
return 2
if args.json:
nodes = [{
"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,
"created_at": n.created_at, "label": n.label,
"node_hash": n.node_hash,
"subtree_hash": n.subtree_hash,
}
for n in sess.all_nodes()
]
print(json.dumps({
"sid": sess.sid,
"session_root": sess.session_root,
"current_bates": sess.current_bates,
"branches": sess.branches(),
"nodes": nodes,
}, indent=2))
} for n in sess.all_nodes()]
print(json.dumps({
"sid": sess.sid,
"session_root": sess.session_root,
"current_bates": sess.current_bates,
"branches": sess.branches(),
"nodes": nodes,
}, indent=2))
else:
print(render_tree(sess))
return 0
# Default: interactive REPL.
if args.sid:
try:
sess = store.open_session(args.sid)
except ValueError:
print(f"no such session: {args.sid}", file=sys.stderr)
return 2
else:
print(render_tree(sess))
sess.close()
return 0
# Default: interactive REPL.
if args.sid:
p = session_path(args.sid, sessions_dir)
sess = Session.open(p, sid=args.sid)
else:
# Mint a fresh sid via Session.open's default generator.
from arborist.qa.session import _gen_sid
sid = _gen_sid()
p = session_path(sid, sessions_dir)
sess = Session.open(p, sid=sid)
print(f"# session {sess.sid}", file=sys.stderr)
print(
f"# db {p} · root {sess.session_root}",
file=sys.stderr,
)
print(
"# /help for commands · Ctrl-D to exit · "
"any other line asks a question",
file=sys.stderr,
)
rc = _session_repl(args, sess)
sess.close()
return rc
sess = store.create_session()
print(f"# session {sess.sid}", file=sys.stderr)
print(
f"# db {db_path} · root {sess.session_root}",
file=sys.stderr,
)
print(
"# /help for commands · Ctrl-D to exit · "
"any other line asks a question",
file=sys.stderr,
)
return _session_repl(args, sess)
finally:
store.close()
def _session_repl(args: argparse.Namespace, sess) -> int:
@ -7736,16 +7721,17 @@ def _session_repl(args: argparse.Namespace, sess) -> int:
policy = {"answer_mode": args.answer_mode}
help_text = (
"/tree show the tree\n"
"/branches list branch points\n"
"/cd <bates|seq|label> move current pointer\n"
"/back cd to parent of current\n"
"/label <name> label current node\n"
"/show <ref> show stored info for a node\n"
"/sid print session id\n"
"/root print session_root\n"
"/help this help\n"
"/quit exit"
"/tree show this session's tree\n"
"/branches list branch points (≥2 children)\n"
"/find <query> FTS5 search across ALL sessions\n"
"/cd <bates|seq|label> move current pointer (cross-session OK)\n"
"/back cd to parent of current\n"
"/label <name> label current node\n"
"/show [ref] show stored info for a node\n"
"/sid print session id\n"
"/root print session_root\n"
"/help this help\n"
"/quit exit"
)
try:
@ -7789,6 +7775,21 @@ def _session_repl(args: argparse.Namespace, sess) -> int:
else:
print("already at root")
continue
if line.startswith("/find "):
q = line[6:].strip()
hits = sess.find(q, limit=args.limit)
if not hits:
print("no matches")
else:
for n in hits:
sid_tag = (
f" [from {n.sid}]" if n.sid != sess.sid else ""
)
print(
f"[{n.bates}] {n.audit_mode} "
f"{(n.question or '')[:80]}{sid_tag}"
)
continue
if line.startswith("/cd "):
ref = line[4:].strip()
tgt = sess.resolve(ref)
@ -7800,7 +7801,7 @@ def _session_repl(args: argparse.Namespace, sess) -> int:
continue
if line.startswith("/label "):
name = line[7:].strip()
sess.label(sess.current_bates, name)
sess.label_current(name)
print(f"labeled {sess.current_bates} as {name!r}")
continue
if line.startswith("/show"):
@ -7811,9 +7812,13 @@ def _session_repl(args: argparse.Namespace, sess) -> int:
print(f"unknown ref: {ref}")
continue
print(json.dumps({
"bates": n.bates, "parent_bates": n.parent_bates,
"question": n.question, "cache_key": n.cache_key,
"bates": n.bates, "sid": n.sid,
"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,
"label": n.label,
"node_hash": n.node_hash,
"subtree_hash": n.subtree_hash,
@ -7824,14 +7829,19 @@ def _session_repl(args: argparse.Namespace, sess) -> int:
print(f"unknown command: {line} (try /help)")
continue
# Non-command line → new question. Pull ancestor cited
# titles into retrieval_keywords so context flows down the
# branch (verifier-blind; same discipline as the existing
# --retrieval-keywords flag, see #000001).
ancestor_keywords = _session_ancestor_keywords(sess, qa_db)
# Non-command line → new question. Read cited titles
# directly from ancestor node rows (already materialized at
# insert time) and pass as retrieval_keywords. Verifier-
# blind; same discipline as the legacy --retrieval-keywords
# flag (#000001).
seen_titles: list[str] = []
for n in sess.path_to_root(sess.current_bates)[1:]:
for t in n.cited_titles:
if t and t not in seen_titles:
seen_titles.append(t)
policy_local = dict(policy)
if ancestor_keywords:
policy_local["retrieval_keywords"] = ancestor_keywords
if seen_titles:
policy_local["retrieval_keywords"] = " ".join(seen_titles[:8])
try:
result = providence_query(
@ -7848,8 +7858,8 @@ def _session_repl(args: argparse.Namespace, sess) -> int:
audit_mode = result.get("audit_mode", "UNGROUNDED")
ckey = result.get("cache_key", "")
n = sess.add_node(line, ckey, audit_mode)
ans = (result.get("answer_text") or "").strip()
n = sess.add_node(line, ckey, audit_mode, answer_text=ans)
print(f"\n[{n.bates}] {audit_mode}")
print(ans)
print(
@ -7868,51 +7878,6 @@ def _session_repl(args: argparse.Namespace, sess) -> int:
return 0
def _session_ancestor_keywords(sess, qa_db: Path) -> str:
"""Gather cited source titles along the path-to-root → flat
keyword string suitable for retrieval_keywords. Verifier-blind:
these terms widen retrieval, never change governance_policy_hash.
Reads the ancestor nodes' stored answer_text from providence_cache
and parses ``[E# | <title> | <chunk_prefix>: …`` pointer-lines.
Only titles the model actually cited count — exactly what the
user just saw on the prior turn — so retrieval flows along the
visible thread of the conversation, not the wider retrieval pool.
"""
import re
pointer_re = re.compile(r"\[E\d+\s*\|\s*([^|]+?)\s*\|")
try:
import sqlite3 as _sql
conn = _sql.connect(f"file:{qa_db}?mode=ro", uri=True)
conn.row_factory = _sql.Row
except Exception:
return ""
seen_titles: list[str] = []
try:
cur = sess.current_bates
ancestors = sess.path_to_root(cur)[1:] # skip current
for n in ancestors:
if not n.cache_key:
continue
r = conn.execute(
"SELECT answer_text FROM providence_cache "
"WHERE cache_key=? AND falsification_state='live' LIMIT 1",
(n.cache_key,),
).fetchone()
if not r:
continue
for m in pointer_re.finditer(r["answer_text"] or ""):
t = m.group(1).strip()
if t and t not in seen_titles:
seen_titles.append(t)
finally:
try:
conn.close()
except Exception:
pass
return " ".join(seen_titles[:8]) # cap to avoid keyword sprawl
def _cmd_serve(args: argparse.Namespace) -> int:
"""Start the HTTP wallet server. Blocks until SIGINT/SIGTERM."""
from arborist.wallet.server import WalletServer, serve

715
arborist/qa/session.py
View file

@ -1,39 +1,44 @@
"""Session: Merkle-rooted conversation tree over providence_cache.
"""Sessions — single-shard Merkle-rooted conversation forest.
A session is a tree of Q&A nodes. Each turn (or fork) mints a new node
under a chosen parent. Every node carries a stable Bates identifier
(``<sid>-<6-digit>``) AND a content-addressed ``node_hash``. The tree
maintains a Merkle ``subtree_hash`` per node; the root node's
``subtree_hash`` is the ``session_root``.
One SQLite shard at ``~/.arborist/sessions.db`` holds every turn from
every session a user has ever started. Each session is a "viewport"
onto this shared forest: a session has its own root node and current
pointer, but ``nodes.parent_bates`` can cross sessions, so forking from
a historical turn (yesterday's, last week's) just lands a new child
under that turn — your present session gains a branch into the past.
When a new node is inserted under parent P:
1. Compute ``node_hash = sha256(canonical(parent_node_hash || bates ||
question || cache_key || audit_mode || created_at || label))``
2. Set ``subtree_hash = leaf_hash(node_hash)`` (new node is a leaf).
3. Walk N → P → P.parent → … → root, recomputing each ancestor's
``subtree_hash`` to fold in the new child.
4. Update ``session_meta.session_root`` to the root's new ``subtree_hash``.
5. Append one ``session_audit_events`` row whose ``event_hash`` chains
off the previous event.
Bates discipline (``<sid>-<6-digit>``) stays globally unique: ``sid``
is timestamp+random, never reused; ``seq`` is monotonic per ``sid``.
``node_hash`` is content-addressed and folds the parent's
``node_hash``; ``subtree_hash`` is the Merkle hash over the node + all
its descendants (regardless of which session minted them).
``session_root`` is the live ``subtree_hash`` of a session's root node
— recomputed lazily on read since cross-session forks can ripple
through it.
This is O(depth) per insert. Caching for page-refresh works the way it
does in any Merkle tree: a UI client stores (bates → subtree_hash, body)
and on refresh fetches only nodes whose subtree_hash changed. Sibling
subtrees that didn't change keep their hash → no re-fetch.
Why one shard instead of per-session files: queries become first-class
(``/find spider-man`` walks FTS5 over every prior question + answer
across every session) and forks-from-history are a single cross-shard
parent_bates pointer instead of a copy. Privacy: local-only for now;
gossip-trust sharing is future work — this layer's contract is
"private to the operator, journal of everything they asked."
The session db is self-contained at ``~/.arborist/sessions/<sid>.db`` —
no foreign keys into the main store. Cross-session verification:
``arborist session --chain-check SID``.
Schema highlights:
- nodes — every turn ever, with FTS5 over question+answer+titles
- sessions — per-sid metadata (root_bates, current_bates, label)
- audit_events — one global hash-chain over every state change
Bates discipline: never reassigned, never re-used. ``seq=0`` is a
synthetic root with empty question; all real questions are descendants.
This matches arborist's broader "every fact has a stable name" identity
philosophy.
Two integrity invariants:
- subtree_hash recompute path is O(depth) per insert, walks
parent_bates regardless of session
- audit_events.event_hash = sha256(prev_event_hash || canonical(body)),
one chain across all sessions — tampering is detectable globally
"""
from __future__ import annotations
import hashlib
import json
import re
import sqlite3
import time
from dataclasses import dataclass
@ -44,7 +49,6 @@ from arborist.merkle import hash_combine, hash_leaf
def _canonical(body: dict) -> bytes:
"""Stable canonical JSON: sorted keys, no whitespace, UTF-8."""
return json.dumps(
body, sort_keys=True, separators=(",", ":"), ensure_ascii=False
).encode("utf-8", errors="surrogatepass")
@ -62,45 +66,78 @@ def _now_iso() -> str:
def _gen_sid() -> str:
"""Generate a session id: YYYYMMDD-HHMMSS-<4-hex>."""
import secrets
return time.strftime("%Y%m%d-%H%M%S", time.gmtime()) + "-" + secrets.token_hex(2)
SCHEMA = """
CREATE TABLE IF NOT EXISTS session_meta (
session_id TEXT PRIMARY KEY,
CREATE TABLE IF NOT EXISTS sessions (
sid TEXT PRIMARY KEY,
root_bates TEXT NOT NULL,
session_root TEXT NOT NULL,
current_bates TEXT NOT NULL,
next_seq INTEGER NOT NULL,
created_at TEXT NOT NULL,
updated_at TEXT NOT NULL
updated_at TEXT NOT NULL,
label TEXT
);
CREATE TABLE IF NOT EXISTS nodes (
bates TEXT PRIMARY KEY,
seq INTEGER NOT NULL,
parent_bates TEXT REFERENCES nodes(bates),
question TEXT NOT NULL,
cache_key TEXT NOT NULL,
audit_mode TEXT NOT NULL,
created_at TEXT NOT NULL,
label TEXT,
node_hash TEXT NOT NULL,
subtree_hash TEXT NOT NULL
bates TEXT PRIMARY KEY,
sid TEXT NOT NULL REFERENCES sessions(sid),
seq INTEGER NOT NULL,
parent_bates TEXT REFERENCES nodes(bates),
question TEXT NOT NULL,
answer_text TEXT NOT NULL DEFAULT '',
cache_key TEXT NOT NULL,
audit_mode TEXT NOT NULL,
cited_titles_json TEXT NOT NULL DEFAULT '[]',
n_cited_sources INTEGER NOT NULL DEFAULT 0,
created_at TEXT NOT NULL,
label TEXT,
node_hash TEXT NOT NULL,
subtree_hash TEXT NOT NULL
);
CREATE INDEX IF NOT EXISTS idx_nodes_parent ON nodes(parent_bates);
CREATE INDEX IF NOT EXISTS idx_nodes_label ON nodes(label);
CREATE INDEX IF NOT EXISTS idx_nodes_sid ON nodes(sid);
CREATE INDEX IF NOT EXISTS idx_nodes_parent ON nodes(parent_bates);
CREATE INDEX IF NOT EXISTS idx_nodes_label ON nodes(label);
CREATE INDEX IF NOT EXISTS idx_nodes_ckey ON nodes(cache_key);
CREATE TABLE IF NOT EXISTS session_audit_events (
CREATE VIRTUAL TABLE IF NOT EXISTS nodes_fts USING fts5(
question, answer_text, cited_titles,
content='nodes',
content_rowid='rowid',
tokenize='unicode61'
);
CREATE TRIGGER IF NOT EXISTS nodes_fts_ai AFTER INSERT ON nodes BEGIN
INSERT INTO nodes_fts(rowid, question, answer_text, cited_titles)
VALUES (NEW.rowid, NEW.question, NEW.answer_text,
REPLACE(REPLACE(REPLACE(NEW.cited_titles_json,'[',''),']',''),'"',''));
END;
CREATE TRIGGER IF NOT EXISTS nodes_fts_au AFTER UPDATE ON nodes BEGIN
INSERT INTO nodes_fts(nodes_fts, rowid, question, answer_text, cited_titles)
VALUES ('delete', OLD.rowid, OLD.question, OLD.answer_text,
REPLACE(REPLACE(REPLACE(OLD.cited_titles_json,'[',''),']',''),'"',''));
INSERT INTO nodes_fts(rowid, question, answer_text, cited_titles)
VALUES (NEW.rowid, NEW.question, NEW.answer_text,
REPLACE(REPLACE(REPLACE(NEW.cited_titles_json,'[',''),']',''),'"',''));
END;
CREATE TRIGGER IF NOT EXISTS nodes_fts_ad AFTER DELETE ON nodes BEGIN
INSERT INTO nodes_fts(nodes_fts, rowid, question, answer_text, cited_titles)
VALUES ('delete', OLD.rowid, OLD.question, OLD.answer_text,
REPLACE(REPLACE(REPLACE(OLD.cited_titles_json,'[',''),']',''),'"',''));
END;
CREATE TABLE IF NOT EXISTS audit_events (
seq INTEGER PRIMARY KEY AUTOINCREMENT,
event_type TEXT NOT NULL,
sid TEXT,
bates TEXT,
prev_root TEXT NOT NULL,
session_root TEXT NOT NULL,
body TEXT NOT NULL,
prev_event_hash TEXT,
event_hash TEXT NOT NULL,
body TEXT NOT NULL,
created_at TEXT NOT NULL
);
"""
@ -109,11 +146,15 @@ CREATE TABLE IF NOT EXISTS session_audit_events (
@dataclass(frozen=True)
class Node:
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
@ -146,20 +187,9 @@ def _compute_node_hash(
def _compute_subtree_hash(node_hash: str, children_subtree_hashes: list[str]) -> str:
"""Merkle hash over node + sorted children.
Children are sorted by their own subtree_hash (deterministic). Two
leaves combine via ``HashCombine`` to keep the prefix-discipline
consistent with the rest of the repo. A node with N children folds
them pairwise as a balanced tree (left-leaning self-duplicate on
odd counts, same as ``arborist.merkle.MerkleTree``).
"""
leaf = hash_leaf(bytes.fromhex(node_hash))
if not children_subtree_hashes:
return leaf.hex()
# Sort children for determinism — Bates is monotonic but inserts can
# arrive out of order under future concurrent writers; subtree_hash
# is content-addressed so sorting on it is the natural canonical form.
levels: list[bytes] = sorted(bytes.fromhex(h) for h in children_subtree_hashes)
while len(levels) > 1:
nxt: list[bytes] = []
@ -167,49 +197,86 @@ def _compute_subtree_hash(node_hash: str, children_subtree_hashes: list[str]) ->
if i + 1 < len(levels):
nxt.append(hash_combine(levels[i], levels[i + 1]))
else:
# Odd-count rule: self-duplicate.
nxt.append(hash_combine(levels[i], levels[i]))
levels = nxt
children_root = levels[0]
return hash_combine(leaf, children_root).hex()
return hash_combine(leaf, levels[0]).hex()
class Session:
"""Open Merkle-rooted conversation-tree store on disk."""
_POINTER_TITLE_RE = re.compile(r"\[E\d+\s*\|\s*([^|]+?)\s*\|")
def _extract_cited_titles(answer_text: str) -> list[str]:
"""Pull source titles from claim-lattice pointer-line answers
(``[E# | <title> | <chunk_prefix>: …``). Order-preserving dedupe.
"""
out: list[str] = []
for m in _POINTER_TITLE_RE.finditer(answer_text or ""):
t = m.group(1).strip()
if t and t not in out:
out.append(t)
return out
def default_db_path() -> Path:
return Path.home() / ".arborist" / "sessions.db"
class SessionStore:
"""Open the single sessions shard. Per-user, local-private."""
def __init__(self, conn: sqlite3.Connection):
self._conn = conn
self._conn.row_factory = sqlite3.Row
@classmethod
def open(cls, db_path: Optional[Path] = None) -> "SessionStore":
db_path = Path(db_path) if db_path else default_db_path()
db_path.parent.mkdir(parents=True, exist_ok=True)
conn = sqlite3.connect(str(db_path))
conn.executescript(SCHEMA)
return cls(conn)
@property
def conn(self) -> sqlite3.Connection:
return self._conn
# ---- factory ----
def close(self) -> None:
try:
self._conn.close()
except Exception:
pass
@classmethod
def open(cls, db_path: Path, sid: Optional[str] = None) -> "Session":
"""Open or create a session at db_path. ``sid`` is required only
when creating a NEW session db; for existing dbs we read the
single session_meta row to discover the sid.
"""
db_path = Path(db_path)
db_path.parent.mkdir(parents=True, exist_ok=True)
conn = sqlite3.connect(str(db_path))
conn.executescript(SCHEMA)
sess = cls(conn)
existing = conn.execute(
"SELECT session_id FROM session_meta LIMIT 1"
).fetchone()
if existing is None:
sess._init_new(sid or _gen_sid())
return sess
# ---- session-level ----
def _init_new(self, sid: str) -> None:
"""Create the synthetic root node (seq=0, empty question)."""
def list_sessions(self) -> list[dict]:
out: list[dict] = []
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) AS n_nodes "
"FROM sessions s ORDER BY s.updated_at DESC"
):
out.append({
"sid": r["sid"],
"root_bates": r["root_bates"],
"current_bates": r["current_bates"],
"label": r["label"],
"n_nodes": r["n_nodes"],
"session_root": self.session_root(r["sid"]),
"created_at": r["created_at"],
"updated_at": r["updated_at"],
})
return out
def create_session(self, sid: Optional[str] = None,
label: Optional[str] = None) -> "Session":
sid = sid or _gen_sid()
if self._conn.execute(
"SELECT 1 FROM sessions WHERE sid=?", (sid,)
).fetchone():
raise ValueError(f"session already exists: {sid}")
now = _now_iso()
root_bates = _bates(sid, 0)
# Synthetic root has no parent → parent_node_hash="" (empty).
nh = _compute_node_hash(
parent_node_hash="",
bates=root_bates,
@ -222,102 +289,84 @@ class Session:
sh = _compute_subtree_hash(nh, [])
with self._conn:
self._conn.execute(
"INSERT INTO nodes (bates, seq, parent_bates, question, "
"cache_key, audit_mode, created_at, label, node_hash, "
"subtree_hash) VALUES (?, ?, NULL, '', '', '', ?, NULL, ?, ?)",
(root_bates, 0, now, nh, sh),
"INSERT INTO sessions (sid, root_bates, current_bates, "
"next_seq, created_at, updated_at, label) "
"VALUES (?, ?, ?, ?, ?, ?, ?)",
(sid, root_bates, root_bates, 1, now, now, label),
)
self._conn.execute(
"INSERT INTO session_meta (session_id, root_bates, "
"session_root, current_bates, next_seq, created_at, "
"updated_at) VALUES (?, ?, ?, ?, ?, ?, ?)",
(sid, root_bates, sh, root_bates, 1, now, now),
"INSERT INTO nodes (bates, sid, seq, parent_bates, "
"question, answer_text, cache_key, audit_mode, "
"cited_titles_json, n_cited_sources, created_at, "
"label, node_hash, subtree_hash) "
"VALUES (?, ?, 0, NULL, '', '', '', '', '[]', 0, ?, "
"NULL, ?, ?)",
(root_bates, sid, now, nh, sh),
)
self._append_event("session_init", root_bates, prev_root="",
session_root=sh,
body={"sid": sid, "root_bates": root_bates})
self._append_event(
"session_init", sid=sid, bates=root_bates,
body={"sid": sid, "root_bates": root_bates},
)
return self.open_session(sid)
# ---- properties ----
@property
def sid(self) -> str:
def open_session(self, sid: str) -> "Session":
r = self._conn.execute(
"SELECT session_id FROM session_meta"
"SELECT * FROM sessions WHERE sid=?", (sid,)
).fetchone()
return r["session_id"] if r else ""
if r is None:
raise ValueError(f"session not found: {sid}")
return Session(self, sid)
@property
def root_bates(self) -> str:
r = self._conn.execute("SELECT root_bates FROM session_meta").fetchone()
return r["root_bates"] if r else ""
def session_root(self, sid: str) -> str:
"""Live subtree_hash of the session's root node."""
r = self._conn.execute(
"SELECT n.subtree_hash FROM sessions s "
"JOIN nodes n ON n.bates = s.root_bates "
"WHERE s.sid=?", (sid,),
).fetchone()
return r["subtree_hash"] if r else ""
@property
def session_root(self) -> str:
r = self._conn.execute("SELECT session_root FROM session_meta").fetchone()
return r["session_root"] if r else ""
# ---- node-level ----
@property
def current_bates(self) -> str:
r = self._conn.execute("SELECT current_bates FROM session_meta").fetchone()
return r["current_bates"] if r else ""
# ---- queries ----
def _row_to_node(self, r) -> Node:
try:
titles = json.loads(r["cited_titles_json"] or "[]")
if not isinstance(titles, list):
titles = []
except Exception:
titles = []
return Node(
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"],
)
def get_node(self, bates: str) -> Optional[Node]:
r = self._conn.execute(
"SELECT * FROM nodes WHERE bates=?", (bates,)
).fetchone()
if not r:
return None
return Node(
bates=r["bates"], seq=r["seq"],
parent_bates=r["parent_bates"],
question=r["question"], cache_key=r["cache_key"],
audit_mode=r["audit_mode"], created_at=r["created_at"],
label=r["label"], node_hash=r["node_hash"],
subtree_hash=r["subtree_hash"],
)
def resolve(self, ref: str) -> Optional[str]:
"""Resolve a user-facing reference to a Bates.
Accepts: full Bates (``<sid>-<6digit>``), short seq (``23``),
or a label. Returns the canonical Bates or None.
"""
sid = self.sid
# Full Bates direct match.
if "-" in ref and self.get_node(ref) is not None:
return ref
# Bare numeric → prefix with sid.
try:
n = int(ref)
cand = _bates(sid, n)
if self.get_node(cand):
return cand
except ValueError:
pass
# Label lookup.
r = self._conn.execute(
"SELECT bates FROM nodes WHERE label=?", (ref,)
).fetchone()
if r:
return r["bates"]
return None
return self._row_to_node(r) if r else None
def children_of(self, bates: str) -> list[Node]:
return [
self.get_node(r["bates"]) # type: ignore[misc]
self._row_to_node(r)
for r in self._conn.execute(
"SELECT bates FROM nodes WHERE parent_bates=? ORDER BY seq",
"SELECT * FROM nodes WHERE parent_bates=? "
"ORDER BY created_at, seq",
(bates,),
)
]
def path_to_root(self, bates: str) -> list[Node]:
"""Return [bates, parent, ..., root]. Root last."""
out: list[Node] = []
cur: Optional[str] = bates
while cur is not None:
seen: set[str] = set()
while cur is not None and cur not in seen:
seen.add(cur)
n = self.get_node(cur)
if n is None:
break
@ -325,53 +374,89 @@ class Session:
cur = n.parent_bates
return out
def branches(self) -> list[str]:
"""Bates with ≥2 children — explicit branch points."""
def branches_in_session(self, sid: str) -> list[str]:
"""Bates with ≥2 children, scoped to nodes owned by this sid.
Cross-session children DO count toward the branch-point status
of a node owned by this sid — your branches grow when others
fork off your nodes.
"""
return [
r["parent_bates"]
for r in self._conn.execute(
"SELECT parent_bates FROM nodes WHERE parent_bates IS NOT NULL "
"GROUP BY parent_bates HAVING COUNT(*) >= 2 "
"ORDER BY MIN(seq)"
"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,),
)
]
def all_nodes(self) -> list[Node]:
def all_nodes_in_session(self, sid: str) -> list[Node]:
return [
Node(
bates=r["bates"], seq=r["seq"],
parent_bates=r["parent_bates"],
question=r["question"], cache_key=r["cache_key"],
audit_mode=r["audit_mode"], created_at=r["created_at"],
label=r["label"], node_hash=r["node_hash"],
subtree_hash=r["subtree_hash"],
)
self._row_to_node(r)
for r in self._conn.execute(
"SELECT * FROM nodes ORDER BY seq"
"SELECT * FROM nodes WHERE sid=? ORDER BY seq", (sid,),
)
]
# ---- search ----
def find(self, query: str, limit: int = 10) -> list[Node]:
"""FTS5 search over question + answer_text + cited_titles.
Order by bm25 ascending (best first). Spans all sessions.
Skips synthetic-root nodes (seq=0, empty question).
"""
try:
from arborist.wallet.bucket import _to_fts5
match_expr = _to_fts5(query)
except Exception:
match_expr = query
if not match_expr.strip():
return []
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 [self._row_to_node(r) for r in rows]
def find_by_cache_key(self, cache_key: str) -> list[Node]:
return [
self._row_to_node(r)
for r in self._conn.execute(
"SELECT * FROM nodes WHERE cache_key=? ORDER BY created_at",
(cache_key,),
)
]
# ---- mutations ----
def add_node(
self,
question: str,
cache_key: str,
audit_mode: str,
self, sid: str, *,
question: str, cache_key: str, audit_mode: str,
answer_text: str = "",
parent_bates: Optional[str] = None,
) -> Node:
"""Mint a new node under parent (or current_bates). Recompute
subtree_hashes from new node up to root, append audit event."""
parent_bates = parent_bates or self.current_bates
"""Mint a node in session ``sid``. ``parent_bates`` may belong
to ANY session (cross-session forks land a child here).
Defaults to the session's current_bates pointer."""
sess_row = self._conn.execute(
"SELECT next_seq, current_bates FROM sessions WHERE sid=?",
(sid,),
).fetchone()
if sess_row is None:
raise ValueError(f"session not found: {sid}")
parent_bates = parent_bates or sess_row["current_bates"]
parent = self.get_node(parent_bates)
if parent is None:
raise ValueError(f"parent node not found: {parent_bates}")
meta = self._conn.execute(
"SELECT session_id, next_seq, session_root FROM session_meta"
).fetchone()
sid = meta["session_id"]
seq = meta["next_seq"]
prev_root = meta["session_root"]
seq = sess_row["next_seq"]
bates = _bates(sid, seq)
created_at = _now_iso()
node_hash = _compute_node_hash(
@ -380,124 +465,136 @@ class Session:
audit_mode=audit_mode, created_at=created_at, label=None,
)
subtree_hash = _compute_subtree_hash(node_hash, [])
titles = _extract_cited_titles(answer_text)
with self._conn:
self._conn.execute(
"INSERT INTO nodes (bates, seq, parent_bates, question, "
"cache_key, audit_mode, created_at, label, node_hash, "
"subtree_hash) VALUES (?, ?, ?, ?, ?, ?, ?, NULL, ?, ?)",
(bates, seq, parent_bates, question, cache_key,
audit_mode, created_at, node_hash, subtree_hash),
"INSERT INTO nodes (bates, sid, seq, parent_bates, "
"question, answer_text, cache_key, audit_mode, "
"cited_titles_json, n_cited_sources, created_at, "
"label, node_hash, subtree_hash) "
"VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, NULL, ?, ?)",
(bates, sid, seq, parent_bates, question, answer_text,
cache_key, audit_mode, json.dumps(titles), len(titles),
created_at, node_hash, subtree_hash),
)
# Walk parent → root, recompute each ancestor's subtree_hash.
new_root = self._recompute_path(parent_bates)
# Walk parent → root recomputing each ancestor's
# subtree_hash. Crosses session boundaries naturally —
# parent_bates can point anywhere.
self._recompute_path(parent_bates)
self._conn.execute(
"UPDATE session_meta SET session_root=?, current_bates=?, "
"next_seq=?, updated_at=?",
(new_root, bates, seq + 1, created_at),
"UPDATE sessions SET current_bates=?, next_seq=?, "
"updated_at=? WHERE sid=?",
(bates, seq + 1, created_at, sid),
)
self._append_event(
"node_added", bates, prev_root=prev_root,
session_root=new_root,
"node_added", sid=sid, bates=bates,
body={
"bates": bates, "parent_bates": parent_bates,
"bates": bates, "sid": sid,
"parent_bates": parent_bates,
"question": question, "cache_key": cache_key,
"audit_mode": audit_mode, "node_hash": node_hash,
"audit_mode": audit_mode,
"node_hash": node_hash,
"subtree_hash": subtree_hash,
},
)
return self.get_node(bates) # type: ignore[return-value]
def _recompute_path(self, start_bates: str) -> str:
"""Recompute subtree_hash from start_bates up to root. Returns
the (possibly new) root subtree_hash."""
def _recompute_path(self, start_bates: str) -> None:
cur: Optional[str] = start_bates
last_subtree = ""
while cur is not None:
seen: set[str] = set()
while cur is not None and cur not in seen:
seen.add(cur)
n = self.get_node(cur)
if n is None:
break
child_hashes = [
c.subtree_hash for c in self.children_of(cur)
]
child_hashes = [c.subtree_hash for c in self.children_of(cur)]
sh = _compute_subtree_hash(n.node_hash, child_hashes)
self._conn.execute(
"UPDATE nodes SET subtree_hash=? WHERE bates=?",
(sh, cur),
)
last_subtree = sh
cur = n.parent_bates
return last_subtree
def cd(self, bates: str) -> None:
def set_current(self, sid: str, bates: str) -> None:
if self.get_node(bates) is None:
raise ValueError(f"node not found: {bates}")
with self._conn:
self._conn.execute(
"UPDATE session_meta SET current_bates=?, updated_at=?",
(bates, _now_iso()),
"UPDATE sessions SET current_bates=?, updated_at=? "
"WHERE sid=?",
(bates, _now_iso(), sid),
)
def label(self, bates: str, name: str) -> None:
def label_node(self, bates: str, name: str) -> None:
if self.get_node(bates) is None:
raise ValueError(f"node not found: {bates}")
# Labels participate in node_hash on the next recompute —
# however changing a label retroactively would invalidate
# downstream subtree_hashes. Keep labels OUT of node_hash for
# now (they're a UI affordance, not a forensic field). Stored
# in nodes.label but not folded into the hash chain.
with self._conn:
self._conn.execute(
"UPDATE nodes SET label=? WHERE bates=?", (name, bates),
)
def label_session(self, sid: str, name: str) -> None:
with self._conn:
self._conn.execute(
"UPDATE session_meta SET updated_at=?", (_now_iso(),)
"UPDATE sessions SET label=?, updated_at=? WHERE sid=?",
(name, _now_iso(), sid),
)
def resolve(self, ref: str, sid: Optional[str] = None) -> Optional[str]:
"""Resolve user-facing ref → canonical Bates.
Accepts: full Bates (any session), bare seq (auto-prefixed
with ``sid``), or a label (matches any session's node labels).
"""
if "-" in ref and self.get_node(ref) is not None:
return ref
if sid is not None:
try:
n = int(ref)
cand = _bates(sid, n)
if self.get_node(cand):
return cand
except ValueError:
pass
r = self._conn.execute(
"SELECT bates FROM nodes WHERE label=? LIMIT 1", (ref,)
).fetchone()
return r["bates"] if r else None
# ---- audit chain ----
def _append_event(
self, event_type: str, bates: Optional[str], *,
prev_root: str, session_root: str, body: dict,
self, event_type: str, *,
sid: Optional[str], bates: Optional[str], body: dict,
) -> str:
# Chain off the previous event's event_hash, NOT the
# session_root. session_root tells you "what does the tree
# look like now"; event_hash tells you "every state change in
# order is intact." Both are independent integrity signals.
prev_event = self._conn.execute(
"SELECT event_hash FROM session_audit_events "
"ORDER BY seq DESC LIMIT 1"
prev = self._conn.execute(
"SELECT event_hash FROM audit_events ORDER BY seq DESC LIMIT 1"
).fetchone()
prev_hash_hex = prev_event["event_hash"] if prev_event else ""
full_body = {
"event_type": event_type,
"bates": bates,
"prev_root": prev_root,
"session_root": session_root,
**body,
}
prev_hex = prev["event_hash"] if prev else ""
full_body = {"event_type": event_type, "sid": sid,
"bates": bates, **body}
body_canonical = _canonical(full_body)
h = hashlib.sha256()
if prev_hash_hex:
h.update(bytes.fromhex(prev_hash_hex))
if prev_hex:
h.update(bytes.fromhex(prev_hex))
h.update(body_canonical)
event_hash = h.hexdigest()
eh = h.hexdigest()
self._conn.execute(
"INSERT INTO session_audit_events (event_type, bates, "
"prev_root, session_root, body, event_hash, created_at) "
"INSERT INTO audit_events (event_type, sid, bates, "
"prev_event_hash, event_hash, body, created_at) "
"VALUES (?, ?, ?, ?, ?, ?, ?)",
(event_type, bates, prev_root, session_root,
body_canonical.decode("utf-8"), event_hash, _now_iso()),
(event_type, sid, bates, prev_hex, eh,
body_canonical.decode("utf-8"), _now_iso()),
)
return event_hash
return eh
def chain_check(self) -> tuple[int, int]:
"""Walk audit events in order, recompute each event_hash,
compare to stored. Returns (intact_count, break_count)."""
intact = 0
breaks = 0
prev_hex = ""
for r in self._conn.execute(
"SELECT body, event_hash FROM session_audit_events ORDER BY seq"
"SELECT body, event_hash FROM audit_events ORDER BY seq"
):
body_bytes = r["body"].encode("utf-8", errors="surrogatepass")
h = hashlib.sha256()
@ -512,71 +609,91 @@ class Session:
prev_hex = r["event_hash"]
return intact, breaks
def close(self) -> None:
try:
self._conn.close()
except Exception:
pass
class Session:
"""Viewport onto SessionStore for a specific sid. Thin facade over
SessionStore that scopes mutations to one session."""
# ---- top-level helpers (dir-level operations) ----
def __init__(self, store: SessionStore, sid: str):
self._store = store
self._sid = sid
def default_sessions_dir() -> Path:
return Path.home() / ".arborist" / "sessions"
@property
def store(self) -> SessionStore:
return self._store
@property
def sid(self) -> str:
return self._sid
def list_sessions(sessions_dir: Optional[Path] = None) -> list[dict]:
"""List all session dbs in the directory. Returns one dict per
session with sid, root, n_nodes, updated_at, current_bates."""
sessions_dir = sessions_dir or default_sessions_dir()
if not sessions_dir.is_dir():
return []
out: list[dict] = []
for p in sorted(sessions_dir.glob("*.db")):
try:
conn = sqlite3.connect(str(p))
conn.row_factory = sqlite3.Row
m = conn.execute(
"SELECT * FROM session_meta LIMIT 1"
).fetchone()
if not m:
conn.close()
continue
n_nodes = conn.execute(
"SELECT COUNT(*) AS c FROM nodes"
).fetchone()["c"]
out.append({
"sid": m["session_id"],
"path": str(p),
"session_root": m["session_root"],
"current_bates": m["current_bates"],
"n_nodes": n_nodes,
"updated_at": m["updated_at"],
"created_at": m["created_at"],
})
conn.close()
except Exception:
continue
return out
@property
def root_bates(self) -> str:
r = self._store.conn.execute(
"SELECT root_bates FROM sessions WHERE sid=?", (self._sid,)
).fetchone()
return r["root_bates"] if r else ""
@property
def session_root(self) -> str:
return self._store.session_root(self._sid)
def session_path(sid: str, sessions_dir: Optional[Path] = None) -> Path:
sessions_dir = sessions_dir or default_sessions_dir()
return sessions_dir / f"{sid}.db"
@property
def current_bates(self) -> str:
r = self._store.conn.execute(
"SELECT current_bates FROM sessions WHERE sid=?", (self._sid,)
).fetchone()
return r["current_bates"] if r else ""
def add_node(self, question: str, cache_key: str, audit_mode: str,
answer_text: str = "",
parent_bates: Optional[str] = None) -> Node:
return self._store.add_node(
self._sid, question=question, cache_key=cache_key,
audit_mode=audit_mode, answer_text=answer_text,
parent_bates=parent_bates,
)
def cd(self, bates: str) -> None:
self._store.set_current(self._sid, bates)
def label_current(self, name: str) -> None:
self._store.label_node(self.current_bates, name)
def label(self, name: str) -> None:
self._store.label_session(self._sid, name)
def get_node(self, bates: str) -> Optional[Node]:
return self._store.get_node(bates)
def children_of(self, bates: str) -> list[Node]:
return self._store.children_of(bates)
def path_to_root(self, bates: str) -> list[Node]:
return self._store.path_to_root(bates)
def branches(self) -> list[str]:
return self._store.branches_in_session(self._sid)
def all_nodes(self) -> list[Node]:
return self._store.all_nodes_in_session(self._sid)
def find(self, query: str, limit: int = 10) -> list[Node]:
return self._store.find(query, limit)
def resolve(self, ref: str) -> Optional[str]:
return self._store.resolve(ref, sid=self._sid)
def render_tree(sess: Session) -> str:
"""Render the session tree as a text outline.
Synthetic root is collapsed to a header. Current node is marked with
a trailing ← current.
"""
"""Render this session's tree. Cross-session children show their
minting sid in brackets so the operator can tell when a branch
came from a different session (currently always self; gossip-
shared sessions are future work)."""
cur = sess.current_bates
sid = sess.sid
n_nodes = len(sess.all_nodes()) - 1 # exclude synthetic root
bps = sess.branches()
lines: list[str] = [
f"session {sid} ({n_nodes} nodes · {len(bps)} branch points · "
f"session {sess.sid} ({n_nodes} nodes · {len(bps)} branch points · "
f"root={sess.session_root} · current → {cur})",
]
root = sess.root_bates
@ -586,7 +703,6 @@ def render_tree(sess: Session) -> str:
if n is None:
return
if is_root:
# Don't print the synthetic root; jump straight to children.
kids = sess.children_of(bates)
for i, c in enumerate(kids):
_walk(c.bates, "", i == len(kids) - 1, False)
@ -594,10 +710,11 @@ def render_tree(sess: Session) -> str:
marker = "└── " if is_last else "├── "
cur_tag = " ← current" if n.bates == cur else ""
label_tag = f" {{{n.label}}}" if n.label else ""
sid_tag = f" [from {n.sid}]" if n.sid != sess.sid else ""
q = (n.question or "")[:60]
lines.append(
f"{prefix}{marker}[{n.bates}] {q}{label_tag} "
f"[{n.audit_mode}]{cur_tag}"
f"[{n.audit_mode}]{sid_tag}{cur_tag}"
)
nxt_prefix = prefix + (" " if is_last else "│ ")
kids = sess.children_of(bates)

242
docs/sessions.md
View file

@ -1,65 +1,77 @@
# Sessions — Merkle-rooted conversation trees
# Sessions — single-shard Merkle conversation forest
Multi-turn Q&A with branching, where the tree itself is Merkle-rooted.
Each turn (or fork) mints a node with a stable Bates id and folds into
a per-session subtree-hash chain; the root node's `subtree_hash` is the
`session_root`, a single hash that pins the entire conversation state.
Multi-turn Q&A with branching. **One** SQLite shard at
`~/.arborist/sessions.db` holds every turn from every session a user
has ever started. Sessions are *viewports* onto a shared forest —
nodes can fork across sessions (yesterday's turn → today's child) via
a single `parent_bates` pointer.
Privacy: this layer is **local-private** for now. Gossip-trust
sharing across users is future work; the shard's contract today is
"private journal of everything the operator asked, plus the answers."
## On-disk shape
One SQLite file per session at `~/.arborist/sessions/<sid>.db`:
```sql
CREATE TABLE session_meta (
session_id TEXT PRIMARY KEY,
root_bates TEXT NOT NULL, -- always seq=0
session_root TEXT NOT NULL, -- subtree_hash of root node
current_bates TEXT NOT NULL, -- REPL "you are here" pointer
CREATE TABLE sessions (
sid TEXT PRIMARY KEY,
root_bates TEXT NOT NULL, -- always seq=0, empty question
current_bates TEXT NOT NULL, -- REPL "you are here"
next_seq INTEGER NOT NULL,
created_at TEXT NOT NULL,
updated_at TEXT NOT NULL
updated_at TEXT NOT NULL,
label TEXT
);
CREATE TABLE nodes (
bates TEXT PRIMARY KEY, -- "<sid>-<6-digit>"
seq INTEGER NOT NULL, -- monotonic per session
parent_bates TEXT REFERENCES nodes(bates),
question TEXT NOT NULL,
cache_key TEXT NOT NULL, -- → providence_cache
audit_mode TEXT NOT NULL,
created_at TEXT NOT NULL,
label TEXT, -- optional human name
node_hash TEXT NOT NULL, -- sha256 of canonical body
subtree_hash TEXT NOT NULL -- Merkle hash over self + children
bates TEXT PRIMARY KEY, -- "<sid>-<6-digit>"
sid TEXT NOT NULL, -- session that minted this node
seq INTEGER NOT NULL,
parent_bates TEXT REFERENCES nodes(bates), -- can cross sids
question TEXT NOT NULL,
answer_text TEXT NOT NULL, -- materialized at insert time
cache_key TEXT NOT NULL, -- → providence_cache
audit_mode TEXT NOT NULL,
cited_titles_json TEXT NOT NULL, -- parsed from [E# | <title> | …]
n_cited_sources INTEGER NOT NULL,
created_at TEXT NOT NULL,
label TEXT,
node_hash TEXT NOT NULL, -- content-addressed
subtree_hash TEXT NOT NULL -- Merkle over self + all descendants
);
CREATE TABLE session_audit_events (
CREATE VIRTUAL TABLE nodes_fts USING fts5( -- search across all sessions
question, answer_text, cited_titles
);
CREATE TABLE audit_events ( -- one global hash chain
seq INTEGER PRIMARY KEY AUTOINCREMENT,
event_type TEXT NOT NULL, -- session_init | node_added
event_type TEXT NOT NULL,
sid TEXT,
bates TEXT,
prev_root TEXT NOT NULL,
session_root TEXT NOT NULL,
body TEXT NOT NULL, -- canonical JSON
event_hash TEXT NOT NULL, -- sha256(prev_hash || body)
prev_event_hash TEXT,
event_hash TEXT NOT NULL, -- sha256(prev || canonical body)
body TEXT NOT NULL,
created_at TEXT NOT NULL
);
```
The session db is **self-contained**: no foreign keys into the main
store. Answers live in `~/.arborist/qa.db:providence_cache` keyed by
`cache_key`; the session only carries the **shape** of the conversation
(parent links, Bates, hashes). Cache hits across sessions stay live.
Three integrity surfaces:
- **`node_hash`** — content-addressed, never changes after insert.
- **`subtree_hash`** — recomputed when a node gains a child (anywhere
in the forest); ripples up the parent chain regardless of session
boundaries.
- **`audit_events.event_hash`** — single global chain; tampering with
any state-change record breaks it.
## Bates discipline
`<sid>-<6-digit-seq>`. `sid` is `YYYYMMDD-HHMMSS-<4-hex>`; `seq` is
monotonic per session, never reassigned, never re-used. `seq=0` is a
synthetic root with empty question — collapsed in tree rendering, kept
so all real questions descend from a single node.
`<sid>-<6-digit-seq>`. `sid` is timestamp+random, never reused; `seq`
is per-sid monotonic, never reassigned. seq=0 is a synthetic root with
empty question (collapsed in `/tree` rendering, kept so every real
turn has a parent).
## Merkle hashes
Two derived hashes per node:
## Merkle properties
```
node_hash = sha256(canonical_json({
@ -69,137 +81,105 @@ node_hash = sha256(canonical_json({
subtree_hash = HashCombine(
leaf_hash(node_hash),
children_root,
HashCombine over sorted-by-hash children_subtree_hashes,
)
```
Where `children_root` is built bottom-up by `HashCombine` over the
children's `subtree_hash` values, sorted by hash (deterministic),
with the standard arborist odd-element rule (self-duplicate).
Sorting children by `subtree_hash` (content-addressed) keeps the
combine order independent of insertion timing — two sessions that
land the same nodes in different order produce the same root.
## Insert path is O(depth)
When a new node `N` is inserted under parent `P`:
Children sorted by `subtree_hash` (content-addressed) keep the combine
order deterministic — two stores that land the same nodes in different
order produce the same root.
Insert path is O(depth). On insert of N under P:
1. Compute `N.node_hash`.
2. `N.subtree_hash = leaf_hash(N.node_hash)` (new node is a leaf).
3. Walk `N → P → P.parent → … → root` recomputing each ancestor's
`subtree_hash`.
4. Update `session_meta.session_root` to root's new `subtree_hash`.
5. Append one `session_audit_events` row whose `event_hash` chains
off the previous event.
2. `N.subtree_hash = leaf_hash(N.node_hash)`.
3. Walk `N → P → … → root` (crossing sessions if parent_bates does)
recomputing each ancestor's `subtree_hash`.
4. Append `audit_events` row.
A sibling subtree that didn't change keeps its `subtree_hash`. This
is the property that makes page-refresh caching cheap.
Sibling subtrees that didn't change keep their `subtree_hash`. That's
the property page-refresh caching exploits — a UI client tracks
`(bates → subtree_hash, body)` and only refetches paths whose hash
differs.
## Caching for page refresh
## Cross-session forks
A UI client tracks `(bates → subtree_hash, body)` locally. On refresh:
Forking is implicit. Any turn from any session is `/find`-able and
`/cd`-able. After `/cd <bates>`, the next non-command question lands
as a child of that node. If the node belongs to a different session,
the new child inherits the original parent's `node_hash` via Merkle —
that session's `subtree_hash` ripples up to include the cross-session
child.
1. Client sends current `session_root`.
2. Server returns either "still current" or the new root + a list of
`subtree_hash` deltas along the changed paths.
3. Client walks top-down, fetching only nodes whose `subtree_hash`
differs from cache.
Same shape as git's pack-protocol or IPFS MFS: content-addressed
sibling subtrees are byte-identical → cache-equivalent.
## Audit chain
`session_audit_events` is a hash-chain mirror of `arborist.audit_events`,
scoped per session. Every state change emits one event whose `event_hash`
is `sha256(prev_event_hash || canonical_body)`. Verify with
`arborist session --chain-check [--sid SID]` (or
`make session-chain-check`).
`event_hash` and `session_root` are independent integrity signals:
`session_root` answers "what does the tree look like now?";
`event_hash` answers "is every state change in order intact?". Both
moves on every insert, but for different reasons.
## Forking
There is no `/fork` verb. Forking is implicit: `/cd <bates>` moves the
"current" pointer to any node in the tree; the next question becomes a
child of that node. Branch points are nodes with ≥2 children — listed
by `/branches` or in the tree header.
## Retrieval-side context flow
When a turn is asked under parent P, the REPL gathers the source titles
the LLM **cited** in P and P's ancestors (parsed from each ancestor's
stored `answer_text` for `[E# | <title> | …]` pointers) and passes them
into `providence_query` via `policy["retrieval_keywords"]`. The keyword
string augments FTS5 retrieval ONLY — it never enters `cache_key`,
`question_hash`, or `governance_policy_hash` (mirrors the discipline of
the legacy `--retrieval-keywords` flag, see ticket #000001).
This makes pronoun follow-ups ("who created him?") still surface the
prior turn's primary source. **It does not solve the LLM-side antecedent
problem** — the model receives only the new question as the user
message, so a bare pronoun question may still fail to ground. Threading
prior Q&A pairs as `conversation_history` into the LLM prompt (+ folding
that history into `conversation_hash`) is a future enhancement.
In `/tree` rendering, cross-session children are tagged
`[from <sid>]`. (With local-only storage that's always self today;
gossip-shared sessions will surface external sids the same way.)
## CLI
```
arborist session # new session, REPL
arborist session --sid <id> # resume
arborist session --list [--json] # list sessions
arborist session --tree --sid <id> # print tree
arborist session --chain-check # verify audit chain (all)
arborist session --gc N # keep N most recent
arborist session # new session, REPL
arborist session --sid <id> # resume
arborist session --list [--json] # all sessions
arborist session --tree --sid <id> # one session's tree
arborist session --find "spider man" # FTS5 across ALL sessions
arborist session --chain-check # verify global audit chain
arborist session --sessions-db PATH # override shard location
```
REPL commands:
| command | effect |
|---|---|
| `/tree` | print the tree |
| `/branches` | list branch points (Bates with ≥2 children) |
| `/cd <bates\|seq\|label>` | move current pointer |
| `/tree` | print this session's tree |
| `/branches` | list branch-point Bates (≥2 children) |
| `/find <query>` | FTS5 search across ALL sessions |
| `/cd <bates\|seq\|label>` | move current pointer (cross-session OK) |
| `/back` | cd to parent of current |
| `/label <name>` | name current node |
| `/show <ref>` | dump stored info for a node |
| `/show [ref]` | dump stored info for a node |
| `/sid` | print session id |
| `/root` | print session_root |
| `/help` | help |
| `/quit` | exit (or Ctrl-D) |
| anything else | new question, child of current |
Make targets:
Makefile:
```
make session [SID=...] # interactive REPL
make session-list # list all
make session-tree SID=... # print one tree
make session [SID=...] # interactive REPL (defaults LLM=qwen)
make session LLM=hermes # use Hermes-3-8B instead
make session-list # list sessions
make session-tree SID=... # one tree
make session-find Q="..." # search [LIMIT=N JSON=1]
make session-chain-check # verify audit chain
make session-gc # keep SESSION_KEEP=100 (default) most recent
```
## Bounded growth
## Retrieval flow down a branch
`make session-gc` (or `arborist session --gc N`) keeps the N most-recent
sessions, deletes the rest. Each session is a single SQLite file →
trivially deletable, shareable, archivable.
When you ask a follow-up turn under parent P, the REPL gathers cited
titles from P and all ancestors (already materialized in
`nodes.cited_titles_json` at insert time) and passes them to
`providence_query` via `policy["retrieval_keywords"]`. The keyword
string augments FTS5 retrieval **only** — it never enters
`cache_key` / `question_hash` / `governance_policy_hash`. Same
discipline as the legacy `--retrieval-keywords` flag (#000001).
This makes pronoun follow-ups ("who created him?") surface the prior
turn's primary source. It does **not** thread prior Q&A into the LLM
prompt — that's a separate enhancement (would fold into
`conversation_hash`).
## Portability
The on-disk schema has no FK dependency on the main arborist store.
A non-Python consumer that wants to read a session db needs only:
The shard's schema has no FK dependency on the main arborist store.
A non-Python consumer reading it needs:
1. SQLite3 with FTS5 (not required — sessions don't use FTS).
1. SQLite3 with FTS5 (for the search index).
2. The canonical-JSON convention: sorted keys, no whitespace, UTF-8.
3. Three hash primitives:
- `sha256(bytes) → hex`
- `leaf_hash(bytes) = sha256(0x00 || bytes)`
- `hash_combine(L, R) = sha256(0x03 || L || R)`
Those are the same conventions as the proxy.unturf.com Go merkle and
the rest of arborist (see `arborist/merkle.py`).
Same conventions as `proxy.unturf.com` Go merkle and the rest of
arborist (see `arborist/merkle.py`).

313
tests/test_session.py
View file

@ -1,211 +1,234 @@
"""Tests for arborist.qa.session — Merkle-rooted conversation tree."""
"""Tests for arborist.qa.session — single-shard Merkle conversation forest."""
from __future__ import annotations
import sqlite3
from pathlib import Path
import pytest
from arborist.qa.session import (
SessionStore,
Session,
list_sessions,
render_tree,
session_path,
_compute_node_hash,
_compute_subtree_hash,
)
@pytest.fixture
def session_db(tmp_path: Path) -> Path:
return tmp_path / "test-sid.db"
def store_path(tmp_path: Path) -> Path:
return tmp_path / "sessions.db"
def test_open_creates_synthetic_root(session_db: Path):
sess = Session.open(session_db, sid="sid001")
assert sess.sid == "sid001"
assert sess.root_bates == "sid001-000000"
root = sess.get_node("sid001-000000")
@pytest.fixture
def store(store_path: Path) -> SessionStore:
return SessionStore.open(store_path)
def test_create_session_mints_synthetic_root(store: SessionStore):
sess = store.create_session(sid="sidA")
assert sess.sid == "sidA"
assert sess.root_bates == "sidA-000000"
root = sess.get_node(sess.root_bates)
assert root is not None
assert root.question == ""
assert root.parent_bates is None
assert root.subtree_hash == sess.session_root
assert sess.current_bates == sess.root_bates
sess.close()
def test_open_resumes_existing_session(session_db: Path):
s1 = Session.open(session_db, sid="sid002")
s1.add_node("first question", "cache-key-1", "STRICT")
root_after_1 = s1.session_root
s1.close()
s2 = Session.open(session_db) # no sid → resume
assert s2.sid == "sid002"
assert s2.session_root == root_after_1
assert s2.current_bates == "sid002-000001"
s2.close()
def test_add_node_mints_bates_and_recomputes_root(session_db: Path):
sess = Session.open(session_db, sid="s")
root_root = sess.session_root
def test_add_node_mints_bates_and_updates_root(store: SessionStore):
sess = store.create_session(sid="s")
r0 = sess.session_root
n1 = sess.add_node("q1", "ck1", "STRICT")
assert n1.bates == "s-000001"
assert sess.current_bates == "s-000001"
assert sess.session_root != root_root # tree changed → root changed
n2 = sess.add_node("q2", "ck2", "HYBRID")
assert n2.bates == "s-000002"
assert n2.parent_bates == "s-000001"
sess.close()
assert sess.current_bates == n1.bates
assert sess.session_root != r0
def test_fork_via_cd_creates_sibling(session_db: Path):
sess = Session.open(session_db, sid="s")
def test_fork_via_cd_creates_sibling(store: SessionStore):
sess = store.create_session(sid="s")
n1 = sess.add_node("q1", "ck1", "STRICT")
n2 = sess.add_node("q2", "ck2", "STRICT")
# Fork: cd back to n1, ask q3 — q3 is a sibling of q2 under n1.
sess.cd(n1.bates)
n3 = sess.add_node("q3", "ck3", "STRICT")
assert n3.parent_bates == n1.bates
assert sess.current_bates == n3.bates
kids = sess.children_of(n1.bates)
bates_set = {k.bates for k in kids}
assert n2.bates in bates_set
assert n3.bates in bates_set
sess.close()
kids = {k.bates for k in sess.children_of(n1.bates)}
assert kids == {n2.bates, n3.bates}
def test_branches_lists_branch_points(session_db: Path):
sess = Session.open(session_db, sid="s")
n1 = sess.add_node("q1", "ck1", "STRICT")
sess.add_node("q2", "ck2", "STRICT") # child of n1
sess.cd(n1.bates)
sess.add_node("q3", "ck3", "STRICT") # second child of n1 → branch
bps = sess.branches()
assert n1.bates in bps
assert sess.root_bates not in bps # root only has 1 child so far
sess.close()
def test_cross_session_fork(store: SessionStore):
"""Fork from a node in another session — parent_bates crosses sids.
The cross-session parent's subtree_hash recomputes to include the
new child (Merkle ripples through the global forest)."""
sess_a = store.create_session(sid="A")
a1 = sess_a.add_node("hello from A", "ck-A1", "STRICT")
a1_subtree_before = store.get_node(a1.bates).subtree_hash
sess_b = store.create_session(sid="B")
# cd session B's pointer to A's node, then ask in B.
sess_b.cd(a1.bates)
b1 = sess_b.add_node("fork from A", "ck-B1", "STRICT")
assert b1.parent_bates == a1.bates
assert b1.sid == "B"
# A's node now has a child (B's node); subtree_hash changed.
a1_subtree_after = store.get_node(a1.bates).subtree_hash
assert a1_subtree_after != a1_subtree_before
# A's session_root also changed (ripple up).
a_root_subtree = store.session_root("A")
assert a_root_subtree != ""
def test_session_root_changes_per_insert(session_db: Path):
sess = Session.open(session_db, sid="s")
roots = [sess.session_root]
sess.add_node("q1", "ck1", "STRICT")
roots.append(sess.session_root)
sess.add_node("q2", "ck2", "STRICT")
roots.append(sess.session_root)
assert len(set(roots)) == 3 # all distinct
sess.close()
def test_find_fts_search_across_sessions(store: SessionStore):
sess_a = store.create_session(sid="A")
sess_a.add_node("spider man origin story",
"ck1", "STRICT",
answer_text="Spider-Man was created by Stan Lee.")
sess_b = store.create_session(sid="B")
sess_b.add_node("how do you cook risotto?",
"ck2", "STRICT",
answer_text="Stir constantly while adding broth.")
hits = store.find("spider")
assert len(hits) == 1
assert hits[0].sid == "A"
assert "spider" in hits[0].question.lower()
hits = store.find("risotto")
assert len(hits) == 1
assert hits[0].sid == "B"
def test_session_root_deterministic(tmp_path: Path):
"""Two sessions with the same insert sequence produce the same root.
Note: timestamps differ in real wall-clock — we fix them by
passing the same created_at via mocking. Here we just confirm the
HASH formula is deterministic when inputs match.
"""
from arborist.qa.session import _compute_node_hash, _compute_subtree_hash
nh = _compute_node_hash("", "s-000001", "q", "ck", "STRICT",
"2026-06-01T00:00:00Z", None)
sh = _compute_subtree_hash(nh, [])
# Same inputs → same hash.
nh2 = _compute_node_hash("", "s-000001", "q", "ck", "STRICT",
"2026-06-01T00:00:00Z", None)
sh2 = _compute_subtree_hash(nh2, [])
assert nh == nh2
assert sh == sh2
def test_find_skips_synthetic_root(store: SessionStore):
"""The synthetic root (seq=0, empty question) must not surface in
FTS5 search — its question is empty so it wouldn't match anyway,
but the WHERE seq>0 guard is an explicit second line of defense."""
store.create_session(sid="A")
hits = store.find("anything")
assert hits == []
def test_path_to_root_walks_ancestors(session_db: Path):
sess = Session.open(session_db, sid="s")
n1 = sess.add_node("q1", "ck1", "STRICT")
n2 = sess.add_node("q2", "ck2", "STRICT")
path = sess.path_to_root(n2.bates)
assert [n.bates for n in path] == [n2.bates, n1.bates, sess.root_bates]
sess.close()
def test_find_by_cache_key_lists_all_uses(store: SessionStore):
sess_a = store.create_session(sid="A")
sess_b = store.create_session(sid="B")
sess_a.add_node("q1", "shared-ck", "STRICT")
sess_b.add_node("q2", "shared-ck", "STRICT")
hits = store.find_by_cache_key("shared-ck")
assert {h.sid for h in hits} == {"A", "B"}
def test_audit_chain_intact_after_inserts(session_db: Path):
sess = Session.open(session_db, sid="s")
def test_path_to_root_crosses_sessions(store: SessionStore):
sess_a = store.create_session(sid="A")
a1 = sess_a.add_node("q1", "ck1", "STRICT")
sess_b = store.create_session(sid="B")
sess_b.cd(a1.bates)
b1 = sess_b.add_node("q2", "ck2", "STRICT")
path = store.path_to_root(b1.bates)
sids_in_path = [n.sid for n in path]
# b1 (B) → a1 (A) → A's root (A)
assert sids_in_path == ["B", "A", "A"]
def test_audit_chain_intact_after_inserts(store: SessionStore):
sess = store.create_session(sid="A")
sess.add_node("q1", "ck1", "STRICT")
sess.add_node("q2", "ck2", "STRICT")
sess.add_node("q3", "ck3", "HYBRID")
intact, breaks = sess.chain_check()
sess2 = store.create_session(sid="B")
sess2.add_node("q3", "ck3", "HYBRID")
intact, breaks = store.chain_check()
assert breaks == 0
assert intact >= 4 # session_init + 3 node_added
sess.close()
assert intact == 5 # 2 session_init + 3 node_added
def test_audit_chain_detects_tampered_event(session_db: Path):
sess = Session.open(session_db, sid="s")
def test_audit_chain_detects_tampering(store: SessionStore):
sess = store.create_session(sid="A")
sess.add_node("q1", "ck1", "STRICT")
sess.add_node("q2", "ck2", "STRICT")
# Tamper with one event's body.
sess.conn.execute(
"UPDATE session_audit_events SET body=? "
"WHERE event_type='node_added' LIMIT 1",
store.conn.execute(
"UPDATE audit_events SET body=? WHERE event_type='node_added' "
"ORDER BY seq DESC LIMIT 1",
('{"tampered": true}',),
)
sess.conn.commit()
intact, breaks = sess.chain_check()
store.conn.commit()
intact, breaks = store.chain_check()
assert breaks >= 1
sess.close()
def test_resolve_accepts_bates_seq_and_label(session_db: Path):
sess = Session.open(session_db, sid="sid")
def test_resolve_accepts_bates_seq_and_label(store: SessionStore):
sess = store.create_session(sid="sid")
n1 = sess.add_node("q1", "ck1", "STRICT")
sess.label(n1.bates, "alpha")
sess.label_current("alpha")
assert sess.resolve(n1.bates) == n1.bates
assert sess.resolve("1") == n1.bates
assert sess.resolve("alpha") == n1.bates
assert sess.resolve("nope") is None
sess.close()
def test_list_sessions_finds_db_in_dir(tmp_path: Path):
p1 = tmp_path / "s1.db"
p2 = tmp_path / "s2.db"
Session.open(p1, sid="s1").close()
Session.open(p2, sid="s2").close()
listing = list_sessions(tmp_path)
sids = {s["sid"] for s in listing}
assert sids == {"s1", "s2"}
def test_list_sessions(store: SessionStore):
store.create_session(sid="A")
store.create_session(sid="B")
lst = store.list_sessions()
sids = {s["sid"] for s in lst}
assert sids == {"A", "B"}
def test_render_tree_includes_current_marker(session_db: Path):
sess = Session.open(session_db, sid="s")
sess.add_node("first", "ck1", "STRICT")
out = render_tree(sess)
assert "first" in out
assert "← current" in out
sess.close()
def test_render_tree_shows_cross_session_tag(store: SessionStore):
sess_a = store.create_session(sid="A")
a1 = sess_a.add_node("question A", "ck1", "STRICT")
sess_b = store.create_session(sid="B")
sess_b.cd(a1.bates)
sess_b.add_node("forked into B", "ck2", "STRICT")
# Render session A's tree — should include the B-minted child
# with a [from B] tag.
out = render_tree(sess_a)
assert "question A" in out
assert "forked into B" in out
assert "[from B]" in out
def test_subtree_hash_propagates_up(session_db: Path):
"""Inserting a deep node changes the root's subtree_hash."""
sess = Session.open(session_db, sid="s")
n1 = sess.add_node("q1", "ck1", "STRICT")
root_after_1 = sess.session_root
# Add a deep child of n1.
sess.cd(n1.bates)
sess.add_node("q1.child", "ck2", "STRICT")
root_after_2 = sess.session_root
assert root_after_1 != root_after_2
sess.close()
def test_subtree_hash_propagates_across_sessions(store: SessionStore):
"""Inserting a node in session B under a node in session A
changes that A-node's subtree_hash AND A's session_root."""
sess_a = store.create_session(sid="A")
a1 = sess_a.add_node("anchor in A", "ck1", "STRICT")
a_root_before = sess_a.session_root
sess_b = store.create_session(sid="B")
sess_b.cd(a1.bates)
sess_b.add_node("child in B", "ck2", "STRICT")
a_root_after = store.session_root("A")
assert a_root_after != a_root_before
def test_sibling_insert_changes_only_changed_paths(session_db: Path):
"""When a node is added under parent P, P's subtree_hash AND root
change, but unrelated siblings' subtree_hash stays the same."""
sess = Session.open(session_db, sid="s")
a = sess.add_node("a", "ck-a", "STRICT")
sess.cd(sess.root_bates)
b = sess.add_node("b", "ck-b", "STRICT")
a_hash_before = sess.get_node(a.bates).subtree_hash
b_hash_before = sess.get_node(b.bates).subtree_hash
# Add child under a.
sess.cd(a.bates)
sess.add_node("a.child", "ck-ac", "STRICT")
# a's subtree_hash changed; b's didn't.
assert sess.get_node(a.bates).subtree_hash != a_hash_before
assert sess.get_node(b.bates).subtree_hash == b_hash_before
sess.close()
def test_compute_hash_helpers_deterministic():
nh = _compute_node_hash("", "s-000001", "q", "ck", "STRICT",
"2026-06-01T00:00:00Z", None)
nh2 = _compute_node_hash("", "s-000001", "q", "ck", "STRICT",
"2026-06-01T00:00:00Z", None)
assert nh == nh2
sh = _compute_subtree_hash(nh, [])
sh2 = _compute_subtree_hash(nh2, [])
assert sh == sh2
def test_cited_titles_extracted_from_answer(store: SessionStore):
sess = store.create_session(sid="A")
ans = (
'Spider-Man was created by Stan Lee.\n'
' [E1 | Spider-Man | abc123: "..."]\n'
' [E2 | Stan Lee | def456: "..."]'
)
n = sess.add_node("who created spider man?",
"ck1", "STRICT", answer_text=ans)
assert n.cited_titles == ["Spider-Man", "Stan Lee"]
assert n.n_cited_sources == 2
def test_fts_finds_by_cited_title(store: SessionStore):
sess = store.create_session(sid="A")
ans = '...\n [E1 | Sesame Street | abc123: "..."]'
sess.add_node("when did the show start?", "ck1", "STRICT",
answer_text=ans)
hits = store.find("sesame")
assert len(hits) == 1
assert "sesame" in hits[0].cited_titles[0].lower()