Upload folder using huggingface_hub

agents/backend.py

@@ -277,6 +277,102 @@ class HuggingFaceBackend(Backend):
         return text
 
 
+# ── Model families + deterministic dry-run backend (Phase 3 exams) ──────────
+
+_FAMILIES = ("claude", "gemma", "llama", "qwen", "deepseek", "mistral",
+             "gpt", "stub")
+
+
+def family_of(spec: str) -> str:
+    """Model family of a backend spec string.
+
+    Used by atp/exams.py to enforce the ATP standard's judge/candidate
+    separation (docs/ATP.md §2 principle 2): the grading model family must
+    differ from the candidate's. Substring match against the known families;
+    unknown specs map to 'unknown' (which still collides with itself, so two
+    unknown specs are treated as the same family — conservative by design).
+
+    The dry-run specs are special-cased to two DISTINCT synthetic families
+    ('dryrun:candidate' → 'stub-a', 'dryrun:judge' → 'stub-b') so CI can
+    exercise the whole exam loop, separation check included, with no real
+    models. Any other 'dryrun:*' spec is plain 'stub'.
+    """
+    s = (spec or "").lower()
+    if s.startswith("dryrun:candidate"):
+        return "stub-a"
+    if s.startswith("dryrun:judge"):
+        return "stub-b"
+    if s.startswith("dryrun"):
+        return "stub"
+    for fam in _FAMILIES:
+        if fam in s:
+            return fam
+    return "unknown"
+
+
+class DryRunBackend(Backend):
+    """Deterministic stub backend for --dry-run exams and CI.
+
+    complete() is a pure function of (spec, system, user): a SHA-256 over the
+    prompts drives everything — no RNG state, no sampling, no network — so the
+    same inputs always yield byte-identical output (which keeps the Phase 3
+    evidence signatures reproducible). Two behaviours:
+
+      * judge-style prompts (they ask for a JSON object with a "score") get
+        valid strict JSON {"score": <0.55..0.95>, "rationale": "..."} with the
+        score derived from the prompt hash;
+      * anything else gets canned-but-plausible worked-answer text.
+
+    Distinct specs ('dryrun:candidate' vs 'dryrun:judge') hash differently,
+    so candidate and judge never parrot each other.
+    """
+
+    def __init__(self, seed_from_spec: str):
+        self.name = seed_from_spec
+        self.seed = seed_from_spec
+
+    def _digest(self, system: str, user: str) -> str:
+        import hashlib
+        raw = f"{self.seed}\x1f{system}\x1f{user}".encode("utf-8")
+        return hashlib.sha256(raw).hexdigest()
+
+    _RATIONALES = (
+        "Matches the reference result and the key steps are shown.",
+        "Correct final answer; one intermediate step is under-justified.",
+        "Mostly right, but the edge case in the reference is not addressed.",
+        "Sound method; minor imprecision versus the reference criteria.",
+        "Agrees with the reference on substance with small stylistic gaps.",
+    )
+    _OPENERS = (
+        "Working step by step:",
+        "Setting up the problem first:",
+        "Applying the standard method:",
+        "Checking the definition before computing:",
+    )
+    _CLOSERS = (
+        "which matches the expected form.",
+        "so the result follows directly.",
+        "verified by substituting back.",
+        "and the boundary case checks out.",
+    )
+
+    def complete(self, system: str, user: str, max_tokens: int = 1500) -> str:
+        import json
+        h = self._digest(system, user)
+        blob = (system + "\n" + user).lower()
+        if '"score"' in blob or ("json" in blob and "score" in blob):
+            # Judge-style prompt → strict JSON verdict, score in [0.55, 0.95).
+            score = 0.55 + (int(h[:8], 16) % 4000) / 10000.0
+            rationale = self._RATIONALES[int(h[8:12], 16) % len(self._RATIONALES)]
+            return json.dumps({"score": round(score, 4), "rationale": rationale})
+        opener = self._OPENERS[int(h[:4], 16) % len(self._OPENERS)]
+        closer = self._CLOSERS[int(h[4:8], 16) % len(self._CLOSERS)]
+        return (
+            f"{opener} restate the givens, apply the relevant rule, and "
+            f"simplify. Final answer: see derivation ref {h[:12]}, {closer}"
+        )
+
+
 # ── Resolver ─────────────────────────────────────────────────────────────────
 
 def get_backend(spec_str: str) -> Backend:
@@ -289,7 +385,11 @@ def get_backend(spec_str: str) -> Backend:
       "hf:google/gemma-2-9b-it"     → HuggingFaceBackend (local, needs CUDA)
       "hf:base+adapter"             → HuggingFaceBackend with LoRA adapter (CUDA)
       "hf-endpoint:https://..."     → HFEndpointBackend (managed, no local GPU)
+      "dryrun:candidate" / "dryrun:judge" → DryRunBackend (deterministic stub,
+                                      Phase 3 --dry-run exams / CI; no network)
     """
+    if spec_str.startswith("dryrun"):
+        return DryRunBackend(spec_str)
     # Global local-LLM override: when LLM_BACKEND=local (or LOCAL_LLM_BASE is
     # set), route claude/* specialists to the local Gemma server too. The
     # multi-agent debate calls get_backend(spec.base_model).complete() (NOT

agents/confusion.py

@@ -44,53 +44,39 @@ def paragraph_hash(text: str) -> str:
     return hashlib.sha1((text or "").strip().encode()).hexdigest()[:12]
 
 
-# ── Persistence (lightweight table; lazy-creates) ─────────────────────────
-
-def _ensure_table(db):
-    db.execute("""
-        CREATE TABLE IF NOT EXISTS confusion_signals (
-          user_id        TEXT NOT NULL,
-          paragraph_hash TEXT NOT NULL,
-          signal_kind    TEXT NOT NULL,
-          value          REAL NOT NULL,
-          ts             REAL NOT NULL
-        )
-    """)
-    db.execute("""
-        CREATE INDEX IF NOT EXISTS idx_confusion_user_para
-        ON confusion_signals(user_id, paragraph_hash)
-    """)
-
+# ── Persistence (confusion_signals — provisioned by migrations/003) ───────
 
 def log_signal(user_id: str, paragraph_hash_str: str,
                signal_kind: str, value: float = 1.0) -> None:
     if signal_kind not in SIGNAL_WEIGHTS:
         return
     import time
-    from .persistence import open_db
-    db = open_db()
-    _ensure_table(db)
+    from atp import db
+    from .persistence import ensure_schema
+    ensure_schema()
     db.execute(
         """INSERT INTO confusion_signals
              (user_id, paragraph_hash, signal_kind, value, ts)
-           VALUES (?, ?, ?, ?, ?)""",
-        (user_id, paragraph_hash_str, signal_kind, float(value), time.time()),
+           VALUES (:user_id, :paragraph_hash, :signal_kind, :value, :ts)""",
+        {"user_id": user_id, "paragraph_hash": paragraph_hash_str,
+         "signal_kind": signal_kind, "value": float(value),
+         "ts": time.time()},
     )
-    db.commit()
 
 
 def score(user_id: str, paragraph_hash_str: str) -> float:
     """Aggregate signals for one paragraph → score in [0, 1]."""
-    from .persistence import open_db
-    db = open_db()
-    _ensure_table(db)
-    rows = db.execute(
+    from atp import db
+    from .persistence import ensure_schema
+    ensure_schema()
+    rows = db.query(
         """SELECT signal_kind, value FROM confusion_signals
-             WHERE user_id = ? AND paragraph_hash = ?""",
-        (user_id, paragraph_hash_str),
-    ).fetchall()
+             WHERE user_id = :user_id AND paragraph_hash = :paragraph_hash""",
+        {"user_id": user_id, "paragraph_hash": paragraph_hash_str},
+    )
     agg = 0.0
-    for kind, value in rows:
+    for row in rows:
+        kind, value = row["signal_kind"], row["value"]
         weight = SIGNAL_WEIGHTS.get(kind, 0.0)
         if kind == "dwell_s":
             extra = max(0.0, value - 30.0)
@@ -102,18 +88,19 @@ def score(user_id: str, paragraph_hash_str: str) -> float:
 
 def hotspots(user_id: str, top_k: int = 5) -> list[dict]:
     """List the user's most-confusing paragraphs across the corpus."""
-    from .persistence import open_db
-    db = open_db()
-    _ensure_table(db)
-    rows = db.execute(
+    from atp import db
+    from .persistence import ensure_schema
+    ensure_schema()
+    rows = db.query(
         """SELECT paragraph_hash
              FROM confusion_signals
-             WHERE user_id = ?
+             WHERE user_id = :user_id
              GROUP BY paragraph_hash""",
-        (user_id,),
-    ).fetchall()
+        {"user_id": user_id},
+    )
     out = []
-    for (ph,) in rows:
+    for row in rows:
+        ph = row["paragraph_hash"]
         out.append({"paragraph_hash": ph, "score": score(user_id, ph)})
     out.sort(key=lambda d: -d["score"])
     return out[:top_k]

agents/heatmap.py

@@ -23,10 +23,8 @@ from typing import Iterable
 
 def mastery_per_cluster(user_id: str, clusters: Iterable) -> dict:
     """Compute per-cluster mastery for a user."""
-    from .persistence import open_db
-    db = open_db()
-    visited = _visited_nodes(db, user_id)
-    sr_state = _sr_state(db, user_id)
+    visited = _visited_nodes(user_id)
+    sr_state = _sr_state(user_id)
 
     out = {}
     for cluster in clusters:
@@ -70,21 +68,24 @@ def weakest(user_id: str, clusters: Iterable, n: int = 5) -> list[int]:
     return [cid for cid, _ in ordered[:n]]
 
 
-def _visited_nodes(db, user_id: str) -> dict:
-    """Map node → {best_quality, n_visits}. Reads sessions table."""
-    rows = db.execute(
-        """SELECT node, MAX(quiz_quality), COUNT(*)
+def _visited_nodes(user_id: str) -> dict:
+    """Map node → {best_quality, n_visits}. Reads walk_events."""
+    from atp import db
+    from .persistence import ensure_schema
+    ensure_schema()
+    rows = db.query(
+        """SELECT node, MAX(quiz_quality) AS best_quality,
+                  COUNT(*) AS n_visits
              FROM walk_events
-            WHERE user_id = ? AND node IS NOT NULL
+            WHERE user_id = :user_id AND node IS NOT NULL
          GROUP BY node""",
-        (user_id,),
-    ).fetchall()
-    return {r[0]: {"best_quality": r[1], "n_visits": r[2]} for r in rows}
+        {"user_id": user_id},
+    )
+    return {r["node"]: {"best_quality": r["best_quality"],
+                        "n_visits": r["n_visits"]} for r in rows}
 
 
-def _sr_state(db, user_id: str) -> dict:
-    rows = db.execute(
-        "SELECT node, repetitions FROM sr_cards WHERE user_id = ?",
-        (user_id,),
-    ).fetchall()
-    return {r[0]: {"repetitions": r[1]} for r in rows}
+def _sr_state(user_id: str) -> dict:
+    from . import persistence
+    return {card["node"]: {"repetitions": card["repetitions"]}
+            for card in persistence.all_sr_cards(user_id)}

agents/jobs.py

@@ -2,108 +2,136 @@
 Lightweight background job queue.
 
 Heavy work (podcast/video render, local TTS) should not block the HTTP
-request — it can take 30-120s. This runs such work on a bounded thread pool
-and tracks status in SQLite so any request can poll it.
+request — it can take 30-120s. This runs such work on a bounded in-process
+thread pool and tracks status in the shared database so any request — on any
+instance — can poll it.
 
-In-process + single-instance by design (matches a single Render web service).
-For multi-instance horizontal scaling, swap the store for Redis/RQ — the
-public API (submit / get) stays the same.
+The fn itself still executes in the submitting process (the closure can't
+cross instances), but every status transition is a guarded UPDATE
+(`WHERE job_id = :id AND status = :expected`), so with multiple instances
+sharing one database a job can never be claimed or double-transitioned
+twice. For fully distributed execution, swap the executor for Redis/RQ —
+the public API (submit / get) stays the same.
 
 Public:
-  submit(kind, fn, *args, **kwargs) -> job_id      # runs fn in background
-  get(job_id) -> dict | None                        # status + result/error
+  submit(kind, fn, *args, org_id="org-demo", **kwargs) -> job_id
+  get(job_id, org_id="org-demo") -> dict | None     # status + result/error
 
 Job dict:
   {job_id, kind, status: queued|running|done|error,
    result: <json>|None, error: str|None, created_at, updated_at}
 
 The fn's return value MUST be JSON-serializable (dicts/lists/str/num).
+
+Tenancy (Phase 2, docs/TENANCY.md leak surface #2): `jobs` is a tenant table
+(migration 005 — org_id column, Postgres RLS deny-by-default), so ALL access
+goes through the atp/tenant_db.py scoped helpers, never raw atp.db
+query()/execute(). Every row carries the submitting org's id: `org_id` is
+consumed by submit() itself (NOT forwarded to fn) and MUST come from the
+verified request state. get() only sees rows belonging to the org it is
+called with — the /jobs route in api/server.py turns a cross-tenant miss
+into the same 404 as an unknown id.
+
+Env:
+  BU_JOB_WORKERS — thread-pool size (default 2; Render is RAM/CPU heavy,
+                   keep concurrency low so the box doesn't OOM).
 """
 
 from __future__ import annotations
 
 import json
+import os
 import time
 import traceback
 import uuid
 from concurrent.futures import ThreadPoolExecutor
 
-from .persistence import open_db
-
-# Render is RAM/CPU heavy — keep concurrency low so the box doesn't OOM.
-_EXECUTOR = ThreadPoolExecutor(max_workers=2, thread_name_prefix="bu-job")
-
-
-def _ensure_table(db) -> None:
-    db.execute("""
-        CREATE TABLE IF NOT EXISTS jobs (
-          job_id     TEXT PRIMARY KEY,
-          kind       TEXT NOT NULL,
-          status     TEXT NOT NULL,
-          result     TEXT,
-          error      TEXT,
-          created_at REAL NOT NULL,
-          updated_at REAL NOT NULL
-        )
-    """)
-
-
-def _set(job_id: str, *, status: str, result=None, error: str | None = None) -> None:
-    db = open_db()
-    _ensure_table(db)
-    db.execute(
-        """UPDATE jobs SET status=?, result=?, error=?, updated_at=?
-             WHERE job_id=?""",
-        (status,
-         json.dumps(result) if result is not None else None,
-         error, time.time(), job_id),
+from atp import tenant_db
+from atp.tenant_db import DEFAULT_ORG
+
+from .persistence import ensure_schema
+
+ensure_schema()  # jobs table lives in migrations/001_core (+ org_id in 005)
+
+_EXECUTOR = ThreadPoolExecutor(
+    max_workers=int(os.environ.get("BU_JOB_WORKERS", "2")),
+    thread_name_prefix="bu-job",
+)
+
+
+def _transition(job_id: str, org_id: str, *, expect: str, to: str,
+                result=None, error: str | None = None) -> bool:
+    """Move a job expect→to atomically. True iff THIS call made the change.
+
+    The `status = :expect` guard means a second worker/instance racing on
+    the same row loses (rowcount 0) instead of overwriting the winner.
+    """
+    res = tenant_db.scoped_execute(
+        org_id,
+        """UPDATE jobs
+              SET status = :to, result = :result, error = :error,
+                  updated_at = :now
+            WHERE job_id = :job_id AND status = :expect
+              AND org_id = :org""",
+        {"to": to,
+         "result": json.dumps(result) if result is not None else None,
+         "error": error, "now": time.time(),
+         "job_id": job_id, "expect": expect, "org": org_id},
     )
-    db.commit()
+    return res.rowcount > 0
+
 
+def submit(kind: str, fn, *args, org_id: str = DEFAULT_ORG, **kwargs) -> str:
+    """Enqueue fn(*args, **kwargs) to run in the background. Returns job_id.
 
-def submit(kind: str, fn, *args, **kwargs) -> str:
-    """Enqueue fn(*args, **kwargs) to run in the background. Returns job_id."""
+    org_id tags the row with the submitting tenant (docs/TENANCY.md); it is
+    consumed here and never forwarded to fn.
+    """
+    org_id = org_id or DEFAULT_ORG
     job_id = "job_" + uuid.uuid4().hex[:12]
     now = time.time()
-    db = open_db()
-    _ensure_table(db)
-    db.execute(
-        """INSERT INTO jobs (job_id, kind, status, created_at, updated_at)
-           VALUES (?, ?, 'queued', ?, ?)""",
-        (job_id, kind, now, now),
-    )
-    db.commit()
+    tenant_db.insert_scoped("jobs", {
+        "job_id": job_id, "kind": kind, "status": "queued",
+        "created_at": now, "updated_at": now,
+    }, org_id)
 
     def _run():
-        _set(job_id, status="running")
+        if not _transition(job_id, org_id, expect="queued", to="running"):
+            return  # already claimed elsewhere — never run the same job twice
         try:
             result = fn(*args, **kwargs)
-            _set(job_id, status="done", result=result)
+            _transition(job_id, org_id, expect="running", to="done", result=result)
         except Exception as e:  # noqa: BLE001 — capture everything for the poller
-            _set(job_id, status="error",
-                 error=f"{type(e).__name__}: {e}\n{traceback.format_exc()[-1500:]}")
+            _transition(
+                job_id, org_id, expect="running", to="error",
+                error=f"{type(e).__name__}: {e}\n{traceback.format_exc()[-1500:]}",
+            )
 
     _EXECUTOR.submit(_run)
     return job_id
 
 
-def get(job_id: str) -> dict | None:
-    """Return the job's current status dict, or None if unknown."""
-    db = open_db()
-    _ensure_table(db)
-    row = db.execute(
+def get(job_id: str, org_id: str = DEFAULT_ORG) -> dict | None:
+    """Return the job's current status dict, or None if unknown.
+
+    Org-scoped: a job belonging to another org is indistinguishable from a
+    nonexistent one (None) — no cross-tenant existence oracle (TENANCY.md #6).
+    """
+    rows = tenant_db.scoped_query(
+        org_id or DEFAULT_ORG,
         """SELECT job_id, kind, status, result, error, created_at, updated_at
-             FROM jobs WHERE job_id=?""",
-        (job_id,),
-    ).fetchone()
-    if not row:
+             FROM jobs WHERE job_id = :job_id AND org_id = :org""",
+        {"job_id": job_id, "org": org_id or DEFAULT_ORG},
+    )
+    if not rows:
         return None
+    row = rows[0]
     return {
-        "job_id":     row[0],
-        "kind":       row[1],
-        "status":     row[2],
-        "result":     json.loads(row[3]) if row[3] else None,
-        "error":      row[4],
-        "created_at": row[5],
-        "updated_at": row[6],
+        "job_id":     row["job_id"],
+        "kind":       row["kind"],
+        "status":     row["status"],
+        "result":     json.loads(row["result"]) if row["result"] else None,
+        "error":      row["error"],
+        "created_at": row["created_at"],
+        "updated_at": row["updated_at"],
     }

agents/paper_drop.py

@@ -200,9 +200,9 @@ def cascade(url_or_id: str, emit: Callable[[str, dict], None] | None = None,
         step("warn", {"step": "semsch", "error": "no_data"})
 
     try:
-        from graph.corpus_graph import build_corpus_graph
+        from graph.corpus_graph import build_mixed_graph
         from graph.clusters import cluster_graph
-        G = build_corpus_graph()
+        G = build_mixed_graph()
         clusters = cluster_graph(G)
     except Exception as e:
         clusters = []

agents/persistence.py

@@ -1,140 +1,148 @@
 """
-SQLite persistence layer.
+Persistence layer — app state via the shared DAL (atp.db).
 
 Stores:
-  users         — anonymous user_id + display_name + created_at
-  walks         — recorded tours (start_node + planned stops + completed)
-  walk_events   — per-stop event log (node, lesson_mode, quiz_quality, ts)
-  sr_cards      — spaced-repetition state per (user, node)
-  podcast_state — per-user per-episode play position + interjections
-  feedback      — thumbs-up/down on lessons (feeds RL ranker)
-
-Default path: data/brain_university.db (override with BRAIN_DB env var).
+  users             — anonymous user_id + display_name + created_at
+  walks             — recorded tours (start_node + planned stops + completed)
+  walk_events       — per-stop event log (node, lesson_mode, quiz_quality, ts)
+  sr_cards          — spaced-repetition state per (user, node)
+  podcast_state     — per-user per-episode play position + interjections
+  feedback          — thumbs-up/down on lessons (feeds RL ranker)
+  socratic_sessions — Socratic dialogue state per session_id
+
+Schema lives in migrations/ (001_core ports the old inline `_SCHEMA`,
+003_confusion the table agents/confusion.py used to lazy-create) and is
+applied once per process at import via ensure_schema() (idempotent).
+
+Database location — resolved by atp.db, in precedence order:
+  1. DATABASE_URL — full SQLAlchemy URL (Postgres in prod:
+     postgresql+psycopg2://…; any sqlite:/// URL also works)
+  2. BRAIN_DB     — legacy env var: filesystem path to a SQLite file
+  3. default     — data/brain_university.db (SQLite; dev/demo)
+
+open_db() survives as a DEPRECATED, SQLite-only shim with ZERO in-repo
+callers — kept only for external scripts holding hand-written sqlite SQL.
+Every call emits a DeprecationWarning naming the caller, and it raises under
+Postgres — use atp.db.query/execute or the helpers below instead.
 """
 
 from __future__ import annotations
 
-import os
+import inspect
+import json
+import re
 import sqlite3
 import time
+import warnings
 from pathlib import Path
-from typing import Iterable
+
+from atp import db
 
 PROJECT_ROOT = Path(__file__).parent.parent
 DEFAULT_DB = PROJECT_ROOT / "data" / "brain_university.db"
 
+_SCHEMA_READY = False
 
-_SCHEMA = """
-CREATE TABLE IF NOT EXISTS users (
-  user_id      TEXT PRIMARY KEY,
-  display_name TEXT,
-  created_at   REAL NOT NULL
-);
-
-CREATE TABLE IF NOT EXISTS walks (
-  walk_id      INTEGER PRIMARY KEY AUTOINCREMENT,
-  user_id      TEXT NOT NULL,
-  start_node   TEXT NOT NULL,
-  planned_json TEXT NOT NULL,
-  completed    INTEGER NOT NULL DEFAULT 0,
-  started_at   REAL NOT NULL,
-  completed_at REAL
-);
-
-CREATE TABLE IF NOT EXISTS walk_events (
-  event_id      INTEGER PRIMARY KEY AUTOINCREMENT,
-  walk_id       INTEGER NOT NULL,
-  user_id       TEXT NOT NULL,
-  node          TEXT,
-  step_idx      INTEGER,
-  lesson_mode   TEXT,
-  quiz_quality  INTEGER,
-  duration_s    REAL,
-  ts            REAL NOT NULL,
-  FOREIGN KEY (walk_id) REFERENCES walks(walk_id)
-);
-CREATE INDEX IF NOT EXISTS idx_walk_events_user_node
-  ON walk_events(user_id, node);
-
-CREATE TABLE IF NOT EXISTS sr_cards (
-  user_id         TEXT NOT NULL,
-  node            TEXT NOT NULL,
-  repetitions     INTEGER NOT NULL DEFAULT 0,
-  ef              REAL    NOT NULL DEFAULT 2.5,
-  interval_days   REAL    NOT NULL DEFAULT 0,
-  due_ts          REAL    NOT NULL DEFAULT 0,
-  last_quality    INTEGER NOT NULL DEFAULT -1,
-  last_review_ts  REAL    NOT NULL DEFAULT 0,
-  PRIMARY KEY (user_id, node)
-);
-CREATE INDEX IF NOT EXISTS idx_sr_due ON sr_cards(user_id, due_ts);
-
-CREATE TABLE IF NOT EXISTS podcast_state (
-  user_id       TEXT NOT NULL,
-  episode_key   TEXT NOT NULL,
-  current_turn  INTEGER NOT NULL DEFAULT 0,
-  interjections TEXT,
-  last_played   REAL NOT NULL,
-  PRIMARY KEY (user_id, episode_key)
-);
-
-CREATE TABLE IF NOT EXISTS feedback (
-  feedback_id INTEGER PRIMARY KEY AUTOINCREMENT,
-  user_id     TEXT NOT NULL,
-  target_kind TEXT NOT NULL,
-  target_id   TEXT NOT NULL,
-  thumb       INTEGER NOT NULL,
-  note        TEXT,
-  ts          REAL NOT NULL
-);
-CREATE INDEX IF NOT EXISTS idx_feedback_target
-  ON feedback(target_kind, target_id);
-
-CREATE TABLE IF NOT EXISTS socratic_sessions (
-  session_id  TEXT PRIMARY KEY,
-  user_id     TEXT NOT NULL,
-  node        TEXT NOT NULL,
-  state_json  TEXT NOT NULL,
-  updated_at  REAL NOT NULL
-);
-"""
+
+def ensure_schema() -> None:
+    """Apply pending migrations once per process (idempotent, guarded)."""
+    global _SCHEMA_READY
+    if not _SCHEMA_READY:
+        db.run_migrations()
+        _SCHEMA_READY = True
+
+
+ensure_schema()
 
 
+# ── Deprecated raw-connection shim ──────────────────────────────────────────
+
 def open_db(path: str | Path | None = None) -> sqlite3.Connection:
-    """Open (and lazily create) the SQLite DB. Returns a Connection."""
-    target = Path(path or os.environ.get("BRAIN_DB", DEFAULT_DB))
+    """DEPRECATED — raw sqlite3.Connection onto the app database.
+
+    No in-repo callers remain (all ported to atp.db / the helpers in this
+    module); kept only for external scripts holding hand-written sqlite SQL.
+    Works ONLY when the app runs on SQLite; under Postgres (DATABASE_URL) it
+    raises — use atp.db.query/execute or the helpers in this module.
+
+    With an explicit `path`, opens that SQLite file and applies the sqlite
+    migration DDL so ad-hoc scratch DBs keep working as before.
+    """
+    caller = inspect.stack()[1]
+    warnings.warn(
+        f"agents.persistence.open_db() is deprecated and SQLite-only "
+        f"(called from {caller.filename}:{caller.lineno} in "
+        f"{caller.function}) — use atp.db.query/execute or the "
+        f"agents.persistence helpers instead.",
+        DeprecationWarning,
+        stacklevel=2,
+    )
+    if path is None:
+        if db.is_postgres():
+            raise RuntimeError(
+                "agents.persistence.open_db() is a SQLite-only legacy shim, "
+                "but this deployment runs on Postgres (DATABASE_URL). Use "
+                "atp.db.query/execute or the agents.persistence helpers."
+            )
+        ensure_schema()  # engine + migrations create the file/schema
+        target = Path(db.get_engine().url.database or DEFAULT_DB)
+        return sqlite3.connect(str(target))
+
+    target = Path(path)
     target.parent.mkdir(parents=True, exist_ok=True)
     conn = sqlite3.connect(str(target))
-    conn.executescript(_SCHEMA)
+    for script in _sqlite_migration_scripts():
+        conn.executescript(script)
     return conn
 
 
+def _sqlite_migration_scripts() -> list[str]:
+    """SQLite DDL of every migration, dialect-preferred (.sqlite.sql > .sql).
+
+    Only for the explicit-path open_db() compat case — the real database is
+    migrated by atp.db.run_migrations(). Assumes idempotent DDL (IF NOT
+    EXISTS), which the shipped migrations use. `--;;` separator lines are
+    plain SQL comments, so executescript() handles the files as-is.
+    """
+    if not db.MIGRATIONS_DIR.is_dir():
+        return []
+    by_base: dict[str, dict[str | None, Path]] = {}
+    for p in db.MIGRATIONS_DIR.iterdir():
+        m = re.match(r"^(\d+_\w+?)(?:\.(pg|sqlite))?\.sql$", p.name)
+        if m:
+            by_base.setdefault(m.group(1), {})[m.group(2)] = p
+    scripts: list[str] = []
+    for base in sorted(by_base):
+        best = by_base[base].get("sqlite") or by_base[base].get(None)
+        if best is not None:  # pg-only bases don't apply to a sqlite file
+            scripts.append(best.read_text())
+    return scripts
+
+
 # ── User helpers ────────────────────────────────────────────────────────────
 
 def ensure_user(user_id: str, display_name: str | None = None) -> None:
-    db = open_db()
     db.execute(
         """INSERT INTO users (user_id, display_name, created_at)
-           VALUES (?, ?, ?)
+           VALUES (:user_id, :display_name, :created_at)
            ON CONFLICT(user_id) DO UPDATE SET
              display_name = COALESCE(excluded.display_name, users.display_name)""",
-        (user_id, display_name, time.time()),
+        {"user_id": user_id, "display_name": display_name,
+         "created_at": time.time()},
     )
-    db.commit()
 
 
 # ── Walks ──────────────────────────────────────────────────────────────────
 
 def start_walk(user_id: str, start_node: str, planned_stops: list[str]) -> int:
-    import json
-    db = open_db()
-    cur = db.execute(
+    res = db.execute(
         """INSERT INTO walks (user_id, start_node, planned_json, started_at)
-           VALUES (?, ?, ?, ?)""",
-        (user_id, start_node, json.dumps(planned_stops), time.time()),
+           VALUES (:user_id, :start_node, :planned_json, :started_at)
+           RETURNING walk_id""",
+        {"user_id": user_id, "start_node": start_node,
+         "planned_json": json.dumps(planned_stops), "started_at": time.time()},
     )
-    db.commit()
-    return cur.lastrowid
+    return int(res.lastrowid)
 
 
 def log_walk_event(
@@ -143,82 +151,139 @@ def log_walk_event(
     lesson_mode: str | None = None, quiz_quality: int | None = None,
     duration_s: float | None = None,
 ) -> None:
-    db = open_db()
     db.execute(
         """INSERT INTO walk_events
              (walk_id, user_id, node, step_idx, lesson_mode,
               quiz_quality, duration_s, ts)
-           VALUES (?, ?, ?, ?, ?, ?, ?, ?)""",
-        (walk_id, user_id, node, step_idx, lesson_mode,
-         quiz_quality, duration_s, time.time()),
+           VALUES (:walk_id, :user_id, :node, :step_idx, :lesson_mode,
+                   :quiz_quality, :duration_s, :ts)""",
+        {"walk_id": walk_id, "user_id": user_id, "node": node,
+         "step_idx": step_idx, "lesson_mode": lesson_mode,
+         "quiz_quality": quiz_quality, "duration_s": duration_s,
+         "ts": time.time()},
     )
-    db.commit()
 
 
 def complete_walk(walk_id: int) -> None:
-    db = open_db()
     db.execute(
-        "UPDATE walks SET completed = 1, completed_at = ? WHERE walk_id = ?",
-        (time.time(), walk_id),
+        """UPDATE walks SET completed = 1, completed_at = :completed_at
+            WHERE walk_id = :walk_id""",
+        {"completed_at": time.time(), "walk_id": walk_id},
     )
-    db.commit()
 
 
 def user_walks(user_id: str, limit: int = 50) -> list[dict]:
-    db = open_db()
-    rows = db.execute(
+    rows = db.query(
         """SELECT walk_id, start_node, planned_json, completed,
                   started_at, completed_at
-             FROM walks WHERE user_id = ?
-            ORDER BY started_at DESC LIMIT ?""",
-        (user_id, limit),
-    ).fetchall()
-    import json
+             FROM walks WHERE user_id = :user_id
+            ORDER BY started_at DESC LIMIT :limit""",
+        {"user_id": user_id, "limit": limit},
+    )
     return [
         {
-            "walk_id": r[0], "start_node": r[1],
-            "planned": json.loads(r[2]),
-            "completed": bool(r[3]),
-            "started_at": r[4], "completed_at": r[5],
+            "walk_id": r["walk_id"], "start_node": r["start_node"],
+            "planned": json.loads(r["planned_json"]),
+            "completed": bool(r["completed"]),
+            "started_at": r["started_at"], "completed_at": r["completed_at"],
         }
         for r in rows
     ]
 
 
+# ── Spaced-repetition cards ────────────────────────────────────────────────
+
+def upsert_sr_card(
+    user_id: str, node: str, *,
+    repetitions: int = 0, ef: float = 2.5, interval_days: float = 0.0,
+    due_ts: float = 0.0, last_quality: int = -1, last_review_ts: float = 0.0,
+) -> None:
+    """Insert-or-update the full SM-2 state for one (user, node) card."""
+    db.execute(
+        """INSERT INTO sr_cards
+               (user_id, node, repetitions, ef, interval_days, due_ts,
+                last_quality, last_review_ts)
+           VALUES (:user_id, :node, :repetitions, :ef, :interval_days,
+                   :due_ts, :last_quality, :last_review_ts)
+           ON CONFLICT(user_id, node) DO UPDATE SET
+               repetitions     = excluded.repetitions,
+               ef              = excluded.ef,
+               interval_days   = excluded.interval_days,
+               due_ts          = excluded.due_ts,
+               last_quality    = excluded.last_quality,
+               last_review_ts  = excluded.last_review_ts""",
+        {"user_id": user_id, "node": node, "repetitions": repetitions,
+         "ef": ef, "interval_days": interval_days, "due_ts": due_ts,
+         "last_quality": last_quality, "last_review_ts": last_review_ts},
+    )
+
+
+def get_sr_card(user_id: str, node: str) -> dict | None:
+    rows = db.query(
+        """SELECT user_id, node, repetitions, ef, interval_days, due_ts,
+                  last_quality, last_review_ts
+             FROM sr_cards WHERE user_id = :user_id AND node = :node""",
+        {"user_id": user_id, "node": node},
+    )
+    return rows[0] if rows else None
+
+
+def due_sr_cards(user_id: str, now: float | None = None,
+                 limit: int = 10) -> list[dict]:
+    """Cards whose due_ts <= now, oldest-due first."""
+    return db.query(
+        """SELECT user_id, node, repetitions, ef, interval_days, due_ts,
+                  last_quality, last_review_ts
+             FROM sr_cards
+            WHERE user_id = :user_id AND due_ts <= :now
+         ORDER BY due_ts ASC LIMIT :limit""",
+        {"user_id": user_id, "now": now or time.time(), "limit": limit},
+    )
+
+
+def all_sr_cards(user_id: str) -> list[dict]:
+    return db.query(
+        """SELECT user_id, node, repetitions, ef, interval_days, due_ts,
+                  last_quality, last_review_ts
+             FROM sr_cards WHERE user_id = :user_id ORDER BY due_ts""",
+        {"user_id": user_id},
+    )
+
+
 # ── Podcast state ──────────────────────────────────────────────────────────
 
 def save_podcast_state(user_id: str, episode_key: str, current_turn: int,
                         interjections: list[dict] | None = None) -> None:
-    import json
-    db = open_db()
     db.execute(
         """INSERT INTO podcast_state
              (user_id, episode_key, current_turn, interjections, last_played)
-           VALUES (?, ?, ?, ?, ?)
+           VALUES (:user_id, :episode_key, :current_turn, :interjections,
+                   :last_played)
            ON CONFLICT(user_id, episode_key) DO UPDATE SET
              current_turn  = excluded.current_turn,
              interjections = excluded.interjections,
              last_played   = excluded.last_played""",
-        (user_id, episode_key, current_turn,
-         json.dumps(interjections or []), time.time()),
+        {"user_id": user_id, "episode_key": episode_key,
+         "current_turn": current_turn,
+         "interjections": json.dumps(interjections or []),
+         "last_played": time.time()},
     )
-    db.commit()
 
 
 def load_podcast_state(user_id: str, episode_key: str) -> dict | None:
-    db = open_db()
-    row = db.execute(
+    rows = db.query(
         """SELECT current_turn, interjections, last_played
-             FROM podcast_state WHERE user_id = ? AND episode_key = ?""",
-        (user_id, episode_key),
-    ).fetchone()
-    if not row:
+             FROM podcast_state
+            WHERE user_id = :user_id AND episode_key = :episode_key""",
+        {"user_id": user_id, "episode_key": episode_key},
+    )
+    if not rows:
         return None
-    import json
+    row = rows[0]
     return {
-        "current_turn":  row[0],
-        "interjections": json.loads(row[1] or "[]"),
-        "last_played":   row[2],
+        "current_turn":  row["current_turn"],
+        "interjections": json.loads(row["interjections"] or "[]"),
+        "last_played":   row["last_played"],
     }
 
 
@@ -226,22 +291,65 @@ def load_podcast_state(user_id: str, episode_key: str) -> dict | None:
 
 def add_feedback(user_id: str, target_kind: str, target_id: str,
                  thumb: int, note: str | None = None) -> None:
-    db = open_db()
     db.execute(
         """INSERT INTO feedback
              (user_id, target_kind, target_id, thumb, note, ts)
-           VALUES (?, ?, ?, ?, ?, ?)""",
-        (user_id, target_kind, target_id, thumb, note, time.time()),
+           VALUES (:user_id, :target_kind, :target_id, :thumb, :note, :ts)""",
+        {"user_id": user_id, "target_kind": target_kind,
+         "target_id": target_id, "thumb": thumb, "note": note,
+         "ts": time.time()},
     )
-    db.commit()
 
 
 def feedback_for_target(target_kind: str, target_id: str) -> list[dict]:
-    db = open_db()
-    rows = db.execute(
+    return db.query(
         """SELECT user_id, thumb, note, ts FROM feedback
-            WHERE target_kind = ? AND target_id = ?
+            WHERE target_kind = :target_kind AND target_id = :target_id
             ORDER BY ts DESC""",
-        (target_kind, target_id),
-    ).fetchall()
-    return [{"user_id": r[0], "thumb": r[1], "note": r[2], "ts": r[3]} for r in rows]
+        {"target_kind": target_kind, "target_id": target_id},
+    )
+
+
+# ── Socratic sessions ──────────────────────────────────────────────────────
+
+def save_socratic_session(session_id: str, user_id: str, node: str,
+                          state: dict) -> None:
+    """Insert-or-replace one Socratic session (state: JSON-serializable)."""
+    db.execute(
+        """INSERT INTO socratic_sessions
+             (session_id, user_id, node, state_json, updated_at)
+           VALUES (:session_id, :user_id, :node, :state_json, :updated_at)
+           ON CONFLICT(session_id) DO UPDATE SET
+             state_json = excluded.state_json,
+             updated_at = excluded.updated_at""",
+        {"session_id": session_id, "user_id": user_id, "node": node,
+         "state_json": json.dumps(state), "updated_at": time.time()},
+    )
+
+
+def load_socratic_session(session_id: str) -> dict | None:
+    rows = db.query(
+        """SELECT session_id, user_id, node, state_json, updated_at
+             FROM socratic_sessions WHERE session_id = :session_id""",
+        {"session_id": session_id},
+    )
+    if not rows:
+        return None
+    row = rows[0]
+    return {
+        "session_id": row["session_id"], "user_id": row["user_id"],
+        "node": row["node"], "state": json.loads(row["state_json"]),
+        "updated_at": row["updated_at"],
+    }
+
+
+def update_socratic_session(session_id: str, state: dict) -> bool:
+    """Update an existing session's state. Returns False if it doesn't exist."""
+    res = db.execute(
+        """UPDATE socratic_sessions
+              SET state_json = :state_json, updated_at = :updated_at
+            WHERE session_id = :session_id""",
+        {"state_json": json.dumps(state), "updated_at": time.time(),
+         "session_id": session_id},
+    )
+    return res.rowcount > 0

agents/spaced_repetition.py

@@ -2,8 +2,9 @@
 SM-2 spaced repetition scheduler.
 
 Each quiz attempt for a (user, node) pair feeds a quality grade 0-5 into
-SuperMemo-2. The scheduler stores per-card state in SQLite (lib/persistence)
-and surfaces "due now" cards back through the trailhead / dashboard.
+SuperMemo-2. The scheduler stores per-card state through agents.persistence
+(SQLite or Postgres via the atp.db DAL) and surfaces "due now" cards back
+through the trailhead / dashboard.
 
 Quality grading (from quiz):
   5  correct on first try, < 5s
@@ -22,8 +23,6 @@ from __future__ import annotations
 
 import time
 from dataclasses import dataclass, asdict
-from datetime import datetime, timedelta
-from pathlib import Path
 
 DAY_SECONDS = 86400
 
@@ -64,88 +63,34 @@ def update_sm2(state: CardState, quality: int, now: float | None = None) -> Card
     return state
 
 
-# ── Persistence ────────────────────────────────────────────────────────────
+# ── Persistence (agents.persistence helpers — SQLite + Postgres) ──────────
 
 def schedule_review(user_id: str, node: str, quality: int,
                     now: float | None = None) -> float:
     """Record a quiz attempt and return the next due timestamp."""
-    from .persistence import open_db
-    db = open_db()
-    state = _load(db, user_id, node) or CardState(user_id=user_id, node=node)
+    from . import persistence
+    row = persistence.get_sr_card(user_id, node)
+    state = CardState(**row) if row else CardState(user_id=user_id, node=node)
     update_sm2(state, quality, now=now)
-    _save(db, state)
+    persistence.upsert_sr_card(**asdict(state))
     return state.due_ts
 
 
 def due_cards(user_id: str, now: float | None = None,
               limit: int = 10) -> list[dict]:
     """Return cards whose due_ts <= now, oldest-due first."""
-    from .persistence import open_db
-    now = now or time.time()
-    db = open_db()
-    rows = db.execute(
-        """SELECT user_id, node, repetitions, ef, interval_days, due_ts,
-                  last_quality, last_review_ts
-             FROM sr_cards
-            WHERE user_id = ? AND due_ts <= ?
-         ORDER BY due_ts ASC LIMIT ?""",
-        (user_id, now, limit),
-    ).fetchall()
-    return [_row_to_dict(r) for r in rows]
+    from . import persistence
+    rows = persistence.due_sr_cards(user_id, now=now, limit=limit)
+    return [_with_due_in_days(r) for r in rows]
 
 
 def all_cards(user_id: str) -> list[dict]:
-    from .persistence import open_db
-    db = open_db()
-    rows = db.execute(
-        """SELECT user_id, node, repetitions, ef, interval_days, due_ts,
-                  last_quality, last_review_ts
-             FROM sr_cards WHERE user_id = ? ORDER BY due_ts""",
-        (user_id,),
-    ).fetchall()
-    return [_row_to_dict(r) for r in rows]
-
-
-def _load(db, user_id: str, node: str) -> CardState | None:
-    row = db.execute(
-        """SELECT user_id, node, repetitions, ef, interval_days, due_ts,
-                  last_quality, last_review_ts
-             FROM sr_cards WHERE user_id = ? AND node = ?""",
-        (user_id, node),
-    ).fetchone()
-    if not row:
-        return None
-    return CardState(
-        user_id=row[0], node=row[1], repetitions=row[2], ef=row[3],
-        interval_days=row[4], due_ts=row[5], last_quality=row[6],
-        last_review_ts=row[7],
-    )
-
-
-def _save(db, state: CardState) -> None:
-    db.execute(
-        """INSERT INTO sr_cards
-               (user_id, node, repetitions, ef, interval_days, due_ts,
-                last_quality, last_review_ts)
-           VALUES (?, ?, ?, ?, ?, ?, ?, ?)
-           ON CONFLICT(user_id, node) DO UPDATE SET
-               repetitions     = excluded.repetitions,
-               ef              = excluded.ef,
-               interval_days   = excluded.interval_days,
-               due_ts          = excluded.due_ts,
-               last_quality    = excluded.last_quality,
-               last_review_ts  = excluded.last_review_ts""",
-        (state.user_id, state.node, state.repetitions, state.ef,
-         state.interval_days, state.due_ts, state.last_quality,
-         state.last_review_ts),
-    )
-    db.commit()
-
-
-def _row_to_dict(row) -> dict:
-    return {
-        "user_id": row[0], "node": row[1], "repetitions": row[2],
-        "ef": row[3], "interval_days": row[4], "due_ts": row[5],
-        "last_quality": row[6], "last_review_ts": row[7],
-        "due_in_days": (row[5] - time.time()) / DAY_SECONDS,
-    }
+    from . import persistence
+    rows = persistence.all_sr_cards(user_id)
+    return [_with_due_in_days(r) for r in rows]
+
+
+def _with_due_in_days(row: dict) -> dict:
+    out = dict(row)
+    out["due_in_days"] = (out["due_ts"] - time.time()) / DAY_SECONDS
+    return out

agents/time_travel.py

@@ -20,38 +20,38 @@ from __future__ import annotations
 import time as _time
 from typing import Iterable
 
-from .persistence import open_db
+from atp import db
+from .persistence import ensure_schema  # import also applies migrations
 
 
 def timeline(user_id: str) -> list[dict]:
     """Distinct timestamps where the user's state changed. Used as
     slider tick positions on the frontend."""
-    db = open_db()
-    rows = db.execute(
+    ensure_schema()
+    rows = db.query(
         """SELECT ts, node, quiz_quality FROM walk_events
-             WHERE user_id = ? AND ts IS NOT NULL
+             WHERE user_id = :user_id AND ts IS NOT NULL
              ORDER BY ts ASC""",
-        (user_id,),
-    ).fetchall()
-    out = []
-    for r in rows:
-        out.append({"ts": r[0], "node": r[1], "quiz_quality": r[2]})
-    return out
+        {"user_id": user_id},
+    )
+    return [{"ts": r["ts"], "node": r["node"],
+             "quiz_quality": r["quiz_quality"]} for r in rows]
 
 
 def snapshot_at(user_id: str, ts: float | None = None) -> dict:
     """Map node → mastery score in [0, 1] as of `ts` (defaults to now)."""
     if ts is None:
         ts = _time.time()
-    db = open_db()
-    rows = db.execute(
-        """SELECT node, MAX(quiz_quality) FROM walk_events
-             WHERE user_id = ? AND node IS NOT NULL AND ts <= ?
+    ensure_schema()
+    rows = db.query(
+        """SELECT node, MAX(quiz_quality) AS max_q FROM walk_events
+             WHERE user_id = :user_id AND node IS NOT NULL AND ts <= :ts
              GROUP BY node""",
-        (user_id, ts),
-    ).fetchall()
+        {"user_id": user_id, "ts": ts},
+    )
     out: dict[str, float] = {}
-    for node, max_q in rows:
+    for row in rows:
+        node, max_q = row["node"], row["max_q"]
         if node is None:
             continue
         if max_q is None: