Loremaster — from zero to a working RAG application

wsl --install

sudo apt update && sudo apt upgrade -y

pwd                  # where am I?
ls -la               # what is here (including hidden files)?
cd foldername        # go into a folder
cd ..                # go up one level
cd ~                 # go home
mkdir -p a/b/c       # create folders (-p makes parents too)
cat file.txt         # print a file
rm file.txt          # delete a file
rm -rf foldername    # delete a folder and everything in it -- careful
cp a.txt b.txt       # copy
mv a.txt b.txt       # move or rename
grep -r "text" .     # search for text in all files under here
Ctrl+C               # stop the running program
Tab                  # autocomplete
Up arrow             # previous command

sudo apt install -y python3-venv python3-pip python3-dev build-essential
python3 --version

cd ~/code/loremaster
python3 -m venv .venv
source .venv/bin/activate
# prompt should now show (.venv)
pip install --upgrade pip

git config --global user.name "Your Name"
git config --global user.email "you@example.com"
git config --global init.defaultBranch main

sudo apt install -y gh
gh auth login

.venv/
__pycache__/
*.pyc
.env
.env.local
*.db
node_modules/
.next/
.DS_Store
data/
uploads/

OPENAI_API_KEY=sk-proj-your-key-here
# or, if you chose Anthropic:
# ANTHROPIC_API_KEY=sk-ant-your-key-here

from dotenv import load_dotenv
from litellm import completion

load_dotenv()          # reads .env and puts the key into the environment

response = completion(
    model="openai/gpt-4o-mini",
    messages=[
        {"role": "system", "content": "You are a concise assistant."},
        {"role": "user",   "content": "In two sentences: what is a tabletop RPG?"},
    ],
    temperature=0.1,
)

print(response.choices[0].message.content)
print("---")
print(response.usage)

import numpy as np
from dotenv import load_dotenv
from litellm import embedding

load_dotenv()

def embed(texts: list[str]) -> list[list[float]]:
    resp = embedding(model="openai/text-embedding-3-small", input=texts)
    return [d["embedding"] for d in resp.data]

def cosine(a, b) -> float:
    a, b = np.array(a), np.array(b)
    return float(a @ b / (np.linalg.norm(a) * np.linalg.norm(b)))

# --- Block 1: meaning beats wording ---
texts = [
    "Harm is recorded in three levels: lesser, moderate, and severe.",
    "How badly can a character get hurt in this game?",
    "The tavern keeper offers a room for two silver pieces.",
]
v = embed(texts)
print("len(vector) =", len(v[0]))
print("first 5 numbers:", v[0][:5])
print()
print("injury rule  <-> injury question :", round(cosine(v[0], v[1]), 3))
print("injury rule  <-> tavern sentence :", round(cosine(v[0], v[2]), 3))

# --- Block 2: the proper-noun weakness ---
names = [
    "Peter cast Eldritch Blast at the guard.",
    "Michael cast Eldritch Blast at the guard.",
    "Peter drank a potion and climbed the wall.",
]
n = embed(names)
print()
print("Peter/Michael, same action :", round(cosine(n[0], n[1]), 3))
print("Peter, different action    :", round(cosine(n[0], n[2]), 3))
print("^ if the first number is not much lower than the second,")
print("  embeddings cannot reliably tell you WHO did something.")

import numpy as np
from dotenv import load_dotenv
from litellm import completion, embedding

load_dotenv()

CHUNKS = [
    {"id": 1, "source": "Blades in the Dark, p.42, Combat > Harm",
     "text": "Harm is recorded in three levels: lesser, moderate, and severe. "
             "Severe harm needs long-term recovery."},
    {"id": 2, "source": "House Rules, p.2",
     "text": "House rule: our table adds a fourth harm level called 'grievous', "
             "which requires a full session of downtime."},
    {"id": 3, "source": "Session 2025-03-14, turn 88",
     "text": "GM: Peter, you're up. PETER: I cast Eldritch Blast at the guard."},
]

SYSTEM = '''Answer using ONLY the numbered sources below. Cite every factual
claim with its bracket number, e.g. [2]. If the sources do not contain the
answer, reply exactly: "I couldn't find that in your materials."
Do not use general knowledge about tabletop games -- the user's house rules
may contradict published rules, and the sources are authoritative.
If sources conflict, present both and note the conflict.'''

def embed(texts):
    r = embedding(model="openai/text-embedding-3-small", input=texts)
    return [np.array(d["embedding"]) for d in r.data]

def cosine(a, b):
    return float(a @ b / (np.linalg.norm(a) * np.linalg.norm(b)))

# --- Ingest: embed every chunk once ---
vectors = embed([c["text"] for c in CHUNKS])

# --- Query ---
question = "How many levels of harm are there?"
qvec = embed([question])[0]

scored = [(cosine(qvec, vec), chunk) for chunk, vec in zip(CHUNKS, vectors)]
scored.sort(key=lambda pair: -pair[0])
top = [chunk for _, chunk in scored[:2]]

context = "\n\n".join(
    f'[{i+1}] ({c["source"]})\n{c["text"]}' for i, c in enumerate(top)
)

answer = completion(
    model="openai/gpt-4o-mini",
    temperature=0.1,
    messages=[
        {"role": "system", "content": SYSTEM},
        {"role": "user", "content": f"{context}\n\nQuestion: {question}"},
    ],
).choices[0].message.content

print("RETRIEVED:")
for i, c in enumerate(top):
    print(f"  [{i+1}] {c['source']}")
print()
print("ANSWER:")
print(answer)

docker ps                    # what is running right now
docker ps -a                 # including stopped ones
docker images                # downloaded images
docker logs <name>           # print a container's output
docker logs -f <name>        # ...and keep following it
docker exec -it <name> bash  # open a shell INSIDE a container
docker compose up -d         # start everything in docker-compose.yml
docker compose down          # stop it all
docker compose down -v       # stop it all AND delete the data volumes

services:
  db:
    image: pgvector/pgvector:pg16
    container_name: loremaster-db
    environment:
      POSTGRES_USER: loremaster
      POSTGRES_PASSWORD: localdevpassword
      POSTGRES_DB: loremaster
    ports:
      - "5432:5432"
    volumes:
      - pgdata:/var/lib/postgresql/data
    healthcheck:
      test: ["CMD-SHELL", "pg_isready -U loremaster"]
      interval: 5s
      retries: 10

volumes:
  pgdata:

DATABASE_URL=postgresql+psycopg://loremaster:localdevpassword@localhost:5432/loremaster

-- 1. Create a table
CREATE TABLE books (
  id         serial PRIMARY KEY,
  title      text NOT NULL,
  doc_type   text NOT NULL CHECK (doc_type IN ('ruleset','transcript')),
  created_at timestamptz NOT NULL DEFAULT now()
);

-- 2. A child table with a foreign key
CREATE TABLE pages (
  id      serial PRIMARY KEY,
  book_id int NOT NULL REFERENCES books(id) ON DELETE CASCADE,
  number  int NOT NULL,
  content text NOT NULL
);

-- 3. Insert
INSERT INTO books (title, doc_type) VALUES ('Blades in the Dark', 'ruleset');
INSERT INTO pages (book_id, number, content)
VALUES (1, 42, 'Harm is recorded in three levels.');

-- 4. Select with a filter
SELECT id, title FROM books WHERE doc_type = 'ruleset';

-- 5. Join
SELECT b.title, p.number, p.content
FROM pages p
JOIN books b ON b.id = p.book_id
WHERE b.doc_type = 'ruleset';

-- 6. Order and limit
SELECT * FROM pages ORDER BY number DESC LIMIT 5;

-- 7. Index (makes the filter in #4 fast at scale)
CREATE INDEX ON books (doc_type);

-- 8. Delete, and watch the cascade
DELETE FROM books WHERE id = 1;
SELECT * FROM pages;   -- empty: the cascade removed them

CREATE TABLE demo_chunks (
  id        serial PRIMARY KEY,
  content   text NOT NULL,
  embedding vector(1536)
);

CREATE INDEX ON demo_chunks USING hnsw (embedding vector_cosine_ops);

import os
import psycopg
from dotenv import load_dotenv
from litellm import embedding
from pgvector.psycopg import register_vector

load_dotenv()
# psycopg wants a plain URL, without the SQLAlchemy "+psycopg" part
DSN = os.environ["DATABASE_URL"].replace("postgresql+psycopg", "postgresql")

def embed(texts):
    r = embedding(model="openai/text-embedding-3-small", input=texts)
    return [d["embedding"] for d in r.data]

SENTENCES = [
    "Harm is recorded in three levels: lesser, moderate, and severe.",
    "A successful Prowl roll lets you move unseen past a single guard.",
    "The crew's heat rises when a score attracts official attention.",
]

with psycopg.connect(DSN) as conn:
    register_vector(conn)

    # --- store ---
    vecs = embed(SENTENCES)
    with conn.cursor() as cur:
        cur.execute("TRUNCATE demo_chunks;")
        for text, vec in zip(SENTENCES, vecs):
            cur.execute(
                "INSERT INTO demo_chunks (content, embedding) VALUES (%s, %s)",
                (text, vec),
            )
    conn.commit()

    # --- search ---
    question = "how much can a character be injured"
    qvec = embed([question])[0]
    with conn.cursor() as cur:
        cur.execute(
            """SELECT content, 1 - (embedding <=> %s::vector) AS similarity
               FROM demo_chunks
               ORDER BY embedding <=> %s::vector
               LIMIT 3""",
            (qvec, qvec),
        )
        print(f"Q: {question}\n")
        for content, sim in cur.fetchall():
            print(f"{sim:.3f}  {content}")

cd ~/code/loremaster
mkdir -p app/{ingestion,retrieval,generation,db} experiments eval scripts data
touch app/__init__.py app/config.py
touch app/ingestion/__init__.py app/retrieval/__init__.py
touch app/generation/__init__.py app/db/__init__.py

from pydantic_settings import BaseSettings, SettingsConfigDict


class Settings(BaseSettings):
    model_config = SettingsConfigDict(env_file=".env", extra="ignore")

    database_url: str

    # Models -- section 9 of the design doc: chat and utility are separate
    chat_model: str = "openai/gpt-4o-mini"
    utility_model: str = "openai/gpt-4o-mini"
    embedding_model: str = "openai/text-embedding-3-small"
    embedding_dim: int = 1536

    # Chunking -- section 6, step 3
    chunk_max_tokens: int = 700
    chunk_overlap_tokens: int = 100
    chunk_min_tokens: int = 80

    # Retrieval -- section 7
    retrieve_candidates: int = 30
    retrieve_final: int = 8

    temperature: float = 0.1


settings = Settings()

CHAT_MODEL=openai/gpt-4o-mini
UTILITY_MODEL=openai/gpt-4o-mini
EMBEDDING_MODEL=openai/text-embedding-3-small
EMBEDDING_DIM=1536

from dataclasses import dataclass
from pathlib import Path

import pymupdf
import pymupdf4llm


@dataclass
class Page:
    number: int          # 1-based
    markdown: str


def is_scanned(path: Path, min_chars_per_page: int = 100) -> bool:
    """Design doc section 6, step 0: sample pages, not just the first one.
    Art-heavy pages in a real text PDF can be sparse too."""
    doc = pymupdf.open(path)
    n = doc.page_count
    sample_idx = sorted({0, n // 2, n - 1, *range(0, n, max(1, n // 8))})
    lengths = [len(doc[i].get_text().strip()) for i in sample_idx if i < n]
    doc.close()
    if not lengths:
        return True
    return (sum(lengths) / len(lengths)) < min_chars_per_page


def extract_pages(path: Path) -> list[Page]:
    """Markdown per page, headings preserved, page numbers retained."""
    raw = pymupdf4llm.to_markdown(str(path), page_chunks=True)
    return [
        Page(number=p["metadata"]["page_number"], markdown=p["text"])
        for p in raw
    ]


if __name__ == "__main__":
    import sys

    path = Path(sys.argv[1])
    if is_scanned(path):
        print("REJECTED: this PDF appears to be a scan with no text layer.")
        raise SystemExit(1)

    pages = extract_pages(path)
    print(f"{len(pages)} pages extracted\n")
    print(pages[0].markdown[:2000])

import re
from collections import Counter

from app.ingestion.extract import Page


def find_repeated_lines(pages: list[Page], threshold: float = 0.6) -> set[str]:
    """Lines appearing on more than `threshold` of pages are headers/footers."""
    counts: Counter[str] = Counter()
    for page in pages:
        # a line can repeat within a page; count it once per page
        seen = {ln.strip() for ln in page.markdown.splitlines() if ln.strip()}
        counts.update(seen)

    cutoff = len(pages) * threshold
    return {line for line, c in counts.items() if c > cutoff and len(line) < 120}


def clean_page(markdown: str, furniture: set[str]) -> str:
    lines = [ln for ln in markdown.splitlines() if ln.strip() not in furniture]
    text = "\n".join(lines)

    text = re.sub(r"[ \t]+", " ", text)          # collapse runs of spaces
    text = re.sub(r"\n{3,}", "\n\n", text)        # at most one blank line
    text = re.sub(r"(\w)-\n(\w)", r"\1\2", text)  # rejoin hyphenated line breaks
    return text.strip()


def normalise(pages: list[Page]) -> list[Page]:
    furniture = find_repeated_lines(pages)
    return [Page(p.number, clean_page(p.markdown, furniture)) for p in pages]

import tiktoken

_enc = tiktoken.get_encoding("cl100k_base")


def count_tokens(text: str) -> int:
    return len(_enc.encode(text))


def truncate_to_tokens(text: str, limit: int) -> str:
    ids = _enc.encode(text)
    return _enc.decode(ids[:limit]) if len(ids) > limit else text


if __name__ == "__main__":
    s = "Harm is recorded in three levels: lesser, moderate, and severe."
    ids = _enc.encode(s)
    print(f"{len(s)} characters -> {len(ids)} tokens")
    print([_enc.decode([i]) for i in ids])

import re
from dataclasses import dataclass, field

from app.config import settings
from app.ingestion.extract import Page
from app.ingestion.tokens import count_tokens

HEADING = re.compile(r"^(#{1,6})\s+(.*)$")
TABLE_ROW = re.compile(r"^\s*\|.*\|\s*$")


@dataclass
class Chunk:
    content: str
    heading_path: str
    page_from: int
    page_to: int
    ordinal: int = 0
    token_count: int = field(default=0)


def _sections(pages: list[Page]):
    """Walk the document, tracking the current heading stack."""
    stack: list[str] = []
    buf: list[str] = []
    start_page = pages[0].number if pages else 1
    end_page = start_page

    for page in pages:
        for line in page.markdown.splitlines():
            m = HEADING.match(line)
            if m:
                if buf and any(x.strip() for x in buf):
                    yield " > ".join(stack), "\n".join(buf), start_page, end_page
                level, title = len(m.group(1)), m.group(2).strip()
                stack = stack[: level - 1] + [title]
                buf, start_page = [], page.number
            else:
                buf.append(line)
        end_page = page.number

    if buf and any(x.strip() for x in buf):
        yield " > ".join(stack), "\n".join(buf), start_page, end_page


def _split_paragraphs(text: str, max_tok: int, overlap_tok: int) -> list[str]:
    """Split oversized text on blank lines; never break a table."""
    blocks, current, in_table = [], [], False
    for para in text.split("\n\n"):
        is_table = bool(TABLE_ROW.match(para.strip().splitlines()[0])) if para.strip() else False
        if is_table:
            if current:
                blocks.append("\n\n".join(current)); current = []
            blocks.append(para)          # table stays whole, whatever its size
        else:
            current.append(para)
    if current:
        blocks.append("\n\n".join(current))

    out, buf = [], []
    for block in blocks:
        candidate = buf + [block]
        if count_tokens("\n\n".join(candidate)) > max_tok and buf:
            out.append("\n\n".join(buf))
            # carry the tail of the previous chunk forward as overlap
            tail, tail_tokens = [], 0
            for prev in reversed(buf):
                tail_tokens += count_tokens(prev)
                tail.insert(0, prev)
                if tail_tokens >= overlap_tok:
                    break
            buf = tail + [block]
        else:
            buf = candidate
    if buf:
        out.append("\n\n".join(buf))
    return out


def chunk_document(pages: list[Page]) -> list[Chunk]:
    max_tok = settings.chunk_max_tokens
    min_tok = settings.chunk_min_tokens
    overlap = settings.chunk_overlap_tokens

    raw: list[Chunk] = []
    for path, text, p_from, p_to in _sections(pages):
        text = text.strip()
        if not text:
            continue
        pieces = ([text] if count_tokens(text) <= max_tok
                  else _split_paragraphs(text, max_tok, overlap))
        for piece in pieces:
            raw.append(Chunk(piece, path, p_from, p_to,
                             token_count=count_tokens(piece)))

    # merge-forward pass for undersized chunks
    merged: list[Chunk] = []
    carry: Chunk | None = None
    for c in raw:
        if carry:
            c = Chunk(carry.content + "\n\n" + c.content,
                      carry.heading_path or c.heading_path,
                      carry.page_from, c.page_to,
                      token_count=count_tokens(carry.content + c.content))
            carry = None
        if c.token_count < min_tok:
            carry = c
            continue
        merged.append(c)
    if carry:
        if merged:
            last = merged[-1]
            last.content += "\n\n" + carry.content
            last.page_to = carry.page_to
            last.token_count = count_tokens(last.content)
        else:
            merged.append(carry)

    for i, c in enumerate(merged):
        c.ordinal = i
    return merged

import time

from litellm import embedding

from app.config import settings
from app.ingestion.chunk import Chunk

BATCH = 100


def contextualise(chunk: Chunk, session_date: str | None = None) -> str:
    """Design doc section 6, step 4: prepend structure BEFORE embedding."""
    if session_date:
        return f"[Session: {session_date}] {chunk.content}"
    if chunk.heading_path:
        return f"{chunk.heading_path} -- {chunk.content}"
    return chunk.content


def embed_texts(texts: list[str], max_retries: int = 5) -> list[list[float]]:
    out: list[list[float]] = []
    for i in range(0, len(texts), BATCH):
        batch = texts[i : i + BATCH]
        for attempt in range(max_retries):
            try:
                resp = embedding(model=settings.embedding_model, input=batch)
                out.extend(d["embedding"] for d in resp.data)
                break
            except Exception as exc:  # rate limits, transient network errors
                if attempt == max_retries - 1:
                    raise
                wait = 2 ** attempt
                print(f"  embed retry {attempt + 1} in {wait}s ({exc})")
                time.sleep(wait)
        print(f"  embedded {min(i + BATCH, len(texts))}/{len(texts)}")
    return out


def embed_chunks(chunks: list[Chunk], session_date: str | None = None):
    return embed_texts([contextualise(c, session_date) for c in chunks])

import uuid
from datetime import date, datetime

from pgvector.sqlalchemy import Vector
from sqlalchemy import (Date, DateTime, ForeignKey, Integer, String, Text,
                        create_engine, func)
from sqlalchemy.dialects.postgresql import UUID
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, sessionmaker

from app.config import settings

engine = create_engine(settings.database_url, echo=False)
SessionLocal = sessionmaker(bind=engine)


class Base(DeclarativeBase):
    pass


class Document(Base):
    __tablename__ = "documents"

    id: Mapped[uuid.UUID] = mapped_column(
        UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
    user_id: Mapped[uuid.UUID] = mapped_column(UUID(as_uuid=True), index=True)
    title: Mapped[str] = mapped_column(Text)
    doc_type: Mapped[str] = mapped_column(String(20))     # ruleset | transcript
    storage_path: Mapped[str] = mapped_column(Text)
    file_hash: Mapped[str | None] = mapped_column(String(64))
    game_system: Mapped[str | None] = mapped_column(Text)
    campaign: Mapped[str | None] = mapped_column(Text)
    session_date: Mapped[date | None] = mapped_column(Date)
    status: Mapped[str] = mapped_column(String(20), default="pending")
    error_message: Mapped[str | None] = mapped_column(Text)
    page_count: Mapped[int | None] = mapped_column(Integer)
    created_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), server_default=func.now())


class Chunk(Base):
    __tablename__ = "chunks"

    id: Mapped[uuid.UUID] = mapped_column(
        UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
    document_id: Mapped[uuid.UUID] = mapped_column(
        ForeignKey("documents.id", ondelete="CASCADE"))
    # Denormalised on purpose -- see design doc section 4
    user_id: Mapped[uuid.UUID] = mapped_column(UUID(as_uuid=True))
    ordinal: Mapped[int] = mapped_column(Integer)
    content: Mapped[str] = mapped_column(Text)
    heading_path: Mapped[str | None] = mapped_column(Text)
    page_from: Mapped[int | None] = mapped_column(Integer)
    page_to: Mapped[int | None] = mapped_column(Integer)
    token_count: Mapped[int] = mapped_column(Integer)
    embedding: Mapped[list[float]] = mapped_column(Vector(settings.embedding_dim))
    created_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), server_default=func.now())

ALTER TABLE chunks
  ADD COLUMN tsv tsvector
  GENERATED ALWAYS AS (to_tsvector('english', content)) STORED;

CREATE INDEX ON chunks USING hnsw (embedding vector_cosine_ops);
CREATE INDEX ON chunks USING gin (tsv);
CREATE INDEX ON chunks (user_id, document_id, ordinal);

import uuid

from sqlalchemy import text

from app.config import settings
from app.db.models import SessionLocal
from app.ingestion.embed import embed_texts

VECTOR_SQL = text("""
    SELECT c.id, c.document_id, c.content, c.heading_path,
           c.page_from, c.page_to, c.ordinal,
           d.title, 1 - (c.embedding <=> (:qvec)::vector) AS score
    FROM chunks c
    JOIN documents d ON d.id = c.document_id
    WHERE c.user_id = :uid
    ORDER BY c.embedding <=> (:qvec)::vector
    LIMIT :k
""")


def vector_search(user_id: uuid.UUID, query: str, k: int | None = None):
    """user_id is the first argument, always. Design doc section 5, layer 1."""
    k = k or settings.retrieve_candidates
    qvec = embed_texts([query])[0]
    with SessionLocal() as session:
        rows = session.execute(
            VECTOR_SQL, {"qvec": str(qvec), "uid": str(user_id), "k": k}
        ).mappings().all()
    return [dict(r) for r in rows]

from app.ingestion.tokens import count_tokens

BUDGET = 12000   # leave room for the system prompt, question, and answer


def build_context(chunks: list[dict], budget: int = BUDGET):
    """Numbered sources with provenance. Design doc section 8, mechanism 1."""
    parts, used, included = [], 0, []

    for chunk in chunks:
        pages = (f"p.{chunk['page_from']}"
                 if chunk["page_from"] == chunk["page_to"]
                 else f"pp.{chunk['page_from']}-{chunk['page_to']}")
        header = f'({chunk["title"]}, {pages}, "{chunk["heading_path"] or ""}")'
        body = f'[{len(included) + 1}] {header}\n{chunk["content"]}'

        cost = count_tokens(body)
        if used + cost > budget:
            break
        parts.append(body)
        included.append(chunk)
        used += cost

    return "\n\n".join(parts), included

import re

from litellm import completion

from app.config import settings

SYSTEM = '''Answer using ONLY the numbered sources below. Cite every factual
claim with its bracket number, e.g. [2]. If the sources do not contain the
answer, reply exactly: "I couldn't find that in your materials."

Do not use general knowledge about tabletop games -- the user's house rules
may contradict published rules, and the sources are authoritative.
If sources conflict, present both and note the conflict.
If the sources answer only part of the question, answer that part and state
what is missing.
State the coverage of your answer, e.g. "Based on the Injury chapter
(pp. 41-46)...", so that a partial answer is visibly partial.'''

REFUSAL = "I couldn't find that in your materials."
CITATION = re.compile(r"\[(\d+)\]")


def verify_citations(answer: str, n_sources: int) -> list[int]:
    """Return any cited numbers that do not exist. Design doc s8, mechanism 4."""
    cited = {int(m) for m in CITATION.findall(answer)}
    return sorted(n for n in cited if n < 1 or n > n_sources)


def generate(question: str, context: str, n_sources: int) -> tuple[str, bool]:
    def call() -> str:
        return completion(
            model=settings.chat_model,
            temperature=settings.temperature,
            messages=[
                {"role": "system", "content": SYSTEM},
                {"role": "user",
                 "content": f"{context}\n\nQuestion: {question}"},
            ],
        ).choices[0].message.content

    answer = call()
    bad = verify_citations(answer, n_sources)
    if bad:
        print(f"  hallucinated citation(s) {bad}; regenerating once")
        answer = call()
        bad = verify_citations(answer, n_sources)
        if bad:
            return answer, False      # caller surfaces an "unverified" flag
    return answer, True

import sys
import uuid

from app.generation.answer import generate
from app.generation.context import build_context
from app.retrieval.search import vector_search

DEV_USER = uuid.UUID("00000000-0000-0000-0000-000000000001")


def main() -> None:
    question = " ".join(sys.argv[1:])
    if not question:
        print('usage: python scripts/ask.py "your question"')
        raise SystemExit(1)

    candidates = vector_search(DEV_USER, question, k=8)
    context, used = build_context(candidates)
    answer, verified = generate(question, context, len(used))

    print(answer)
    print("\n--- sources ---")
    for i, c in enumerate(used, 1):
        print(f"[{i}] {c['title']} p.{c['page_from']} :: {c['heading_path']}")
    if not verified:
        print("\n!! UNVERIFIED: the answer cited a source that does not exist.")


if __name__ == "__main__":
    main()

KEYWORD_SQL = text("""
    SELECT c.id, c.document_id, c.content, c.heading_path,
           c.page_from, c.page_to, c.ordinal, d.title,
           ts_rank_cd(c.tsv, websearch_to_tsquery('english', :q)) AS score
    FROM chunks c
    JOIN documents d ON d.id = c.document_id
    WHERE c.user_id = :uid
      AND c.tsv @@ websearch_to_tsquery('english', :q)
    ORDER BY score DESC
    LIMIT :k
""")


def keyword_search(user_id: uuid.UUID, query: str, k: int | None = None):
    k = k or settings.retrieve_candidates
    with SessionLocal() as session:
        rows = session.execute(
            KEYWORD_SQL, {"q": query, "uid": str(user_id), "k": k}
        ).mappings().all()
    return [dict(r) for r in rows]

from collections import defaultdict

K = 60


def reciprocal_rank_fusion(result_lists: list[list[dict]], k: int = K,
                           limit: int = 30) -> list[dict]:
    """Design doc section 7.1. Ranks only -- scores are never comparable."""
    scores: dict[str, float] = defaultdict(float)
    by_id: dict[str, dict] = {}

    for results in result_lists:
        for rank, row in enumerate(results, start=1):
            cid = str(row["id"])
            scores[cid] += 1.0 / (k + rank)
            by_id.setdefault(cid, row)

    ranked = sorted(scores.items(), key=lambda kv: -kv[1])
    out = []
    for cid, score in ranked[:limit]:
        row = dict(by_id[cid])
        row["rrf_score"] = score
        out.append(row)
    return out

from app.retrieval.fuse import reciprocal_rank_fusion


def hybrid_search(user_id: uuid.UUID, query: str, limit: int | None = None):
    limit = limit or settings.retrieve_candidates
    vec = vector_search(user_id, query, k=limit)
    kw = keyword_search(user_id, query, k=limit)
    return reciprocal_rank_fusion([vec, kw], limit=limit)

import json

from litellm import completion
from pydantic import BaseModel, ValidationError

from app.config import settings

PROMPT = """Score each passage 0-10 for how well it helps answer the question.
10 = directly answers it. 0 = unrelated.
Return ONLY a JSON object: {"scores": [{"i": 0, "s": 7}, ...]}
Include every passage index exactly once.

Question: {question}

Passages:
{passages}"""


class Score(BaseModel):
    i: int
    s: int


class Scores(BaseModel):
    scores: list[Score]


def rerank(question: str, candidates: list[dict], top_n: int | None = None,
           max_retries: int = 2) -> list[dict]:
    top_n = top_n or settings.retrieve_final
    if len(candidates) <= top_n:
        return candidates

    passages = "\n\n".join(
        f"[{i}] {c['content'][:600]}" for i, c in enumerate(candidates)
    )
    prompt = PROMPT.format(question=question, passages=passages)

    for _ in range(max_retries):
        raw = completion(
            model=settings.utility_model,
            temperature=0.0,
            response_format={"type": "json_object"},
            messages=[{"role": "user", "content": prompt}],
        ).choices[0].message.content
        try:
            parsed = Scores.model_validate(json.loads(raw))
            break
        except (json.JSONDecodeError, ValidationError):
            continue
    else:
        # model never complied -- fall back to fusion order rather than failing
        return candidates[:top_n]

    lookup = {s.i: s.s for s in parsed.scores}
    ranked = sorted(
        enumerate(candidates), key=lambda pair: -lookup.get(pair[0], 0)
    )
    out = []
    for idx, cand in ranked[:top_n]:
        row = dict(cand)
        row["rerank_score"] = lookup.get(idx, 0)
        out.append(row)
    return out

- id: q01
  question: "What are the harm levels?"
  kind: factual
  expected_answer: "Lesser, moderate, severe -- plus the house rule 'grievous'."
  expected_chunk_ids: ["<paste-real-uuid>", "<paste-real-uuid>"]
  should_refuse: false

- id: q02
  question: "Was Peter's character a warlock in our March 2025 session?"
  kind: metadata_filtered
  expected_answer: "Yes -- he casts Eldritch Blast in the 2025-03-14 session."
  expected_chunk_ids: ["<paste-real-uuid>"]
  should_refuse: false

- id: q03
  question: "List every injury type in the ruleset."
  kind: aggregative
  expected_answer: "The complete list from the Injury chapter."
  expected_chunk_ids: ["<uuid>", "<uuid>", "<uuid>", "<uuid>"]
  should_refuse: false

- id: q04
  question: "What is the price of a longsword?"
  kind: unanswerable
  expected_answer: null
  expected_chunk_ids: []
  should_refuse: true

import uuid

import yaml
from litellm import completion

from app.config import settings
from app.generation.answer import REFUSAL, generate
from app.generation.context import build_context
from app.retrieval.rerank import rerank
from app.retrieval.search import hybrid_search

DEV_USER = uuid.UUID("00000000-0000-0000-0000-000000000001")

JUDGE = """You are grading an answer for GROUNDEDNESS only.
Is every factual claim in the ANSWER supported by the SOURCES?
Reply with exactly one word: GROUNDED or UNGROUNDED.

SOURCES:
{context}

ANSWER:
{answer}"""


def judge_grounded(context: str, answer: str) -> bool:
    verdict = completion(
        model=settings.utility_model,
        temperature=0.0,
        messages=[{"role": "user",
                   "content": JUDGE.format(context=context, answer=answer)}],
    ).choices[0].message.content
    return "UNGROUNDED" not in verdict.upper()


def main() -> None:
    questions = yaml.safe_load(open("eval/questions.yaml"))
    recalls, grounded, refusal_ok = [], [], []

    for q in questions:
        candidates = hybrid_search(DEV_USER, q["question"])
        top = rerank(q["question"], candidates)
        retrieved = {str(c["id"]) for c in top}

        expected = set(q.get("expected_chunk_ids") or [])
        if expected:
            recalls.append(len(expected & retrieved) / len(expected))

        context, used = build_context(top)
        answer, _ = generate(q["question"], context, len(used))

        refused = REFUSAL.lower() in answer.lower()
        refusal_ok.append(refused == q["should_refuse"])
        if not q["should_refuse"]:
            grounded.append(judge_grounded(context, answer))

        flag = "ok " if refused == q["should_refuse"] else "MISS"
        print(f"{flag} {q['id']:>4} [{q['kind']:<17}] {q['question'][:50]}")

    def pct(xs):
        return f"{100 * sum(xs) / len(xs):.1f}%" if xs else "n/a"

    print("\n=== results ===")
    print(f"recall@{settings.retrieve_final:<3}    {pct(recalls)}")
    print(f"groundedness    {pct(grounded)}")
    print(f"refusal acc.    {pct(refusal_ok)}")


if __name__ == "__main__":
    main()

from contextlib import asynccontextmanager

from fastapi import Depends, FastAPI, HTTPException
from fastapi.middleware.cors import CORSMiddleware
from pydantic import BaseModel
from sqlalchemy import text
from sqlalchemy.orm import Session

from app.db.models import SessionLocal


def get_db():
    """One session per request, always closed. Injected with Depends."""
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()


@asynccontextmanager
async def lifespan(app: FastAPI):
    print("starting up")
    yield
    print("shutting down")


app = FastAPI(title="Loremaster API", lifespan=lifespan)

app.add_middleware(
    CORSMiddleware,
    allow_origins=["http://localhost:3000"],   # the Next.js dev server
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)


class AskRequest(BaseModel):
    question: str


@app.get("/health")
async def health(db: Session = Depends(get_db)):
    db.execute(text("SELECT 1"))
    return {"status": "ok", "database": "connected"}


@app.post("/api/echo")
async def echo(body: AskRequest):
    if not body.question.strip():
        raise HTTPException(status_code=400, detail="question cannot be empty")
    return {"you_asked": body.question, "length": len(body.question)}

from dotenv import load_dotenv

from app.config import settings
from app.db.models import Base

load_dotenv()

config = context.config
config.set_main_option("sqlalchemy.url", settings.database_url)

target_metadata = Base.metadata   # this is what enables --autogenerate

def upgrade() -> None:
    op.execute("""
        ALTER TABLE chunks
        ADD COLUMN tsv tsvector
        GENERATED ALWAYS AS (to_tsvector('english', content)) STORED
    """)
    op.execute("CREATE INDEX ix_chunks_tsv ON chunks USING gin (tsv)")


def downgrade() -> None:
    op.execute("DROP INDEX IF EXISTS ix_chunks_tsv")
    op.execute("ALTER TABLE chunks DROP COLUMN IF EXISTS tsv")

import uuid
from datetime import date, datetime
from typing import Literal

from pydantic import BaseModel, ConfigDict


class DocumentOut(BaseModel):
    model_config = ConfigDict(from_attributes=True)

    id: uuid.UUID
    title: str
    doc_type: Literal["ruleset", "transcript"]
    status: Literal["pending", "processing", "ready", "failed"]
    error_message: str | None = None
    game_system: str | None = None
    campaign: str | None = None
    session_date: date | None = None
    page_count: int | None = None
    created_at: datetime
    # note: no embedding, no storage_path, no user_id


class DocumentCreated(BaseModel):
    id: uuid.UUID
    status: str


class Citation(BaseModel):
    n: int
    chunk_id: uuid.UUID
    document_title: str
    page_from: int | None
    page_to: int | None
    heading_path: str | None


class AnswerOut(BaseModel):
    answer: str
    citations: list[Citation]
    verified: bool

import uuid
from pathlib import Path

from fastapi import (APIRouter, Depends, File, Form, HTTPException, UploadFile,
                     status)
from sqlalchemy.orm import Session

from app.db.models import Document
from app.main import get_db
from app.schemas import DocumentCreated, DocumentOut

router = APIRouter(prefix="/api/documents", tags=["documents"])

# Replaced by the real authenticated user in Lesson 33
DEV_USER = uuid.UUID("00000000-0000-0000-0000-000000000001")
UPLOAD_DIR = Path("data/uploads")


@router.post("", status_code=status.HTTP_202_ACCEPTED,
             response_model=DocumentCreated)
async def upload(
    file: UploadFile = File(...),
    title: str = Form(...),
    doc_type: str = Form(...),
    db: Session = Depends(get_db),
):
    if not file.filename.lower().endswith(".pdf"):
        raise HTTPException(400, "only PDF files are supported")

    UPLOAD_DIR.mkdir(parents=True, exist_ok=True)
    doc_id = uuid.uuid4()
    path = UPLOAD_DIR / f"{doc_id}.pdf"
    path.write_bytes(await file.read())

    doc = Document(id=doc_id, user_id=DEV_USER, title=title,
                   doc_type=doc_type, storage_path=str(path), status="pending")
    db.add(doc)
    db.commit()

    # SEAM: in Lesson 37 this becomes `await redis.enqueue_job("ingest", doc_id)`
    from app.ingestion.pipeline import ingest_document
    ingest_document(doc_id)

    return DocumentCreated(id=doc_id, status="pending")


@router.get("", response_model=list[DocumentOut])
async def list_documents(db: Session = Depends(get_db)):
    return (db.query(Document)
              .filter(Document.user_id == DEV_USER)
              .order_by(Document.created_at.desc())
              .all())


@router.get("/{doc_id}", response_model=DocumentOut)
async def get_document(doc_id: uuid.UUID, db: Session = Depends(get_db)):
    doc = (db.query(Document)
             .filter(Document.id == doc_id, Document.user_id == DEV_USER)
             .first())
    if not doc:
        raise HTTPException(404, "document not found")
    return doc


@router.delete("/{doc_id}", status_code=status.HTTP_204_NO_CONTENT)
async def delete_document(doc_id: uuid.UUID, db: Session = Depends(get_db)):
    doc = (db.query(Document)
             .filter(Document.id == doc_id, Document.user_id == DEV_USER)
             .first())
    if not doc:
        raise HTTPException(404, "document not found")
    Path(doc.storage_path).unlink(missing_ok=True)
    db.delete(doc)          # chunks cascade
    db.commit()

import json
import uuid

from fastapi import APIRouter, Request
from litellm import completion
from pydantic import BaseModel
from sse_starlette.sse import EventSourceResponse

from app.config import settings
from app.generation.answer import SYSTEM, verify_citations
from app.generation.context import build_context
from app.retrieval.rerank import rerank
from app.retrieval.search import hybrid_search

router = APIRouter(prefix="/api", tags=["chat"])
DEV_USER = uuid.UUID("00000000-0000-0000-0000-000000000001")


class ChatRequest(BaseModel):
    question: str


@router.post("/chat")
async def chat(body: ChatRequest, request: Request):

    async def event_stream():
        candidates = hybrid_search(DEV_USER, body.question)
        top = rerank(body.question, candidates)
        context, used = build_context(top)

        stream = completion(
            model=settings.chat_model,
            temperature=settings.temperature,
            stream=True,
            messages=[
                {"role": "system", "content": SYSTEM},
                {"role": "user",
                 "content": f"{context}\n\nQuestion: {body.question}"},
            ],
        )

        parts: list[str] = []
        for chunk in stream:
            if await request.is_disconnected():
                return                       # user left; stop paying for tokens
            delta = chunk.choices[0].delta.content
            if delta:
                parts.append(delta)
                yield {"event": "token", "data": delta}

        answer = "".join(parts)
        citations = [
            {"n": i, "chunk_id": str(c["id"]), "document_title": c["title"],
             "page_from": c["page_from"], "page_to": c["page_to"],
             "heading_path": c["heading_path"]}
            for i, c in enumerate(used, 1)
        ]
        yield {
            "event": "citations",
            "data": json.dumps({
                "citations": citations,
                "verified": not verify_citations(answer, len(used)),
            }),
        }

    return EventSourceResponse(event_stream())

SUPABASE_URL=https://yourproject.supabase.co
SUPABASE_ANON_KEY=eyJ...
SUPABASE_SERVICE_KEY=eyJ...          # server only -- never send to a browser
SUPABASE_JWT_SECRET=your-jwt-secret  # Settings > API > JWT Settings
SUPABASE_DATABASE_URL=postgresql+psycopg://postgres.xxx:pass@...pooler.supabase.com:6543/postgres

import os
import uuid

import jwt
from fastapi import Depends, HTTPException, status
from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer

security = HTTPBearer()
JWT_SECRET = os.environ["SUPABASE_JWT_SECRET"]


async def get_current_user(
    creds: HTTPAuthorizationCredentials = Depends(security),
) -> uuid.UUID:
    try:
        payload = jwt.decode(
            creds.credentials,
            JWT_SECRET,
            algorithms=["HS256"],
            audience="authenticated",
        )
    except jwt.ExpiredSignatureError:
        raise HTTPException(status.HTTP_401_UNAUTHORIZED, "token expired")
    except jwt.InvalidTokenError as exc:
        raise HTTPException(status.HTTP_401_UNAUTHORIZED, f"invalid token: {exc}")

    sub = payload.get("sub")
    if not sub:
        raise HTTPException(status.HTTP_401_UNAUTHORIZED, "token has no subject")
    return uuid.UUID(sub)

@router.get("", response_model=list[DocumentOut])
async def list_documents(
    user_id: uuid.UUID = Depends(get_current_user),
    db: Session = Depends(get_db),
):
    return (db.query(Document)
              .filter(Document.user_id == user_id)
              .order_by(Document.created_at.desc())
              .all())

-- A role that RLS actually applies to
CREATE ROLE loremaster_app LOGIN PASSWORD 'apppassword';
GRANT CONNECT ON DATABASE loremaster TO loremaster_app;
GRANT USAGE ON SCHEMA public TO loremaster_app;
GRANT SELECT, INSERT, UPDATE, DELETE ON ALL TABLES IN SCHEMA public
  TO loremaster_app;
ALTER DEFAULT PRIVILEGES IN SCHEMA public
  GRANT SELECT, INSERT, UPDATE, DELETE ON TABLES TO loremaster_app;

ALTER TABLE documents     ENABLE ROW LEVEL SECURITY;
ALTER TABLE chunks        ENABLE ROW LEVEL SECURITY;
ALTER TABLE conversations ENABLE ROW LEVEL SECURITY;
ALTER TABLE messages      ENABLE ROW LEVEL SECURITY;

CREATE POLICY documents_owner ON documents FOR ALL
  USING (user_id = current_setting('app.current_user_id', true)::uuid);

CREATE POLICY chunks_owner ON chunks FOR ALL
  USING (user_id = current_setting('app.current_user_id', true)::uuid);

CREATE POLICY conversations_owner ON conversations FOR ALL
  USING (user_id = current_setting('app.current_user_id', true)::uuid);

CREATE POLICY messages_owner ON messages FOR ALL
  USING (conversation_id IN (
    SELECT id FROM conversations
    WHERE user_id = current_setting('app.current_user_id', true)::uuid
  ));

import uuid

from fastapi import Depends
from sqlalchemy import text

from app.auth import get_current_user
from app.db.models import SessionLocal


def get_db_for_user(user_id: uuid.UUID = Depends(get_current_user)):
    """One transaction per request, with the RLS identity set on it."""
    db = SessionLocal()
    try:
        db.execute(
            text("SET LOCAL app.current_user_id = :uid"), {"uid": str(user_id)}
        )
        yield db
        db.commit()
    except Exception:
        db.rollback()
        raise
    finally:
        db.close()

import os

import pytest
from fastapi.testclient import TestClient

from app.main import app

client = TestClient(app)
A = {"Authorization": f"Bearer {os.environ['TOKEN_USER_A']}"}
B = {"Authorization": f"Bearer {os.environ['TOKEN_USER_B']}"}
DOC_A = os.environ["DOC_ID_USER_A"]


def test_b_cannot_read_a_document():
    assert client.get(f"/api/documents/{DOC_A}", headers=B).status_code == 404


def test_b_cannot_delete_a_document():
    assert client.delete(f"/api/documents/{DOC_A}", headers=B).status_code == 404
    assert client.get(f"/api/documents/{DOC_A}", headers=A).status_code == 200


def test_b_retrieval_cannot_see_a_content():
    r = client.post("/api/chat", headers=B,
                    json={"question": "what are the harm levels"})
    assert "couldn't find that" in r.text.lower()


def test_no_token_is_rejected():
    for method, path in [("get", "/api/documents"),
                         ("post", "/api/chat"),
                         ("get", f"/api/documents/{DOC_A}")]:
        assert getattr(client, method)(path).status_code == 401


def test_forged_token_is_rejected():
    forged = os.environ["TOKEN_USER_B"][:-4] + "AAAA"
    r = client.get("/api/documents", headers={"Authorization": f"Bearer {forged}"})
    assert r.status_code == 401

  redis:
    image: redis:7-alpine
    container_name: loremaster-redis
    ports:
      - "6379:6379"
    volumes:
      - redisdata:/data

volumes:
  pgdata:
  redisdata:

import asyncio

from arq.connections import RedisSettings

REDIS = RedisSettings(host="localhost", port=6379)


async def hello(ctx, name: str):
    print(f"worker: starting job for {name}")
    await asyncio.sleep(5)
    print(f"worker: done with {name}")
    return f"hello {name}"


class WorkerSettings:
    functions = [hello]
    redis_settings = REDIS
    max_jobs = 2              # embedding calls are the bottleneck, not CPU
    job_timeout = 900         # 15 minutes for a big book

from arq import create_pool

from app.worker import REDIS

redis = await create_pool(REDIS)
job = await redis.enqueue_job("hello", "Anna")
print(job.job_id)

import uuid
from pathlib import Path

from arq.connections import RedisSettings

from app.db.models import Document, SessionLocal
from app.ingestion.chunk import chunk_document
from app.ingestion.embed import embed_chunks
from app.ingestion.extract import extract_pages, is_scanned
from app.ingestion.normalise import normalise
from app.db.store import store_chunks

REDIS = RedisSettings(host="localhost", port=6379)


def _set_status(doc_id, status, error=None, page_count=None):
    with SessionLocal() as db:
        doc = db.get(Document, doc_id)
        doc.status = status
        if error is not None:
            doc.error_message = error
        if page_count is not None:
            doc.page_count = page_count
        db.commit()


async def ingest_document(ctx, document_id: str):
    doc_id = uuid.UUID(document_id)

    with SessionLocal() as db:
        doc = db.get(Document, doc_id)
        path = Path(doc.storage_path)
        user_id = doc.user_id
        session_date = doc.session_date

    try:
        _set_status(doc_id, "processing")

        # Design doc section 6, step 0 -- reject scans with a clear message
        if is_scanned(path):
            _set_status(
                doc_id, "failed",
                error="This PDF contains images rather than selectable text "
                      "(it looks like a scan). Re-export it from the original "
                      "source, or run OCR before uploading.",
            )
            return

        pages = normalise(extract_pages(path))
        chunks = chunk_document(pages)
        if not chunks:
            _set_status(doc_id, "failed",
                        error="No readable text was found in this document.")
            return

        date_str = session_date.isoformat() if session_date else None
        vectors = embed_chunks(chunks, session_date=date_str)

        store_chunks(doc_id, user_id, chunks, vectors)   # one transaction
        _set_status(doc_id, "ready", page_count=len(pages))
        print(f"ingested {doc_id}: {len(chunks)} chunks from {len(pages)} pages")

    except Exception as exc:
        _set_status(doc_id, "failed", error=f"Ingestion failed: {exc}")
        raise            # re-raise so ARQ logs it and retry policy applies


class WorkerSettings:
    functions = [ingest_document]
    redis_settings = REDIS
    max_jobs = 2
    job_timeout = 900

import os
import uuid
from pathlib import Path

from supabase import Client, create_client

BUCKET = "documents"

_client: Client = create_client(
    os.environ["SUPABASE_URL"],
    os.environ["SUPABASE_SERVICE_KEY"],   # server-side only
)


def upload_pdf(user_id: uuid.UUID, doc_id: uuid.UUID, data: bytes) -> str:
    path = f"{user_id}/{doc_id}.pdf"
    _client.storage.from_(BUCKET).upload(
        path, data, {"content-type": "application/pdf"}
    )
    return path


def download_to_temp(storage_path: str) -> Path:
    data = _client.storage.from_(BUCKET).download(storage_path)
    tmp = Path("/tmp") / Path(storage_path).name
    tmp.write_bytes(data)
    return tmp


def signed_url(storage_path: str, expires_seconds: int = 300) -> str:
    res = _client.storage.from_(BUCKET).create_signed_url(
        storage_path, expires_seconds
    )
    return res["signedURL"]


def delete_object(storage_path: str) -> None:
    _client.storage.from_(BUCKET).remove([storage_path])

curl -o- https://raw.githubusercontent.com/nvm-sh/nvm/v0.40.1/install.sh | bash
source ~/.bashrc
nvm install --lts
node --version
npm --version

// Python: def greet(name: str) -> str: return f"Hi {name}"
function greet(name: string): string {
  return `Hi ${name}`;             // backticks, ${} instead of f""
}

// Python: greet = lambda name: f"Hi {name}"
const greet2 = (name: string): string => `Hi ${name}`;

// Python: items = [1, 2, 3];  doubled = [x * 2 for x in items]
const items: number[] = [1, 2, 3];
const doubled = items.map((x) => x * 2);
const evens = items.filter((x) => x % 2 === 0);

// Python: d = {"a": 1};  d["a"]
const d: Record<string, number> = { a: 1 };
console.log(d.a, d["a"]);          // both work

// Python: class + dataclass  ->  interface (types only, no runtime cost)
interface Document {
  id: string;
  title: string;
  status: "pending" | "processing" | "ready" | "failed";
  errorMessage?: string;           // ? means optional
}

// Python: async def get(): r = await client.get(url); return r.json()
async function getDocs(): Promise<Document[]> {
  const res = await fetch("http://localhost:8000/api/documents");
  if (!res.ok) throw new Error(`HTTP ${res.status}`);
  return res.json();
}

import { createBrowserClient } from "@supabase/ssr";

export const supabase = createBrowserClient(
  process.env.NEXT_PUBLIC_SUPABASE_URL!,
  process.env.NEXT_PUBLIC_SUPABASE_ANON_KEY!,
);

"use client";

import { useState } from "react";
import { useRouter } from "next/navigation";
import { supabase } from "@/lib/supabase";

export default function LoginPage() {
  const [email, setEmail] = useState("");
  const [password, setPassword] = useState("");
  const [error, setError] = useState<string | null>(null);
  const router = useRouter();

  async function handleLogin(e: React.FormEvent) {
    e.preventDefault();
    setError(null);
    const { error } = await supabase.auth.signInWithPassword({
      email,
      password,
    });
    if (error) setError(error.message);
    else router.push("/documents");
  }

  return (
    <form onSubmit={handleLogin} className="mx-auto mt-24 max-w-sm space-y-4">
      <h1 className="text-2xl font-semibold">Sign in to Loremaster</h1>
      <input
        type="email"
        value={email}
        onChange={(e) => setEmail(e.target.value)}
        placeholder="you@example.com"
        className="w-full rounded border px-3 py-2"
      />
      <input
        type="password"
        value={password}
        onChange={(e) => setPassword(e.target.value)}
        placeholder="password"
        className="w-full rounded border px-3 py-2"
      />
      {error && <p className="text-sm text-red-600">{error}</p>}
      <button className="w-full rounded bg-black px-3 py-2 text-white">
        Sign in
      </button>
      <button
        type="button"
        onClick={() => supabase.auth.signInWithOAuth({ provider: "google" })}
        className="w-full rounded border px-3 py-2"
      >
        Continue with Google
      </button>
    </form>
  );
}

import { supabase } from "./supabase";

const BASE = process.env.NEXT_PUBLIC_API_URL ?? "http://localhost:8000";

export async function apiFetch(path: string, init: RequestInit = {}) {
  const { data } = await supabase.auth.getSession();
  const token = data.session?.access_token;

  const res = await fetch(`${BASE}${path}`, {
    ...init,
    headers: {
      ...init.headers,
      ...(token ? { Authorization: `Bearer ${token}` } : {}),
    },
  });

  if (!res.ok) throw new Error(`${res.status} ${await res.text()}`);
  return res;
}

"use client";

import { useCallback, useEffect, useState } from "react";
import { apiFetch } from "@/lib/api";

type Doc = {
  id: string;
  title: string;
  doc_type: string;
  status: "pending" | "processing" | "ready" | "failed";
  error_message: string | null;
  page_count: number | null;
};

const TERMINAL = new Set(["ready", "failed"]);

export default function DocumentsPage() {
  const [docs, setDocs] = useState<Doc[]>([]);
  const [busy, setBusy] = useState(false);

  const load = useCallback(async () => {
    const res = await apiFetch("/api/documents");
    setDocs(await res.json());
  }, []);

  useEffect(() => {
    load();
  }, [load]);

  // Poll only while something is still in flight, and always clean up.
  useEffect(() => {
    const pending = docs.some((d) => !TERMINAL.has(d.status));
    if (!pending) return;
    const timer = setInterval(load, 2000);
    return () => clearInterval(timer);
  }, [docs, load]);

  async function handleUpload(e: React.FormEvent<HTMLFormElement>) {
    e.preventDefault();
    setBusy(true);
    try {
      const form = new FormData(e.currentTarget);
      await apiFetch("/api/documents", { method: "POST", body: form });
      e.currentTarget.reset();
      await load();
    } finally {
      setBusy(false);
    }
  }

  return (
    <main className="mx-auto max-w-3xl space-y-8 p-8">
      <h1 className="text-2xl font-semibold">Your materials</h1>

      <form onSubmit={handleUpload} className="space-y-3 rounded border p-4">
        <input name="title" placeholder="Title" required
               className="w-full rounded border px-3 py-2" />
        <select name="doc_type" className="w-full rounded border px-3 py-2">
          <option value="ruleset">Ruleset</option>
          <option value="transcript">Session transcript</option>
        </select>
        <input type="file" name="file" accept=".pdf" required />
        <button disabled={busy}
                className="rounded bg-black px-4 py-2 text-white disabled:opacity-50">
          {busy ? "Uploading..." : "Upload"}
        </button>
      </form>

      <ul className="space-y-2">
        {docs.map((d) => (
          <li key={d.id} className="rounded border p-3">
            <div className="flex items-center justify-between">
              <span className="font-medium">{d.title}</span>
              <StatusBadge status={d.status} />
            </div>
            {d.status === "failed" && d.error_message && (
              <p className="mt-2 text-sm text-red-700">{d.error_message}</p>
            )}
            {d.status === "ready" && (
              <p className="mt-1 text-sm text-gray-500">
                {d.page_count} pages -- ready to search
              </p>
            )}
          </li>
        ))}
      </ul>
    </main>
  );
}

function StatusBadge({ status }: { status: Doc["status"] }) {
  const styles: Record<Doc["status"], string> = {
    pending: "bg-gray-100 text-gray-700",
    processing: "bg-blue-100 text-blue-700",
    ready: "bg-green-100 text-green-700",
    failed: "bg-red-100 text-red-700",
  };
  return (
    <span className={`rounded px-2 py-0.5 text-xs ${styles[status]}`}>
      {status}
    </span>
  );
}

"use client";

import { useState } from "react";
import { supabase } from "@/lib/supabase";

type Citation = {
  n: number;
  chunk_id: string;
  document_title: string;
  page_from: number | null;
  heading_path: string | null;
};

const BASE = process.env.NEXT_PUBLIC_API_URL ?? "http://localhost:8000";

export default function ChatPage() {
  const [question, setQuestion] = useState("");
  const [answer, setAnswer] = useState("");
  const [citations, setCitations] = useState<Citation[]>([]);
  const [verified, setVerified] = useState(true);
  const [phase, setPhase] = useState<"idle" | "searching" | "streaming">("idle");

  async function ask(e: React.FormEvent) {
    e.preventDefault();
    setAnswer("");
    setCitations([]);
    setVerified(true);
    setPhase("searching");

    const { data } = await supabase.auth.getSession();
    const res = await fetch(`${BASE}/api/chat`, {
      method: "POST",
      headers: {
        "Content-Type": "application/json",
        Authorization: `Bearer ${data.session?.access_token}`,
      },
      body: JSON.stringify({ question }),
    });

    const reader = res.body!.getReader();
    const decoder = new TextDecoder();
    let buffer = "";

    while (true) {
      const { done, value } = await reader.read();
      if (done) break;

      buffer += decoder.decode(value, { stream: true });
      const frames = buffer.split("\n\n");
      buffer = frames.pop() ?? "";        // keep the incomplete tail

      for (const frame of frames) {
        let event = "message";
        let payload = "";
        for (const line of frame.split("\n")) {
          if (line.startsWith("event:")) event = line.slice(6).trim();
          else if (line.startsWith("data:")) payload += line.slice(5).trimStart();
        }

        if (event === "token") {
          setPhase("streaming");
          setAnswer((prev) => prev + payload);
        } else if (event === "citations") {
          const parsed = JSON.parse(payload);
          setCitations(parsed.citations);
          setVerified(parsed.verified);
        }
      }
    }
    setPhase("idle");
  }

  return (
    <main className="mx-auto max-w-2xl space-y-6 p-8">
      <form onSubmit={ask} className="flex gap-2">
        <input
          value={question}
          onChange={(e) => setQuestion(e.target.value)}
          placeholder="Ask about your materials..."
          className="flex-1 rounded border px-3 py-2"
        />
        <button className="rounded bg-black px-4 py-2 text-white">Ask</button>
      </form>

      {phase === "searching" && (
        <p className="text-sm text-gray-500">Searching your materials...</p>
      )}

      {answer && (
        <div className="space-y-4">
          <p className="whitespace-pre-wrap leading-relaxed">{answer}</p>

          {!verified && (
            <p className="rounded bg-amber-50 p-2 text-sm text-amber-800">
              Unverified: this answer referenced a source that could not be
              matched. Check the citations carefully.
            </p>
          )}

          <ul className="space-y-1 border-t pt-3 text-sm text-gray-600">
            {citations.map((c) => (
              <li key={c.n}>
                [{c.n}] {c.document_title}
                {c.page_from ? `, p.${c.page_from}` : ""}
                {c.heading_path ? ` -- ${c.heading_path}` : ""}
              </li>
            ))}
          </ul>
        </div>
      )}
    </main>
  );
}

"use client";

import { useState } from "react";
import { apiFetch } from "@/lib/api";

type Citation = {
  n: number;
  chunk_id: string;
  document_title: string;
  page_from: number | null;
  heading_path: string | null;
};

type Source = { content: string } & Citation;

export function AnswerWithCitations({
  answer,
  citations,
}: {
  answer: string;
  citations: Citation[];
}) {
  const [open, setOpen] = useState<Source | null>(null);

  async function show(c: Citation) {
    const res = await apiFetch(`/api/chunks/${c.chunk_id}`);
    const chunk = await res.json();
    setOpen({ ...c, content: chunk.content });
  }

  // Split on [n] and keep the delimiters, so we can map them to buttons.
  const parts = answer.split(/(\[\d+\])/g);

  return (
    <div className="flex gap-6">
      <p className="flex-1 whitespace-pre-wrap leading-relaxed">
        {parts.map((part, i) => {
          const match = part.match(/^\[(\d+)\]$/);
          if (!match) return <span key={i}>{part}</span>;

          const n = Number(match[1]);
          const citation = citations.find((c) => c.n === n);
          if (!citation) return <span key={i}>{part}</span>;   // no silent break

          return (
            <button
              key={i}
              onClick={() => show(citation)}
              title={`${citation.document_title}, p.${citation.page_from}`}
              className="mx-0.5 rounded bg-blue-100 px-1 text-xs font-medium
                         text-blue-800 hover:bg-blue-200"
            >
              {n}
            </button>
          );
        })}
      </p>

      {open && (
        <aside className="w-80 shrink-0 rounded border bg-gray-50 p-4">
          <div className="mb-2 flex items-start justify-between">
            <div className="text-xs text-gray-500">
              <div className="font-medium text-gray-800">
                {open.document_title}
              </div>
              {open.page_from && <div>p.{open.page_from}</div>}
              {open.heading_path && <div>{open.heading_path}</div>}
            </div>
            <button onClick={() => setOpen(null)} className="text-gray-400">
              close
            </button>
          </div>
          <p className="whitespace-pre-wrap text-sm leading-relaxed">
            {open.content}
          </p>
        </aside>
      )}
    </div>
  );
}

import json
import uuid
from datetime import date
from typing import Literal

from litellm import completion
from pydantic import BaseModel, ValidationError

from app.config import settings

PROMPT = """Convert the user's question into a structured search query.

Today's date is {today}.

The user's documents:
{documents}

Return ONLY JSON of this shape:
{{
  "search_text": "the words worth searching for, proper nouns kept",
  "filters": {{
    "doc_type": "ruleset" | "transcript" | null,
    "document_ids": ["uuid", ...],
    "session_date_from": "YYYY-MM-DD" | null,
    "session_date_to": "YYYY-MM-DD" | null
  }},
  "scope": "narrow" | "broad"
}}

"scope" is "broad" when the question asks for a complete list, an overview,
or every instance of something ("what are all the...", "list the...",
"summarise..."). Otherwise it is "narrow".

Question: {question}"""


class Filters(BaseModel):
    doc_type: Literal["ruleset", "transcript"] | None = None
    document_ids: list[uuid.UUID] = []
    session_date_from: date | None = None
    session_date_to: date | None = None


class StructuredQuery(BaseModel):
    search_text: str
    filters: Filters = Filters()
    scope: Literal["narrow", "broad"] = "narrow"


def understand(question: str, documents: list[dict],
               max_retries: int = 2) -> StructuredQuery:
    doc_lines = "\n".join(
        f'- {d["id"]} | "{d["title"]}" | {d["doc_type"]}'
        f'{" | session " + str(d["session_date"]) if d.get("session_date") else ""}'
        for d in documents
    ) or "(none)"

    prompt = PROMPT.format(today=date.today().isoformat(),
                           documents=doc_lines, question=question)

    for _ in range(max_retries):
        raw = completion(
            model=settings.utility_model,
            temperature=0.0,
            response_format={"type": "json_object"},
            messages=[{"role": "user", "content": prompt}],
        ).choices[0].message.content
        try:
            return StructuredQuery.model_validate(json.loads(raw))
        except (json.JSONDecodeError, ValidationError):
            continue

    # Never fail the user's question because the parser misbehaved.
    return StructuredQuery(search_text=question)

def _filter_sql(f: Filters) -> tuple[str, dict]:
    clauses, params = [], {}
    if f.doc_type:
        clauses.append("AND d.doc_type = :doc_type")
        params["doc_type"] = f.doc_type
    if f.document_ids:
        clauses.append("AND d.id = ANY(:doc_ids)")
        params["doc_ids"] = [str(i) for i in f.document_ids]
    if f.session_date_from:
        clauses.append("AND d.session_date >= :date_from")
        params["date_from"] = f.session_date_from
    if f.session_date_to:
        clauses.append("AND d.session_date <= :date_to")
        params["date_to"] = f.session_date_to
    return " ".join(clauses), params

from litellm import completion

from app.config import settings

PROMPT = """Rewrite the user's latest question so it stands alone, resolving
every pronoun and implicit reference using the conversation so far.
Keep proper nouns exactly as written. Change nothing else.
If the question already stands alone, return it unchanged.
Return ONLY the rewritten question.

Conversation so far:
{history}

Latest question: {question}"""


def rewrite_followup(question: str, history: list[dict],
                     max_turns: int = 4) -> str:
    if not history:
        return question

    recent = history[-max_turns:]
    rendered = "\n".join(f'{m["role"]}: {m["content"][:400]}' for m in recent)

    rewritten = completion(
        model=settings.utility_model,
        temperature=0.0,
        messages=[{"role": "user",
                   "content": PROMPT.format(history=rendered,
                                            question=question)}],
    ).choices[0].message.content.strip()

    # Guard against a model that decides to answer instead of rewrite.
    return rewritten if 0 < len(rewritten) < 500 else question

import uuid

from sqlalchemy import text

from app.db.models import SessionLocal

NEIGHBOURS = text("""
    WITH hits AS (
        SELECT ordinal, heading_path
        FROM chunks
        WHERE user_id = :uid AND document_id = :doc AND id = ANY(:ids)
    )
    SELECT c.id, c.content, c.heading_path, c.ordinal,
           c.page_from, c.page_to, d.title
    FROM chunks c
    JOIN documents d ON d.id = c.document_id
    WHERE c.user_id = :uid
      AND c.document_id = :doc
      AND (
        -- everything under the same heading section as a hit
        EXISTS (
          SELECT 1 FROM hits h
          WHERE h.heading_path IS NOT NULL
            AND c.heading_path LIKE h.heading_path || '%'
        )
        -- plus a window of two chunks either side of each hit
        OR EXISTS (
          SELECT 1 FROM hits h
          WHERE c.ordinal BETWEEN h.ordinal - 2 AND h.ordinal + 2
        )
      )
    ORDER BY c.ordinal
""")


def expand_to_neighbours(user_id: uuid.UUID, document_id: uuid.UUID,
                         hit_ids: list[uuid.UUID]) -> list[dict]:
    """Design doc section 7.3: a contiguous span beats scattered fragments."""
    with SessionLocal() as session:
        rows = session.execute(NEIGHBOURS, {
            "uid": str(user_id),
            "doc": str(document_id),
            "ids": [str(i) for i in hit_ids],
        }).mappings().all()
    return [dict(r) for r in rows]

from app.retrieval.expand import expand_to_neighbours
from app.retrieval.rerank import rerank
from app.retrieval.search import hybrid_search


def retrieve(user_id, sq, documents):
    if sq.scope == "broad" and sq.filters.document_ids:
        doc_id = sq.filters.document_ids[0]
        seeds = hybrid_search(user_id, sq.search_text, limit=5)
        seeds = [s for s in seeds if str(s["document_id"]) == str(doc_id)]
        if seeds:
            return expand_to_neighbours(
                user_id, doc_id, [s["id"] for s in seeds]
            )

    candidates = hybrid_search(user_id, sq.search_text)
    return rerank(sq.search_text, candidates)

from collections.abc import AsyncIterator, Iterator
from typing import Protocol

import litellm

from app.config import settings

Message = dict[str, str]


class LLMProvider(Protocol):
    def complete(self, messages: list[Message], *, model: str | None = None,
                 temperature: float = 0.1,
                 stream: bool = False) -> str | Iterator[str]: ...

    def embed(self, texts: list[str]) -> list[list[float]]: ...


class LiteLLMProvider:
    """The only file in the codebase that imports litellm."""

    def complete(self, messages, *, model=None, temperature=0.1,
                 stream=False, **kwargs):
        model = model or settings.chat_model
        resp = litellm.completion(model=model, messages=messages,
                                  temperature=temperature, stream=stream,
                                  **kwargs)
        if stream:
            def gen():
                for chunk in resp:
                    delta = chunk.choices[0].delta.content
                    if delta:
                        yield delta
            return gen()

        usage = resp.usage
        print(f"[llm] {model} in={usage.prompt_tokens} "
              f"out={usage.completion_tokens}")
        return resp.choices[0].message.content

    def embed(self, texts):
        resp = litellm.embedding(model=settings.embedding_model, input=texts)
        print(f"[embed] {settings.embedding_model} n={len(texts)}")
        return [d["embedding"] for d in resp.data]


llm: LLMProvider = LiteLLMProvider()

curl -fsSL https://ollama.com/install.sh | sh
ollama pull llama3.1
ollama pull nomic-embed-text
ollama list
ollama run llama3.1 "say hello in five words"

# then, in .env, change ONE line:
# CHAT_MODEL=ollama/llama3.1

| model                    | recall@8 | groundedness | refusal acc | median latency |
|--------------------------|----------|--------------|-------------|----------------|
| openai/gpt-4o-mini       |          |              |             |                |
| anthropic/claude-sonnet  |          |              |             |                |
| ollama/llama3.1 (local)  |          |              |             |                |

Note: recall@8 should be identical across chat models -- retrieval does not
depend on the chat model. If it moved, something is wrong with your eval,
or your utility model changed too and altered query parsing.

FROM python:3.12-slim

RUN apt-get update && apt-get install -y --no-install-recommends \
    build-essential libpq-dev && rm -rf /var/lib/apt/lists/*

WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY . .

CMD ["sh", "-c", "uvicorn app.main:app --host 0.0.0.0 --port ${PORT:-8000}"]

# Backend service env vars
DATABASE_URL=            # Supabase POOLER url, not the direct one
REDIS_URL=               # from the Railway Redis service
OPENAI_API_KEY=
SUPABASE_URL=
SUPABASE_SERVICE_KEY=
SUPABASE_JWT_SECRET=
CHAT_MODEL=
UTILITY_MODEL=
EMBEDDING_MODEL=
EMBEDDING_DIM=

# Worker service: the SAME list. It is a separate process.

# Frontend (Vercel)
NEXT_PUBLIC_API_URL=
NEXT_PUBLIC_SUPABASE_URL=
NEXT_PUBLIC_SUPABASE_ANON_KEY=

# And do not forget:
#  - CORS allow_origins must include the Vercel domain
#  - Supabase Auth redirect URLs must include the Vercel domain
#  - alembic upgrade head against the production database

from slowapi import Limiter
from slowapi.errors import RateLimitExceeded
from slowapi.util import get_remote_address

from app.auth import get_current_user


def user_key(request):
    """Limit per authenticated user, falling back to IP."""
    return request.headers.get("authorization", get_remote_address(request))


limiter = Limiter(key_func=user_key)
app.state.limiter = limiter


@router.post("")
@limiter.limit("10/hour")
async def upload(request: Request, ...):
    ...


@router.post("/chat")
@limiter.limit("30/hour")
async def chat(request: Request, ...):
    ...

## Where your data goes

Loremaster answers only from the materials you upload. To do that, it sends
the relevant passages of your documents to a language model provider.

**Default configuration** uses OpenAI's API. Your uploaded text is sent to
OpenAI when you ask a question, and when your document is first indexed.
OpenAI's API terms state that API data is not used to train their models and
is retained only briefly for abuse monitoring. Anthropic offers equivalent
terms.

**If that is not acceptable to you**, Loremaster supports running entirely on
your own hardware with Ollama. Set `CHAT_MODEL=ollama/llama3.1` and
`EMBEDDING_MODEL=ollama/nomic-embed-text` and no text leaves your machine.
Answer quality is lower; see the comparison table above.

**Isolation.** Your documents are visible only to your account. This is
enforced in three independent layers: every query filters by user, PostgreSQL
Row-Level Security rejects rows that are not yours regardless of the query,
and there is no shared cache of retrieved passages or embeddings between
users. The isolation tests in `tests/test_isolation.py` verify this.

**Scanned PDFs are rejected**, deliberately. A scan has no text layer, so a
silently-ingested scan would look ready and then fail to answer anything.