added search

.env.example

@@ -38,12 +38,9 @@ LLM_TEMPERATURE=1.0
 LLM_MAX_TOKENS=16000
 LLM_REASONING_EFFORT=low
 
-# ---- Embeddings (separate provider: Z.ai's international API has no embedding model)
-# Must support a `dimensions` parameter so the pgvector column stays EMBEDDING_DIM wide.
-EMBEDDING_MODEL=openai/text-embedding-3-small
-EMBEDDING_DIM=1024
-EMBEDDING_API_BASE=
-EMBEDDING_API_KEY=
+# ---- Research corpus (posts as markdown files; the agent greps and reads them, no vector index)
+CORPUS_CACHE_DIR=.corpus-cache
+RESEARCH_MAX_STEPS=10
 
 # ---- ElevenLabs (Phase 1 audio)
 ELEVENLABS_API_KEY=

.gitignore

@@ -8,3 +8,4 @@ __pycache__/
 .ruff_cache/
 out/
 renderer/build/
+.corpus-cache/

README.md

@@ -18,7 +18,7 @@ docs/       Design doc and task lists
 ## Run locally
 
 ```bash
-cp .env.example .env            # fill LLM_API_KEY (Z.ai), EMBEDDING_API_KEY, GOOGLE_CLIENT_ID, RESEND_API_KEY
+cp .env.example .env            # fill LLM_API_KEY (Z.ai), GOOGLE_CLIENT_ID, RESEND_API_KEY
 docker compose up --build
 ```
 
@@ -31,7 +31,7 @@ Without Docker:
 
 ```bash
 cd backend && python -m venv .venv && .venv/Scripts/pip install -r requirements-dev.txt
-alembic upgrade head && uvicorn app.main:app --reload     # needs Postgres (pgvector) + Redis
+alembic upgrade head && uvicorn app.main:app --reload     # needs Postgres + Redis
 arq app.worker.WorkerSettings                              # second terminal
 cd frontend && npm install && npm run dev
 cd renderer && npm install && npm start                    # or: npm run studio
@@ -44,6 +44,7 @@ cd renderer && npm install && npm start                    # or: npm run studio
 | Cloudflare R2 storage (keys, presign, uploads) | `backend/app/services/storage.py`, `backend/app/routers/storage.py` |
 | Resend email + unsubscribe + campaigns | `backend/app/services/email.py`, `backend/app/worker.py` |
 | Google + email code auth, JWT revocation | `backend/app/auth.py`, `backend/app/routers/auth.py` |
+| Research over posts as files (no embeddings) | `backend/app/corpus.py`, `backend/app/pipeline/research.py` |
 | LLM provider (swap by env) | `backend/app/llm/provider.py`, `.env` `LLM_*` |
 | Three plans, billing off | `backend/app/services/plans.py`, `BILLING_ENABLED` |
 | Blog posts | `frontend/src/content/blogPosts.ts` (`/blogs`, `/blogs/:slug`) |
@@ -61,8 +62,6 @@ LLM_MODEL=anthropic/claude-sonnet-5  LLM_API_BASE=
 LLM_MODEL=openai/gpt-5  LLM_API_BASE=
 ```
 
-Changing `EMBEDDING_DIM` needs a new migration and a re-embed.
-
 ## Rules
 
 - No em dashes in any UI, email or generated copy. `npm run check:copy` and

backend/alembic/versions/0001_initial.py

@@ -1,4 +1,4 @@
-"""initial schema: pgvector + all Phase 0 tables
+"""initial schema: all Phase 0 tables
 
 Initial revision builds from the models so schema and code start identical. Every later change
 must be a normal autogenerated revision (alembic revision --autogenerate -m "...").
@@ -10,7 +10,6 @@ Create Date: 2026-09-22
 from alembic import op
 
 import app.models  # noqa: F401
-from app.config import settings
 from app.db import Base
 
 revision = "0001_initial"
@@ -20,15 +19,7 @@ depends_on = None
 
 
 def upgrade() -> None:
-    op.execute("CREATE EXTENSION IF NOT EXISTS vector")
-    bind = op.get_bind()
-    Base.metadata.create_all(bind=bind)
-    # HNSW index for cosine search over chunk embeddings (pgvector supports HNSW up to 2000 dims).
-    if settings.embedding_dim <= 2000:
-        op.execute(
-            "CREATE INDEX IF NOT EXISTS ix_chunks_embedding_hnsw ON chunks "
-            "USING hnsw (embedding vector_cosine_ops)"
-        )
+    Base.metadata.create_all(bind=op.get_bind())
 
 
 def downgrade() -> None:

backend/app/config.py

@@ -51,12 +51,9 @@ class Settings(BaseSettings):
     # GLM-5.x always reasons (it cannot be disabled); this sets how hard. low | high | max
     llm_reasoning_effort: str = "low"
 
-    # Embeddings are a separate provider: Z.ai's international API has no embedding model.
-    # LiteLLM picks up OPENAI_API_KEY etc. from env when embedding_api_key is empty.
-    embedding_model: str = "openai/text-embedding-3-small"
-    embedding_api_base: str = ""
-    embedding_api_key: str = ""
-    embedding_dim: int = 1024
+    # Corpus: posts as markdown files. R2 is the source of truth; each process keeps a disk cache.
+    corpus_cache_dir: str = ".corpus-cache"
+    research_max_steps: int = 10
 
     # ElevenLabs
     elevenlabs_api_key: str = ""

backend/app/corpus.py

@@ -0,0 +1,265 @@
+"""A workspace's archive as a file system.
+
+Every post is one markdown file. The research agent navigates it with list/search/read tools, the
+same way a coding agent navigates a repo, so there is no vector index to build or keep in sync.
+
+Layout (paths relative to the corpus root):
+
+    INDEX.md                                       one line per post: date | path | title | headings
+    sources/{source-slug}/{date}-{post-slug}.md    front matter + clean markdown
+
+R2 is the source of truth (ws/{id}/corpus/...). Each process keeps a disk cache under
+CORPUS_CACHE_DIR and syncs it from R2 through manifest.json (path -> sha256), so the API and worker
+containers see the same files without sharing a volume.
+"""
+
+import hashlib
+import json
+import re
+import threading
+import unicodedata
+import uuid
+from dataclasses import dataclass
+from datetime import datetime
+from pathlib import Path, PurePosixPath
+
+from app.config import settings
+from app.services.storage import storage as default_storage
+
+MANIFEST = "manifest.json"
+INDEX = "INDEX.md"
+MAX_READ_LINES = 160
+MAX_SEARCH_HITS = 40
+MAX_LINE_CHARS = 240
+
+_locks: dict[uuid.UUID, threading.Lock] = {}
+
+
+def slugify(text: str, max_len: int = 60) -> str:
+    text = unicodedata.normalize("NFKD", text).encode("ascii", "ignore").decode()
+    text = re.sub(r"[^a-zA-Z0-9]+", "-", text).strip("-").lower()
+    return text[:max_len].rstrip("-") or "untitled"
+
+
+def post_path(source_slug: str, title: str, url: str, published_at: datetime | None) -> str:
+    # Prefer the URL slug (stable across title edits), fall back to the title.
+    tail = url.rstrip("/").rsplit("/", 1)[-1] if url else ""
+    slug = slugify(tail if tail and not tail.startswith("?") else title)
+    date = published_at.strftime("%Y-%m-%d") if published_at else "undated"
+    return f"sources/{source_slug}/{date}-{slug}.md"
+
+
+def render_post(*, title: str, url: str, published_at: datetime | None, source: str,
+                sections: list[tuple[str | None, str]]) -> str:
+    safe_title = title.replace('"', "'")
+    lines = [
+        "---",
+        f'title: "{safe_title}"',
+        f"url: {url}",
+        f"published: {published_at.date().isoformat() if published_at else 'unknown'}",
+        f"source: {source}",
+        "---",
+        "",
+        f"# {title}",
+        "",
+    ]
+    for heading, text in sections:
+        if heading:
+            lines += [f"## {heading}", ""]
+        for para in text.split("\n\n"):
+            lines += [para.strip(), ""]
+    return "\n".join(lines).rstrip() + "\n"
+
+
+def index_line(path: str, title: str, published_at: datetime | None, headings: list[str]) -> str:
+    date = published_at.strftime("%Y-%m-%d") if published_at else "undated"
+    heads = "; ".join(h for h in headings if h)[:300]
+    return f"{date} | {path} | {title}" + (f" | {heads}" if heads else "")
+
+
+class CorpusError(ValueError):
+    pass
+
+
+@dataclass
+class Hit:
+    path: str
+    line: int
+    text: str
+
+
+class Corpus:
+    def __init__(self, workspace_id: uuid.UUID, store=None, cache_dir: str | None = None):
+        self.workspace_id = workspace_id
+        self.store = store or default_storage
+        self.root = Path(cache_dir or settings.corpus_cache_dir) / str(workspace_id)
+        self._lock = _locks.setdefault(workspace_id, threading.Lock())
+
+    # Keys and paths
+
+    def _key(self, rel: str) -> str:
+        return f"ws/{self.workspace_id}/corpus/{rel}"
+
+    @staticmethod
+    def clean_rel(rel: str) -> str:
+        rel = rel.strip().lstrip("/").replace("\\", "/")
+        parts = PurePosixPath(rel).parts
+        if not rel or any(p in ("..", "") for p in parts) or rel.startswith("."):
+            raise CorpusError(f"Invalid path: {rel!r}")
+        return str(PurePosixPath(*parts))
+
+    def _local(self, rel: str) -> Path:
+        return self.root / self.clean_rel(rel)
+
+    # Manifest
+
+    def _read_manifest(self, remote: bool) -> dict[str, str]:
+        if remote:
+            if not self.store.exists(self._key(MANIFEST)):
+                return {}
+            return json.loads(self.store.get_bytes(self._key(MANIFEST)))
+        local = self.root / MANIFEST
+        return json.loads(local.read_text("utf-8")) if local.exists() else {}
+
+    def _write_manifest(self, manifest: dict[str, str]) -> None:
+        data = json.dumps(manifest, sort_keys=True, indent=0)
+        self.store.put_text(self._key(MANIFEST), data, "application/json")
+        self.root.mkdir(parents=True, exist_ok=True)
+        (self.root / MANIFEST).write_text(data, "utf-8")
+
+    # Writes (worker)
+
+    def write_files(self, files: dict[str, str], remove: set[str] | None = None) -> None:
+        """Write/replace files and update the manifest once. Values are file contents."""
+        with self._lock:
+            manifest = self._read_manifest(remote=True)
+            for rel, text in files.items():
+                rel = self.clean_rel(rel)
+                self.store.put_text(self._key(rel), text, "text/markdown; charset=utf-8")
+                local = self._local(rel)
+                local.parent.mkdir(parents=True, exist_ok=True)
+                local.write_text(text, "utf-8")
+                manifest[rel] = hashlib.sha256(text.encode()).hexdigest()
+            for rel in remove or set():
+                rel = self.clean_rel(rel)
+                manifest.pop(rel, None)
+                self.store.delete(self._key(rel))
+                self._local(rel).unlink(missing_ok=True)
+            self._write_manifest(manifest)
+
+    # Sync (API / any reader)
+
+    def sync(self) -> dict[str, str]:
+        """Bring the disk cache in line with R2. Downloads only files whose hash changed."""
+        with self._lock:
+            remote = self._read_manifest(remote=True)
+            local = self._read_manifest(remote=False)
+            for rel, digest in remote.items():
+                path = self._local(rel)
+                if local.get(rel) == digest and path.exists():
+                    continue
+                path.parent.mkdir(parents=True, exist_ok=True)
+                path.write_bytes(self.store.get_bytes(self._key(rel)))
+            for rel in set(local) - set(remote):
+                self._local(rel).unlink(missing_ok=True)
+            self.root.mkdir(parents=True, exist_ok=True)
+            (self.root / MANIFEST).write_text(json.dumps(remote, sort_keys=True, indent=0), "utf-8")
+            return remote
+
+    # Reads
+
+    def exists(self, rel: str) -> bool:
+        return self._local(rel).is_file()
+
+    def read_lines(self, rel: str) -> list[str]:
+        path = self._local(rel)
+        if not path.is_file():
+            raise CorpusError(f"No such file: {rel}")
+        return path.read_text("utf-8").splitlines()
+
+
+class ScopedTools:
+    """The agent's tools, restricted to an allowlist of files (a notebook's posts).
+
+    Every call is logged so the UI can stream "Searching..." / "Reading..." steps and so we can
+    verify citations only point at files the agent actually opened or matched.
+    """
+
+    def __init__(self, corpus: Corpus, allowed: dict[str, str], on_step=None):
+        # allowed: path -> title
+        self.corpus = corpus
+        self.allowed = allowed
+        self.on_step = on_step or (lambda kind, detail: None)
+
+    def _check(self, rel: str) -> str:
+        rel = Corpus.clean_rel(rel)
+        if rel != INDEX and rel not in self.allowed:
+            raise CorpusError(f"{rel} is not in this notebook. Use list_files to see what is.")
+        return rel
+
+    def list_files(self, contains: str = "") -> str:
+        """List the posts in this notebook, one per line as `date | path | title`, newest first.
+        Optionally filter to lines containing `contains` (case insensitive)."""
+        self.on_step("list", contains or "all posts")
+        index = {}
+        if self.corpus.exists(INDEX):
+            for line in self.corpus.read_lines(INDEX):
+                parts = [p.strip() for p in line.split("|")]
+                if len(parts) >= 3 and parts[1] in self.allowed:
+                    index[parts[1]] = line
+        rows = [index.get(p, f"undated | {p} | {t}") for p, t in self.allowed.items()]
+        if contains:
+            rows = [r for r in rows if contains.lower() in r.lower()]
+        rows.sort(reverse=True)
+        return "\n".join(rows[:300]) or "No matching posts."
+
+    def search(self, pattern: str, path: str = "") -> str:
+        """Search post text with a case insensitive regular expression (like grep). Use `|` for
+        alternatives, e.g. `pric(e|ing)|paid tier|subscription cost`. Optionally limit to one `path`.
+        Returns `path:line: text` matches."""
+        self.on_step("search", pattern)
+        try:
+            rx = re.compile(pattern, re.IGNORECASE)
+        except re.error:
+            rx = re.compile(re.escape(pattern), re.IGNORECASE)
+        targets = [self._check(path)] if path else list(self.allowed)
+        hits: list[Hit] = []
+        total = 0
+        for rel in targets:
+            for n, line in enumerate(self.corpus.read_lines(rel), start=1):
+                if rx.search(line):
+                    total += 1
+                    if len(hits) < MAX_SEARCH_HITS:
+                        hits.append(Hit(rel, n, line.strip()[:MAX_LINE_CHARS]))
+        if not hits:
+            return "No matches. Try synonyms or a broader pattern."
+        out = "\n".join(f"{h.path}:{h.line}: {h.text}" for h in hits)
+        if total > len(hits):
+            out += f"\n... {total - len(hits)} more matches. Narrow the pattern or pass a path."
+        return out
+
+    def read(self, path: str, start_line: int = 1, end_line: int = 0) -> str:
+        """Read a post with line numbers. Reads at most 160 lines per call; pass start_line and
+        end_line to page through long posts. Cite answers with these line numbers."""
+        rel = self._check(path)
+        lines = self.corpus.read_lines(rel)
+        start = max(1, start_line)
+        end = min(len(lines), end_line if end_line >= start else start + MAX_READ_LINES - 1)
+        end = min(end, start + MAX_READ_LINES - 1)
+        self.on_step("read", f"{self.allowed.get(rel, rel)} (lines {start} to {end})")
+        body = "\n".join(f"{n}| {lines[n - 1]}" for n in range(start, end + 1))
+        more = f"\n[{len(lines) - end} more lines]" if end < len(lines) else ""
+        return f"{rel} ({len(lines)} lines)\n{body}{more}"
+
+    def quote(self, rel: str, start: int, end: int) -> str | None:
+        """Text of a cited line range, or None if the citation does not point at real content."""
+        try:
+            rel = self._check(rel)
+            lines = self.corpus.read_lines(rel)
+        except CorpusError:
+            return None
+        if start < 1 or start > len(lines):
+            return None
+        end = max(start, min(end, len(lines), start + 40))
+        text = "\n".join(lines[start - 1 : end]).strip()
+        return text or None

backend/app/llm/embeddings.py

@@ -1,46 +0,0 @@
-"""Embeddings through LiteLLM. Always requests EMBEDDING_DIM so the pgvector column stays fixed."""
-
-from dataclasses import dataclass
-
-import litellm
-
-from app.config import settings
-
-BATCH = 64
-
-
-@dataclass
-class EmbeddingResult:
-    vectors: list[list[float]]
-    tokens: int
-
-
-def embed_texts(texts: list[str]) -> EmbeddingResult:
-    vectors: list[list[float]] = []
-    tokens = 0
-    for i in range(0, len(texts), BATCH):
-        batch = [t[:8000] for t in texts[i : i + BATCH]]
-        resp = litellm.embedding(
-            model=settings.embedding_model,
-            input=batch,
-            api_base=settings.embedding_api_base or None,
-            api_key=settings.embedding_api_key or None,
-            dimensions=settings.embedding_dim,
-        )
-        vectors.extend(item["embedding"] for item in resp.data)
-        tokens += int(getattr(resp, "usage", None) and resp.usage.prompt_tokens or 0)
-    for v in vectors:
-        if len(v) != settings.embedding_dim:
-            raise ValueError(
-                f"Embedding model returned {len(v)} dims, expected {settings.embedding_dim}. "
-                "Check EMBEDDING_MODEL supports the dimensions parameter."
-            )
-    return EmbeddingResult(vectors=vectors, tokens=tokens)
-
-
-def embedding_provider() -> str:
-    return settings.embedding_model.split("/", 1)[0]
-
-
-def embed_query(text: str) -> list[float]:
-    return embed_texts([text]).vectors[0]

backend/app/llm/signatures.py

@@ -14,15 +14,16 @@ class Section(BaseModel):
     text: str
 
 
-class CitedSentence(BaseModel):
-    sentence: str
-    chunk_ids: list[int] = Field(description="Indices of supporting passages from the context list")
+class SourceRef(BaseModel):
+    """A line range in a corpus file. Every generated fact points at one of these."""
 
+    path: str = Field(description="Corpus path of the post, exactly as the tools showed it")
+    line_start: int
+    line_end: int
 
-class GroundedAnswerOut(BaseModel):
-    answer: str = Field(description="Answer text with [n] markers that refer to passage indices")
-    sentences: list[CitedSentence]
-    unsupported: bool = Field(description="True if the passages do not contain enough to answer")
+
+class FileCitation(SourceRef):
+    marker: int = Field(description="The n in the [n] marker used in the answer")
 
 
 class VoiceProfileOut(BaseModel):
@@ -37,13 +38,13 @@ class VoiceProfileOut(BaseModel):
 
 class Claim(BaseModel):
     claim: str
-    chunk_ids: list[int]
+    sources: list[SourceRef]
 
 
 class ScriptLine(BaseModel):
     speaker: Literal["host_a", "host_b"]
     text: str
-    chunk_ids: list[int] = Field(default_factory=list)
+    sources: list[SourceRef] = Field(default_factory=list)
 
 
 class Scene(BaseModel):
@@ -72,14 +73,20 @@ class CleanAndSegment(dspy.Signature):
     sections: list[Section] = dspy.OutputField()
 
 
-class GroundedAnswer(dspy.Signature):
-    """Answer the question using ONLY the numbered passages. Every factual sentence must cite at least
-    one passage with [n]. If the passages do not support an answer, say so and set unsupported=true.
-    Never use em dashes."""
+class ResearchArchive(dspy.Signature):
+    """You are a research assistant working inside a writer's archive, which is a folder of markdown
+    posts. Find the answer by exploring the files: list_files to see what exists, search (a regex
+    grep; try several phrasings and synonyms) to locate relevant lines, then read the surrounding
+    lines before relying on them. Answer ONLY from text you read. Every factual sentence needs a
+    [n] marker, and every marker needs a citation with the file path and the line range you read.
+    If the archive does not cover the question, say so plainly and set unsupported=true.
+    Write in plain, warm prose. Never use em dashes."""
 
     question: str = dspy.InputField()
-    passages: list[str] = dspy.InputField(desc="Numbered passages from the writer's own posts")
-    result: GroundedAnswerOut = dspy.OutputField()
+    archive_guide: str = dspy.InputField(desc="How this notebook's files are laid out")
+    answer: str = dspy.OutputField(desc="Answer with [n] citation markers")
+    citations: list[FileCitation] = dspy.OutputField()
+    unsupported: bool = dspy.OutputField()
 
 
 class GroundednessJudge(dspy.Signature):
@@ -100,7 +107,7 @@ class BuildVoiceProfile(dspy.Signature):
 class ExtractClaims(dspy.Signature):
     """List the key claims of the post, each with the passages that support it."""
 
-    passages: list[str] = dspy.InputField()
+    passages: list[str] = dspy.InputField(desc="Post text with path and line numbers")
     claims: list[Claim] = dspy.OutputField()
 
 

backend/app/models/__init__.py

@@ -2,7 +2,6 @@ from app.models.content import (
     Artifact,
     CalendarItem,
     Chat,
-    Chunk,
     Citation,
     Document,
     DspyTrace,
@@ -26,7 +25,6 @@ __all__ = [
     "AuthProvider",
     "CalendarItem",
     "Chat",
-    "Chunk",
     "Citation",
     "Document",
     "DspyTrace",

backend/app/models/content.py

@@ -1,7 +1,6 @@
 import uuid
 from datetime import datetime
 
-from pgvector.sqlalchemy import Vector
 from sqlalchemy import (
     JSON,
     BigInteger,
@@ -17,7 +16,6 @@ from sqlalchemy import (
 )
 from sqlalchemy.orm import Mapped, mapped_column, relationship
 
-from app.config import settings
 from app.db import Base
 from app.models.base import IdMixin, TimestampMixin
 
@@ -49,29 +47,12 @@ class Document(IdMixin, TimestampMixin, Base):
     url: Mapped[str] = mapped_column(String(1000))
     published_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
     raw_html_key: Mapped[str | None] = mapped_column(String(500))  # R2 key
+    # Corpus relative path of the clean markdown file, e.g. sources/ada-substack-com/2026-03-01-pricing.md
+    path: Mapped[str | None] = mapped_column(String(500), index=True)
     clean_text: Mapped[str] = mapped_column(Text, default="")
     content_hash: Mapped[str | None] = mapped_column(String(64))
     metadata_json: Mapped[dict] = mapped_column(JSON, default=dict)
 
-    chunks: Mapped[list["Chunk"]] = relationship(back_populates="document", cascade="all, delete-orphan")
-
-
-class Chunk(IdMixin, TimestampMixin, Base):
-    __tablename__ = "chunks"
-
-    workspace_id: Mapped[uuid.UUID] = _ws_fk()
-    document_id: Mapped[uuid.UUID] = mapped_column(
-        Uuid, ForeignKey("documents.id", ondelete="CASCADE"), index=True
-    )
-    position: Mapped[int] = mapped_column(Integer)
-    heading: Mapped[str | None] = mapped_column(String(500))
-    text: Mapped[str] = mapped_column(Text)
-    token_count: Mapped[int] = mapped_column(Integer, default=0)
-    embedding: Mapped[list[float] | None] = mapped_column(Vector(settings.embedding_dim))
-    entities: Mapped[list] = mapped_column(JSON, default=list)
-
-    document: Mapped[Document] = relationship(back_populates="chunks")
-
 
 class Notebook(IdMixin, TimestampMixin, Base):
     __tablename__ = "notebooks"
@@ -139,9 +120,12 @@ class Citation(IdMixin, Base):
     __tablename__ = "citations"
 
     message_id: Mapped[uuid.UUID] = mapped_column(Uuid, ForeignKey("messages.id", ondelete="CASCADE"))
-    chunk_id: Mapped[uuid.UUID] = mapped_column(Uuid, ForeignKey("chunks.id", ondelete="CASCADE"))
+    document_id: Mapped[uuid.UUID] = mapped_column(Uuid, ForeignKey("documents.id", ondelete="CASCADE"))
     marker: Mapped[int] = mapped_column(Integer)  # [1], [2] in the answer text
-    span: Mapped[str | None] = mapped_column(Text)  # quoted supporting text
+    path: Mapped[str] = mapped_column(String(500))
+    line_start: Mapped[int] = mapped_column(Integer)
+    line_end: Mapped[int] = mapped_column(Integer)
+    span: Mapped[str | None] = mapped_column(Text)  # verified text read back from the file
 
 
 class Artifact(IdMixin, TimestampMixin, Base):
@@ -241,7 +225,7 @@ class UsageEvent(IdMixin, TimestampMixin, Base):
 
     workspace_id: Mapped[uuid.UUID] = _ws_fk()
     job_id: Mapped[uuid.UUID | None] = mapped_column(Uuid, ForeignKey("jobs.id", ondelete="SET NULL"))
-    kind: Mapped[str] = mapped_column(String(20))  # llm | embedding | tts | render
+    kind: Mapped[str] = mapped_column(String(20))  # llm | tts | render
     provider: Mapped[str] = mapped_column(String(40))
     model: Mapped[str | None] = mapped_column(String(100))
     quantity: Mapped[float] = mapped_column(Float)  # tokens, characters or seconds

backend/app/pipeline/ingest.py

@@ -1,4 +1,4 @@
-"""Ingestion: feed -> documents (raw HTML to R2) -> clean text -> chunks -> embeddings."""
+"""Ingestion: feed -> raw HTML to R2 -> clean markdown files in the workspace corpus -> INDEX.md."""
 
 import hashlib
 import logging
@@ -11,22 +11,17 @@ from urllib.parse import urlparse
 
 import feedparser
 import httpx
-import tiktoken
 from bs4 import BeautifulSoup
 from sqlalchemy import select
 from sqlalchemy.orm import Session
 
-from app.config import settings
-from app.llm.embeddings import embed_texts, embedding_provider
-from app.models import Chunk, Document, Job, Source
-from app.services.jobs import record_usage, update_job
+from app.corpus import INDEX, Corpus, index_line, post_path, render_post, slugify
+from app.models import Document, Job, Source
+from app.services.jobs import update_job
 from app.services.storage import keys, storage
 
 log = logging.getLogger(__name__)
 
-CHUNK_TOKENS = 350
-CHUNK_OVERLAP = 60
-_enc = tiktoken.get_encoding("cl100k_base")
 _DROP = re.compile(r"(subscribe|share this post|leave a comment|thanks for reading)", re.I)
 
 
@@ -62,20 +57,6 @@ def html_to_sections(html: str) -> list[tuple[str | None, str]]:
     return [(h, "\n\n".join(p)) for h, p in sections if p]
 
 
-def chunk_sections(sections: list[tuple[str | None, str]]) -> list[tuple[str | None, str, int]]:
-    out: list[tuple[str | None, str, int]] = []
-    for heading, text in sections:
-        tokens = _enc.encode(text)
-        start = 0
-        while start < len(tokens):
-            window = tokens[start : start + CHUNK_TOKENS]
-            out.append((heading, _enc.decode(window), len(window)))
-            if start + CHUNK_TOKENS >= len(tokens):
-                break
-            start += CHUNK_TOKENS - CHUNK_OVERLAP
-    return out
-
-
 @dataclass
 class FeedEntry:
     title: str
@@ -103,25 +84,54 @@ def fetch_feed(feed_url: str, limit: int) -> tuple[str | None, list[FeedEntry]]:
     return parsed.feed.get("title"), entries
 
 
+def source_slug(source: Source) -> str:
+    return slugify(urlparse(source.site_url or source.feed_url).netloc or str(source.id), 50)
+
+
+def rebuild_index(db: Session, workspace_id: uuid.UUID) -> str:
+    docs = db.scalars(
+        select(Document).where(Document.workspace_id == workspace_id, Document.path.is_not(None))
+    ).all()
+    rows = sorted(
+        (index_line(d.path, d.title, d.published_at, d.metadata_json.get("headings", [])) for d in docs),
+        reverse=True,
+    )
+    header = "# Archive index\n\ndate | path | title | section headings\n\n"
+    return header + "\n".join(rows) + "\n"
+
+
 def ingest_source(db: Session, source: Source, job: Job, max_posts: int) -> dict:
     source.sync_status = "syncing"
     update_job(db, job, status="running", progress=0.02, message="Contacting your feed")
 
     title, entries = fetch_feed(source.feed_url, max_posts)
     source.title = source.title or title
+    corpus = Corpus(source.workspace_id)
+    folder = source_slug(source)
     total = max(len(entries), 1)
     indexed = skipped = 0
-    embed_tokens = 0
+    files: dict[str, str] = {}
 
+    taken = set(
+        db.scalars(select(Document.path).where(Document.workspace_id == source.workspace_id)).all()
+    )
     for i, entry in enumerate(entries):
         content_hash = hashlib.sha256(entry.html.encode()).hexdigest()
         doc = db.scalar(select(Document).where(Document.source_id == source.id, Document.url == entry.url))
-        if doc and doc.content_hash == content_hash:
+        if doc and doc.content_hash == content_hash and doc.path:
             skipped += 1
             continue
         if not doc:
             doc = Document(id=uuid.uuid4(), workspace_id=source.workspace_id, source_id=source.id, url=entry.url)
             db.add(doc)
+        if not doc.path:
+            path = post_path(folder, entry.title, entry.url, entry.published_at)
+            n = 2
+            while path in taken:
+                path = path.removesuffix(".md").rsplit("--", 1)[0] + f"--{n}.md"
+                n += 1
+            doc.path = path
+            taken.add(path)
         doc.title = entry.title
         doc.published_at = entry.published_at
         doc.content_hash = content_hash
@@ -130,37 +140,18 @@ def ingest_source(db: Session, source: Source, job: Job, max_posts: int) -> dict
         )
         sections = html_to_sections(entry.html)
         doc.clean_text = "\n\n".join(f"## {h}\n{t}" if h else t for h, t in sections)
-        doc.chunks.clear()
-        pieces = chunk_sections(sections)
-        if pieces:
-            result = embed_texts([f"{entry.title}\n{h or ''}\n{t}" for h, t, _ in pieces])
-            embed_tokens += result.tokens
-            for pos, ((heading, text, n_tokens), vec) in enumerate(zip(pieces, result.vectors, strict=True)):
-                doc.chunks.append(
-                    Chunk(
-                        workspace_id=source.workspace_id,
-                        position=pos,
-                        heading=heading,
-                        text=text,
-                        token_count=n_tokens,
-                        embedding=vec,
-                    )
-                )
+        doc.metadata_json = {**(doc.metadata_json or {}), "headings": [h for h, _ in sections if h]}
+        files[doc.path] = render_post(
+            title=entry.title, url=entry.url, published_at=entry.published_at, source=folder, sections=sections
+        )
         indexed += 1
         db.commit()
-        update_job(db, job, progress=0.05 + 0.9 * (i + 1) / total, message=f"{entry.title[:80]} is in orbit")
-
-    if embed_tokens:
-        record_usage(
-            db,
-            workspace_id=source.workspace_id,
-            kind="embedding",
-            provider=embedding_provider(),
-            model=settings.embedding_model,
-            quantity=embed_tokens,
-            unit="tokens",
-            job=job,
-        )
+        update_job(db, job, progress=0.05 + 0.85 * (i + 1) / total, message=f"{entry.title[:80]} is in orbit")
+
+    update_job(db, job, progress=0.93, message="Writing your archive index")
+    files[INDEX] = rebuild_index(db, source.workspace_id)
+    corpus.write_files(files)
+
     source.sync_status = "ok"
     source.sync_error = None
     source.last_synced_at = datetime.now(UTC)

backend/app/pipeline/research.py

@@ -0,0 +1,109 @@
+"""Agentic research over the file system corpus: the model explores files with list/search/read,
+then every citation is checked against the real file before it reaches the writer."""
+
+import re
+from collections.abc import Callable
+from dataclasses import dataclass, field
+
+import dspy
+
+from app.config import settings
+from app.corpus import Corpus, ScopedTools
+from app.llm.postprocess import strip_em_dashes
+from app.llm.provider import main_lm, track_usage
+from app.llm.signatures import ResearchArchive
+
+_MARKER = re.compile(r"\[(\d+)\]")
+
+
+@dataclass
+class VerifiedCitation:
+    marker: int
+    path: str
+    line_start: int
+    line_end: int
+    quote: str
+
+
+@dataclass
+class ResearchResult:
+    text: str
+    unsupported: bool
+    citations: list[VerifiedCitation]
+    prompt_tokens: int = 0
+    completion_tokens: int = 0
+    steps: list[tuple[str, str]] = field(default_factory=list)
+
+
+def archive_guide(tools: ScopedTools) -> str:
+    return (
+        f"This notebook has {len(tools.allowed)} posts stored as markdown files under sources/. "
+        "Each file starts with front matter (title, url, published date), then '# Title' and '## Section' "
+        "headings. Paths look like sources/<site>/<yyyy-mm-dd>-<slug>.md. "
+        "Line numbers from read() and search() are what you cite."
+    )
+
+
+def verify_citations(tools: ScopedTools, answer: str, citations) -> tuple[str, list[VerifiedCitation]]:
+    """Keep only citations that point at real lines, renumber them 1..n, and drop dangling markers."""
+    verified: list[VerifiedCitation] = []
+    remap: dict[int, int] = {}
+    for c in sorted(citations, key=lambda c: c.marker):
+        if c.marker in remap:
+            continue
+        quote = tools.quote(c.path, c.line_start, c.line_end)
+        if not quote:
+            continue
+        new = len(verified) + 1
+        remap[c.marker] = new
+        end = max(c.line_start, min(c.line_end, c.line_start + 40))
+        verified.append(VerifiedCitation(new, c.path, c.line_start, end, quote[:600]))
+
+    def swap(m: re.Match) -> str:
+        n = remap.get(int(m.group(1)))
+        return f"[{n}]" if n else ""
+
+    text = _MARKER.sub(swap, answer)
+    text = re.sub(r"\s+([.,;:!?])", r"\1", text)
+    return text.strip(), verified
+
+
+def research(
+    corpus: Corpus,
+    allowed: dict[str, str],
+    question: str,
+    on_step: Callable[[str, str], None] | None = None,
+) -> ResearchResult:
+    if not allowed:
+        return ResearchResult("This notebook has no posts yet. Add some sources first.", True, [])
+    corpus.sync()
+    steps: list[tuple[str, str]] = []
+
+    def log(kind: str, detail: str) -> None:
+        steps.append((kind, detail))
+        if on_step:
+            on_step(kind, detail)
+
+    tools = ScopedTools(corpus, allowed, on_step=log)
+    agent = dspy.ReAct(
+        ResearchArchive,
+        tools=[tools.list_files, tools.search, tools.read],
+        max_iters=settings.research_max_steps,
+    )
+    lm = main_lm()
+    with track_usage(lm) as usage, dspy.context(lm=lm):
+        pred = agent(question=question, archive_guide=archive_guide(tools))
+
+    text, cites = verify_citations(tools, pred.answer, pred.citations or [])
+    unsupported = bool(pred.unsupported) or not cites
+    if not cites and not pred.unsupported:
+        # The model answered without anything we could verify: do not pass it off as grounded.
+        text = "I could not find support for that in these posts, so I would rather not guess."
+    return ResearchResult(
+        text=strip_em_dashes(text),
+        unsupported=unsupported,
+        citations=cites,
+        prompt_tokens=usage.prompt_tokens,
+        completion_tokens=usage.completion_tokens,
+        steps=steps,
+    )

backend/app/pipeline/retrieval.py

@@ -1,62 +0,0 @@
-"""Vector retrieval + grounded answering with citations."""
-
-import uuid
-from dataclasses import dataclass
-
-import dspy
-from sqlalchemy import select
-from sqlalchemy.orm import Session
-
-from app.llm.embeddings import embed_query
-from app.llm.postprocess import strip_em_dashes
-from app.llm.provider import main_lm, track_usage
-from app.llm.signatures import GroundedAnswer
-from app.models import Chunk, Document, NotebookDocument
-
-
-@dataclass
-class Retrieved:
-    chunk: Chunk
-    document: Document
-
-
-def retrieve(db: Session, workspace_id: uuid.UUID, query: str, notebook_id: uuid.UUID | None, k: int = 8):
-    vec = embed_query(query)
-    stmt = (
-        select(Chunk, Document)
-        .join(Document, Document.id == Chunk.document_id)
-        .where(Chunk.workspace_id == workspace_id, Chunk.embedding.is_not(None))
-    )
-    if notebook_id:
-        stmt = stmt.join(NotebookDocument, NotebookDocument.document_id == Document.id).where(
-            NotebookDocument.notebook_id == notebook_id
-        )
-    stmt = stmt.order_by(Chunk.embedding.cosine_distance(vec)).limit(k)
-    return [Retrieved(c, d) for c, d in db.execute(stmt).all()]
-
-
-@dataclass
-class Answer:
-    text: str
-    unsupported: bool
-    citations: list[tuple[int, Retrieved]]  # (marker, retrieved)
-    prompt_tokens: int
-    completion_tokens: int
-
-
-def grounded_answer(question: str, hits: list[Retrieved]) -> Answer:
-    if not hits:
-        return Answer("I could not find anything about that in your posts yet.", True, [], 0, 0)
-    passages = [f"[{i + 1}] {h.document.title}: {h.chunk.text}" for i, h in enumerate(hits)]
-    lm = main_lm()
-    with track_usage(lm) as usage, dspy.context(lm=lm, adapter=dspy.JSONAdapter()):
-        pred = dspy.Predict(GroundedAnswer)(question=question, passages=passages)
-    result = pred.result
-    used = sorted({i for s in result.sentences for i in s.chunk_ids if 1 <= i <= len(hits)})
-    return Answer(
-        text=strip_em_dashes(result.answer),
-        unsupported=result.unsupported,
-        citations=[(i, hits[i - 1]) for i in used],
-        prompt_tokens=usage.prompt_tokens,
-        completion_tokens=usage.completion_tokens,
-    )

backend/app/routers/notebooks.py

@@ -1,17 +1,18 @@
+import asyncio
 import json
 import uuid
 
 from fastapi import APIRouter, Depends, HTTPException
-from fastapi.concurrency import run_in_threadpool
 from fastapi.responses import StreamingResponse
 from pydantic import BaseModel, Field
 from sqlalchemy import select
 
 from app.auth import Ctx, get_ctx
 from app.config import settings
+from app.corpus import Corpus
 from app.llm.provider import provider_name
 from app.models import Chat, Citation, Document, Message, Notebook, NotebookDocument
-from app.pipeline.retrieval import grounded_answer, retrieve
+from app.pipeline.research import research
 from app.services.jobs import record_usage
 
 router = APIRouter(prefix="/api/notebooks", tags=["notebooks"])
@@ -99,6 +100,9 @@ def _sse(event: str, data: dict) -> str:
     return f"event: {event}\ndata: {json.dumps(data)}\n\n"
 
 
+_STEP_LABEL = {"list": "Scanning the archive", "search": "Searching for", "read": "Reading"}
+
+
 @router.post("/{notebook_id}/chat")
 async def chat(notebook_id: uuid.UUID, body: ChatIn, ctx: Ctx = Depends(get_ctx)):
     nb = _get(ctx, notebook_id)
@@ -111,41 +115,71 @@ async def chat(notebook_id: uuid.UUID, body: ChatIn, ctx: Ctx = Depends(get_ctx)
     db.add(Message(chat_id=chat_row.id, role="user", content=body.question))
     db.commit()
 
+    docs = db.scalars(
+        select(Document)
+        .join(NotebookDocument, NotebookDocument.document_id == Document.id)
+        .where(NotebookDocument.notebook_id == nb.id, Document.path.is_not(None))
+    ).all()
+    allowed = {d.path: d.title for d in docs}
+    by_path = {d.path: d for d in docs}
+    corpus = Corpus(ctx.workspace.id)
+
     async def stream():
-        yield _sse("status", {"chat_id": str(chat_row.id), "message": "Scanning your archive"})
-        hits = await run_in_threadpool(retrieve, db, ctx.workspace.id, body.question, nb.id)
-        yield _sse("status", {"message": f"Found {len(hits)} relevant passages"})
-        answer = await run_in_threadpool(grounded_answer, body.question, hits)
-        msg = Message(chat_id=chat_row.id, role="assistant", content=answer.text)
+        yield _sse("status", {"chat_id": str(chat_row.id), "message": "Opening the archive"})
+        loop = asyncio.get_running_loop()
+        steps: asyncio.Queue = asyncio.Queue()
+
+        def on_step(kind: str, detail: str) -> None:
+            loop.call_soon_threadsafe(steps.put_nowait, (kind, detail))
+
+        task = asyncio.create_task(asyncio.to_thread(research, corpus, allowed, body.question, on_step))
+        while not task.done() or not steps.empty():
+            try:
+                kind, detail = await asyncio.wait_for(steps.get(), timeout=0.25)
+            except TimeoutError:
+                continue
+            yield _sse("step", {"kind": kind, "message": f"{_STEP_LABEL.get(kind, kind)} {detail}".strip()})
+        try:
+            result = task.result()
+        except Exception:
+            yield _sse("error", {"message": "Lost signal while researching. Try again."})
+            return
+
+        msg = Message(chat_id=chat_row.id, role="assistant", content=result.text)
         db.add(msg)
         db.flush()
         citations = []
-        for marker, hit in answer.citations:
-            db.add(Citation(message_id=msg.id, chunk_id=hit.chunk.id, marker=marker, span=hit.chunk.text[:500]))
+        for c in result.citations:
+            doc = by_path.get(c.path)
+            if not doc:
+                continue
+            db.add(Citation(message_id=msg.id, document_id=doc.id, marker=c.marker, path=c.path,
+                            line_start=c.line_start, line_end=c.line_end, span=c.quote))
             citations.append({
-                "marker": marker,
-                "chunk_id": str(hit.chunk.id),
-                "document_id": str(hit.document.id),
-                "title": hit.document.title,
-                "url": hit.document.url,
-                "heading": hit.chunk.heading,
-                "span": hit.chunk.text[:500],
+                "marker": c.marker,
+                "document_id": str(doc.id),
+                "title": doc.title,
+                "url": doc.url,
+                "path": c.path,
+                "line_start": c.line_start,
+                "line_end": c.line_end,
+                "span": c.quote,
             })
         db.commit()
-        if answer.prompt_tokens or answer.completion_tokens:
+        if result.prompt_tokens or result.completion_tokens:
             record_usage(
                 db,
                 workspace_id=ctx.workspace.id,
                 kind="llm",
                 provider=provider_name(),
                 model=settings.llm_model,
-                quantity=answer.prompt_tokens + answer.completion_tokens,
+                quantity=result.prompt_tokens + result.completion_tokens,
                 unit="tokens",
             )
         yield _sse("answer", {
             "message_id": str(msg.id),
-            "text": answer.text,
-            "unsupported": answer.unsupported,
+            "text": result.text,
+            "unsupported": result.unsupported,
             "citations": citations,
         })
         yield _sse("done", {})

backend/requirements.txt

@@ -3,7 +3,6 @@ uvicorn[standard]==0.32.*
 sqlalchemy==2.0.*
 psycopg[binary]==3.2.*
 alembic==1.14.*
-pgvector==0.3.*
 pydantic==2.*
 pydantic-settings==2.*
 email-validator==2.*
@@ -19,4 +18,3 @@ redis==5.*
 dspy==2.6.*
 feedparser==6.*
 beautifulsoup4==4.*
-tiktoken==0.8.*

backend/tests/test_corpus.py

@@ -0,0 +1,148 @@
+import uuid
+from datetime import UTC, datetime
+
+import pytest
+
+from app.corpus import INDEX, Corpus, CorpusError, ScopedTools, index_line, post_path, render_post
+from app.llm.signatures import FileCitation
+from app.pipeline.research import verify_citations
+
+
+class MemoryStore:
+    """Stands in for R2."""
+
+    def __init__(self):
+        self.objects: dict[str, bytes] = {}
+
+    def put_text(self, key, text, content_type=None):
+        self.objects[key] = text.encode()
+        return key
+
+    def get_bytes(self, key):
+        return self.objects[key]
+
+    def exists(self, key):
+        return key in self.objects
+
+    def delete(self, key):
+        self.objects.pop(key, None)
+
+
+PRICING = [("Why I raised prices", "I moved the paid tier from $5 to $8 a month.\n\nChurn stayed flat."),
+           ("What readers said", "Most readers told me the archive alone was worth it.")]
+HABITS = [(None, "I write every morning before email.")]
+
+
+@pytest.fixture()
+def corpus_pair(tmp_path):
+    store = MemoryStore()
+    ws = uuid.uuid4()
+    writer = Corpus(ws, store=store, cache_dir=str(tmp_path / "worker"))
+    d1 = datetime(2026, 3, 1, tzinfo=UTC)
+    d2 = datetime(2025, 11, 9, tzinfo=UTC)
+    p1 = post_path("ada-substack-com", "On pricing", "https://ada.substack.com/p/on-pricing", d1)
+    p2 = post_path("ada-substack-com", "Habits", "https://ada.substack.com/p/habits", d2)
+    writer.write_files({
+        p1: render_post(
+            title="On pricing", url="https://x/p/on-pricing", published_at=d1, source="ada", sections=PRICING
+        ),
+        p2: render_post(title="Habits", url="https://x/p/habits", published_at=d2, source="ada", sections=HABITS),
+        INDEX: "\n".join([index_line(p1, "On pricing", d1, ["Why I raised prices"]), index_line(p2, "Habits", d2, [])]),
+    })
+    # A different process (the API container) with an empty cache
+    reader = Corpus(ws, store=store, cache_dir=str(tmp_path / "api"))
+    reader.sync()
+    return reader, p1, p2
+
+
+def test_paths_are_stable_and_readable():
+    p = post_path("ada-substack-com", "Any Title", "https://ada.substack.com/p/on-pricing/", datetime(2026, 3, 1))
+    assert p == "sources/ada-substack-com/2026-03-01-on-pricing.md"
+
+
+def test_sync_mirrors_r2_into_a_fresh_cache(corpus_pair):
+    reader, p1, _ = corpus_pair
+    assert reader.exists(p1) and reader.exists(INDEX)
+    assert "# On pricing" in reader.read_lines(p1)
+
+
+def test_tools_search_read_and_scope(corpus_pair):
+    reader, p1, p2 = corpus_pair
+    steps = []
+    tools = ScopedTools(reader, {p1: "On pricing"}, on_step=lambda k, d: steps.append(k))
+
+    listing = tools.list_files()
+    assert p1 in listing and p2 not in listing  # notebook scope hides other posts
+
+    hits = tools.search(r"paid tier|price")
+    assert f"{p1}:" in hits and "$5 to $8" in hits
+    assert "Habits" not in hits
+
+    body = tools.read(p1, 1, 5)
+    assert "1| ---" in body
+
+    with pytest.raises(CorpusError):
+        tools.read(p2)  # outside the notebook
+    with pytest.raises(CorpusError):
+        tools.read("../../etc/passwd")
+    assert tools.search("(unclosed")  # bad regex falls back to a literal search
+    assert steps[:3] == ["list", "search", "read"]
+
+
+def test_verify_citations_drops_fakes_and_renumbers(corpus_pair):
+    reader, p1, p2 = corpus_pair
+    tools = ScopedTools(reader, {p1: "On pricing"})
+    line = next(i for i, t in enumerate(reader.read_lines(p1), 1) if "$5 to $8" in t)
+    cites = [
+        FileCitation(marker=1, path=p2, line_start=1, line_end=3),          # out of scope
+        FileCitation(marker=2, path=p1, line_start=line, line_end=line),   # real
+        FileCitation(marker=3, path=p1, line_start=9999, line_end=10000),  # past end of file
+    ]
+    text, verified = verify_citations(tools, "Habits matter [1]. Prices went to $8 [2]. Churn fell [3].", cites)
+    assert [c.marker for c in verified] == [1]
+    assert "$5 to $8" in verified[0].quote
+    assert text == "Habits matter. Prices went to $8 [1]. Churn fell."
+
+
+def test_react_agent_end_to_end_with_scripted_model(corpus_pair, monkeypatch):
+    """Drives the real dspy.ReAct loop with a fake LM: search, read, then answer with a citation."""
+    import dspy
+    from dspy.utils.dummies import DummyLM
+
+    import app.pipeline.research as research_mod
+
+    reader, p1, _ = corpus_pair
+    line = next(i for i, t in enumerate(reader.read_lines(p1), 1) if "$5 to $8" in t)
+    lm = DummyLM([
+        {"next_thought": "Search for pricing.", "next_tool_name": "search", "next_tool_args": {"pattern": "paid tier"}},
+        {"next_thought": "Read the post.", "next_tool_name": "read",
+         "next_tool_args": {"path": p1, "start_line": 1, "end_line": 30}},
+        {"next_thought": "I have what I need.", "next_tool_name": "finish", "next_tool_args": {}},
+        {"reasoning": "The post states the new price.",
+         "answer": "You raised the paid tier from $5 to $8 a month [1].",
+         "citations": [{"marker": 1, "path": p1, "line_start": line, "line_end": line}],
+         "unsupported": False},
+    ])
+    monkeypatch.setattr(research_mod, "main_lm", lambda: lm)
+    monkeypatch.setattr(research_mod, "track_usage", _no_usage)
+    seen = []
+    with dspy.context(adapter=dspy.ChatAdapter()):
+        result = research_mod.research(
+            reader, {p1: "On pricing"}, "What did I charge?", on_step=lambda k, d: seen.append(k)
+        )
+    assert seen == ["search", "read"]
+    assert result.unsupported is False
+    assert result.text.endswith("[1].")
+    assert result.citations[0].path == p1 and "$5 to $8" in result.citations[0].quote
+
+
+class _Usage:
+    prompt_tokens = completion_tokens = 0
+
+
+from contextlib import contextmanager  # noqa: E402
+
+
+@contextmanager
+def _no_usage(lm):
+    yield _Usage()

backend/tests/test_misc.py

@@ -1,7 +1,7 @@
 import uuid
 
 from app.llm.postprocess import clean_output, strip_em_dashes
-from app.pipeline.ingest import chunk_sections, html_to_sections, normalize_feed_url
+from app.pipeline.ingest import html_to_sections, normalize_feed_url
 from app.services.email import make_unsubscribe_token, verify_unsubscribe_token
 from app.services.storage import keys, safe_filename, workspace_owns_key
 
@@ -19,12 +19,11 @@ def test_normalize_feed_url():
     assert normalize_feed_url("https://blog.example.com/rss")[1] == "rss"
 
 
-def test_html_to_sections_and_chunking():
+def test_html_to_sections():
     html = "<h2>Intro</h2><p>" + "word " * 900 + "</p><p>Subscribe now</p><h2>Two</h2><p>Short.</p>"
     sections = html_to_sections(html)
     assert [h for h, _ in sections] == ["Intro", "Two"]
-    chunks = chunk_sections(sections)
-    assert len(chunks) >= 3 and all(n <= 350 for _, _, n in chunks)
+    assert "Subscribe now" not in str(sections)
 
 
 def test_storage_keys_are_workspace_scoped():

docker-compose.yml

@@ -1,7 +1,7 @@
 # Local stack. MinIO stands in for Cloudflare R2 (same S3 API); in prod drop minio and set R2_* vars.
 services:
   postgres:
-    image: pgvector/pgvector:pg16
+    image: postgres:16-alpine
     environment:
       POSTGRES_USER: notestack
       POSTGRES_PASSWORD: notestack
@@ -35,6 +35,8 @@ services:
       R2_ENDPOINT_URL: ${R2_ENDPOINT_URL:-http://minio:9000}
       RENDERER_URL: http://renderer:3100
       API_URL: http://api:8000
+      CORPUS_CACHE_DIR: /cache/corpus
+    volumes: [apicache:/cache]
     ports: ["8000:8000"]
     depends_on:
       postgres: { condition: service_healthy }
@@ -50,6 +52,8 @@ services:
       R2_ENDPOINT_URL: ${R2_ENDPOINT_URL:-http://minio:9000}
       RENDERER_URL: http://renderer:3100
       API_URL: http://api:8000
+      CORPUS_CACHE_DIR: /cache/corpus
+    volumes: [workercache:/cache]
     depends_on: [api]
 
   renderer:
@@ -72,3 +76,5 @@ services:
 volumes:
   pgdata:
   miniodata:
+  apicache:
+  workercache:

docs/DESIGN.md

@@ -27,13 +27,13 @@ Every factual statement Notestack generates must trace back to the writer's own
 | --- | --- |
 | Backend | FastAPI + SQLAlchemy 2 + Alembic (same stack as blog2video) |
 | Worker | arq on Redis (async jobs, progress events over Redis pub/sub, SSE to clients) |
-| Database | Postgres 16 + pgvector |
+| Database | Postgres 16 (no vector extension) |
 | Frontend | React + Vite + TypeScript, plain CSS with design tokens |
 | Renderer | Node service running Remotion (`renderer/`) |
 | Storage | Cloudflare R2 (S3 API) for raw HTML, audio, video, stills, compiled DSPy programs |
 | Email | Resend, behind a provider interface (mirrors blog2video `EmailService`) |
 | Auth | Google sign-in + email 6-digit code signup + password login, JWT access/refresh with `token_version` revocation (mirrors blog2video) |
-| LLM | Provider agnostic through DSPy + LiteLLM. Default Z.ai GLM-5.3 (fast tier GLM-5.3-Flash) via its OpenAI compatible endpoint. Embeddings use a separate provider (OpenAI text-embedding-3-small by default). Any LiteLLM model string works by env var. |
+| LLM | Provider agnostic through DSPy + LiteLLM. Default Z.ai GLM-5.3 (fast tier GLM-5.3-Flash) via its OpenAI compatible endpoint. Retrieval is agentic search over a file system corpus, not embeddings (see 3.2). Any LiteLLM model string works by env var. |
 | Billing | Three tiers defined in code (`app/services/plans.py`). `BILLING_ENABLED=false` for now: everyone gets the top tier limits free, checkout returns 409, pricing page shows "Free during early access". |
 | Blog | Typed `blogPosts.ts` content array, `/blogs` and `/blogs/:slug`, same shape as blog2video |
 
@@ -48,14 +48,14 @@ Every factual statement Notestack generates must trace back to the writer's own
                      |                                               | enqueue (arq)
             +--------+----------+                          +---------v----------+
             | Cloudflare R2     |<-------------------------|  worker (arq)      |
-            | notestack-assets  |   put objects            |  ingest, chunk,    |
-            +--------+----------+                          |  embed, DSPy, TTS  |
+            | notestack-assets  |   put objects            |  ingest, corpus,   |
+            +--------+----------+                          |  DSPy, TTS         |
                      ^                                     +---------+----------+
                      | upload final render                           | POST /render
             +--------+----------+   progress callbacks     +---------v----------+
             | renderer (Remotion)|------------------------>|  api /internal/... |
             +-------------------+                          +--------------------+
-     Postgres + pgvector (api, worker)        Redis (queue, pub/sub, cache)
+     Postgres (api, worker)        Redis (queue, pub/sub, cache)
 ```
 
 ### 3.1 Storage pipeline (R2)
@@ -66,6 +66,9 @@ Key layout (`app/services/storage.py::keys`):
 
 ```
 ws/{workspace_id}/sources/{source_id}/documents/{document_id}/raw.html
+ws/{workspace_id}/corpus/manifest.json                           path -> sha256 of every corpus file
+ws/{workspace_id}/corpus/INDEX.md                                one line per post
+ws/{workspace_id}/corpus/sources/{site}/{date}-{slug}.md         clean post as markdown
 ws/{workspace_id}/uploads/{upload_id}/{filename}
 ws/{workspace_id}/artifacts/{artifact_id}/{variant}.{ext}      audio, video, stills
 ws/{workspace_id}/tts-cache/{sha256(script|voice|settings)}.mp3
@@ -87,7 +90,50 @@ Flows:
 The storage service is an S3 client with R2's endpoint, so MinIO in docker-compose stands in for R2
 locally with zero code changes.
 
-### 3.2 LLM layer
+### 3.2 Research corpus: file system instead of embeddings
+
+Each post is one markdown file (front matter, `# Title`, `## Section` headings) and the workspace
+has an `INDEX.md` with one line per post. Research is an agent loop (`dspy.ReAct`,
+`app/pipeline/research.py`) with three tools scoped to the notebook's files (`app/corpus.py`):
+
+| Tool | Does |
+| --- | --- |
+| `list_files(contains)` | Index lines for the notebook's posts: date, path, title, headings |
+| `search(pattern, path)` | Case insensitive regex grep, returns `path:line: text` |
+| `read(path, start_line, end_line)` | Numbered lines, 160 per call |
+
+The model searches with several phrasings, reads around the hits, and answers with citations of
+the form `(path, line_start, line_end)`. Before anything reaches the writer, every citation is read
+back from the file: citations that point outside the notebook, past the end of a file or at empty
+lines are dropped, markers are renumbered, and an answer left with no verified citation is replaced
+by "not in your posts". The UI streams each step ("Searching for pricing|paid tier", "Reading On
+pricing, lines 1 to 40").
+
+Why this over embeddings:
+- One vendor. Z.ai's international API has no embedding model, and there is no second key,
+  re-embedding job or dimension lock-in.
+- Citations are exact line ranges the model actually read, which is stronger grounding than
+  nearest-neighbour chunks.
+- Writers' archives are small (hundreds to a few thousand posts, a few MB of text), so grep over
+  local files takes milliseconds.
+- The corpus is plain markdown, so the same files feed podcast scripts, launch kits and a future
+  MCP server without conversion.
+
+Trade offs: each question costs several LLM steps instead of one (mitigated by
+`LLM_REASONING_EFFORT=low` and the `RESEARCH_MAX_STEPS` cap), and paraphrase matching depends on
+the model trying synonyms. If quality or latency needs it later, a Postgres full text index can be
+added as a fourth tool without changing the corpus.
+
+Storage and sync: R2 holds the corpus under `ws/{id}/corpus/` with a `manifest.json` of file hashes.
+The worker writes files to R2 and its own cache in one pass. The API calls `Corpus.sync()` before
+each research run, which reads the manifest and downloads only changed files into
+`CORPUS_CACHE_DIR`. Containers share nothing but R2.
+
+Brief items that assumed embeddings: voice similarity uses the LLM judge alone; the topic map is
+built from LLM extracted entities per post (Phase 2); resurfacing and SEO internal links search the
+corpus with the same tools.
+
+### 3.3 LLM layer
 
 `app/llm/provider.py` builds a `dspy.LM` from env:
 
@@ -97,8 +143,6 @@ LLM_FAST_MODEL=openai/glm-5.3-flash    # judges, cleanup, cheap passes
 LLM_API_BASE=https://api.z.ai/api/paas/v4
 LLM_API_KEY=...
 LLM_REASONING_EFFORT=low               # low | high | max
-EMBEDDING_MODEL=openai/text-embedding-3-small
-EMBEDDING_DIM=1024
 ```
 
 Z.ai considerations baked in:
@@ -107,16 +151,15 @@ Z.ai considerations baked in:
   `reasoning_effort=low` by default; modules that benefit (the groundedness judge) ask for
   `main_lm("high")`. Reasoning tokens count against `max_tokens`, so the default is 16K to leave
   room for the structured answer. Z.ai recommends temperature 1.0 for GLM-5.x.
-- 1M token context on GLM-5.3 means whole-notebook prompts (summaries, podcast scripts) rarely
-  need map-reduce; retrieval is still used for chat to keep citations tight and cost low.
-- Z.ai's international API does not offer an embedding model, so embeddings are configured
-  separately. The chosen model must accept a `dimensions` parameter so the pgvector column stays
-  fixed. Changing `EMBEDDING_DIM` requires a migration and re-embed.
+- 1M token context on GLM-5.3 means whole-notebook prompts (summaries, podcast scripts) can read
+  the files directly; chat uses the research agent to keep citations tight and cost low.
+- Tool calling runs through DSPy ReAct, so it works with any provider, not only ones with native
+  function calling.
 - Token usage from every call is written to `usage_events` (metering from day one).
 
 Swapping to Anthropic, OpenAI, Gemini or a local model is only an env change.
 
-### 3.3 Jobs and progress
+### 3.4 Jobs and progress
 
 Every long operation is a row in `jobs` (kind, status, progress, error, cost) plus an arq task.
 Workers publish progress to Redis channel `job:{id}`; `GET /api/jobs/{id}/events` streams it as SSE.
@@ -128,9 +171,9 @@ Implemented in `backend/app/models` and `alembic/versions/0001_initial.py`.
 
 - users, workspaces, workspace_members, subscriptions
 - email_verification_codes, update_emails, update_email_sends
-- sources, documents, chunks (pgvector `embedding`), notebooks, notebook_documents
+- sources, documents (with corpus `path`), notebooks, notebook_documents
 - voice_profiles, voice_consents
-- chats, messages, citations
+- chats, messages, citations (document, path, line range, verified quote)
 - artifacts, jobs (covers render_jobs and tts_jobs via `kind`), uploads
 - calendar_items, tracked_links, engagement_events
 - dspy_traces, usage_events
@@ -138,7 +181,7 @@ Implemented in `backend/app/models` and `alembic/versions/0001_initial.py`.
 ## 5. DSPy program design
 
 See brief Section 6. Signatures live in `backend/app/llm/signatures.py` with Pydantic output models.
-Phase 1 ships CleanAndSegment, GroundedAnswer and BuildVoiceProfile. The rest are stubs with typed
+Phase 1 ships CleanAndSegment, ResearchArchive (the file research agent) and BuildVoiceProfile. All grounded outputs cite `SourceRef(path, line_start, line_end)`. The rest are stubs with typed
 signatures so the schema contract is fixed early.
 
 ## 6. Plans

docs/PHASE0.md

@@ -4,14 +4,14 @@ Legend: [x] done in this scaffold, [ ] open
 
 ## Repo and infra
 - [x] Monorepo layout: `backend/`, `frontend/`, `renderer/`, `docs/`
-- [x] docker-compose: api, worker, postgres (pgvector), redis, minio (local R2 stand-in), renderer, frontend
+- [x] docker-compose: api, worker, postgres, redis, minio (local R2 stand-in), renderer, frontend
 - [x] `.env.example` covering every service
 - [x] CI: backend lint + tests, frontend typecheck + build, renderer typecheck
 - [ ] Production deploy target (Fly, Render or DO App Platform) and R2 bucket + custom domain
 - [ ] Sentry (api, worker, frontend)
 
 ## Backend core
-- [x] Settings (`app/config.py`), DB session, Alembic initial migration with pgvector
+- [x] Settings (`app/config.py`), DB session, Alembic initial migration
 - [x] Full data model from design doc Section 4
 - [x] Health endpoint, CORS, request id logging
 - [x] Jobs table + arq worker + Redis progress pub/sub + SSE endpoint
@@ -39,8 +39,9 @@ Legend: [x] done in this scaffold, [ ] open
 
 ## LLM (agnostic, Z.ai first)
 - [x] `dspy.LM` factory from env, Z.ai defaults
-- [x] Embeddings client with fixed dimensions
-- [x] Signatures + Pydantic outputs for CleanAndSegment, GroundedAnswer, BuildVoiceProfile, stubs for the rest
+- [x] File system corpus in R2 with manifest sync (`app/corpus.py`)
+- [x] Research agent: list/search/read tools, verified line citations, live step streaming
+- [x] Signatures + Pydantic outputs for CleanAndSegment, ResearchArchive, BuildVoiceProfile, stubs for the rest
 - [x] Em dash stripping post-processor
 - [ ] Dev sets and first optimization run (Phase 1)
 

frontend/src/pages/NotebookView.tsx

@@ -6,8 +6,16 @@ import { TelescopeIcon } from "../components/icons/Icons";
 
 type Doc = { id: string; title: string; url: string; published_at: string | null };
 type Notebook = { id: string; title: string; documents: Doc[] };
-type Citation = { marker: number; title: string; url: string; heading: string | null; span: string };
-type Turn = { role: "user" | "assistant"; text: string; citations?: Citation[]; status?: string };
+type Citation = {
+  marker: number;
+  title: string;
+  url: string;
+  path: string;
+  line_start: number;
+  line_end: number;
+  span: string;
+};
+type Turn = { role: "user" | "assistant"; text: string; citations?: Citation[]; status?: string; steps?: string[] };
 
 export default function NotebookView() {
   const { id = "" } = useParams();
@@ -38,6 +46,13 @@ export default function NotebookView() {
           if (event === "status") {
             if (d.chat_id) setChatId(d.chat_id as string);
             patchLast({ status: d.message as string });
+          } else if (event === "step") {
+            setTurns((t) => {
+              const last = t[t.length - 1];
+              return [...t.slice(0, -1), { ...last, status: d.message as string, steps: [...(last.steps ?? []), d.message as string] }];
+            });
+          } else if (event === "error") {
+            patchLast({ text: d.message as string, status: undefined });
           } else if (event === "answer") {
             patchLast({ text: d.text as string, citations: d.citations as Citation[], status: undefined });
           }
@@ -77,7 +92,17 @@ export default function NotebookView() {
           )}
           {turns.map((t, i) => (
             <div key={i} className={`turn turn-${t.role}`}>
-              {t.status && <p className="mono muted">{t.status}...</p>}
+              {t.steps && t.steps.length > 0 && (
+                <details className="agent-steps mono">
+                  <summary>{t.status ? `${t.status}...` : `${t.steps.length} steps through the archive`}</summary>
+                  <ol>
+                    {t.steps.map((s, j) => (
+                      <li key={j}>{s}</li>
+                    ))}
+                  </ol>
+                </details>
+              )}
+              {t.status && !t.steps?.length && <p className="mono muted">{t.status}...</p>}
               {t.text && <p>{t.text}</p>}
               {t.citations && t.citations.length > 0 && (
                 <ol className="cites">
@@ -85,9 +110,12 @@ export default function NotebookView() {
                     <li key={c.marker} value={c.marker}>
                       <a href={c.url} target="_blank" rel="noreferrer">
                         {c.title}
-                        {c.heading ? `: ${c.heading}` : ""}
-                      </a>
-                      <p className="muted cite-span">{c.span.slice(0, 220)}...</p>
+                      </a>{" "}
+                      <span className="mono muted">
+                        lines {c.line_start}
+                        {c.line_end > c.line_start ? ` to ${c.line_end}` : ""}
+                      </span>
+                      <p className="muted cite-span">{c.span.length > 220 ? `${c.span.slice(0, 220)}...` : c.span}</p>
                     </li>
                   ))}
                 </ol>

frontend/src/styles/app.css

@@ -1047,3 +1047,20 @@
   color: var(--white);
   opacity: 0.85;
 }
+
+.agent-steps {
+  font-size: 0.75rem;
+  color: var(--text-2);
+}
+
+.agent-steps summary {
+  cursor: pointer;
+  color: var(--blue);
+}
+
+.agent-steps ol {
+  margin: 8px 0 0;
+  padding-left: 18px;
+  display: grid;
+  gap: 2px;
+}