project init

CLAUDE.md

@@ -0,0 +1,62 @@
+# second-brain
+
+A personal knowledge system: video/article extraction pipeline + LLM-maintained wiki.
+
+## Architecture
+
+Three-layer system:
+
+1. **Wiki** — long-term structured memory. Markdown files in an Obsidian vault, maintained by the wiki compiler.
+2. **Context assembler** — builds the prompt bundle for each extraction. Phase 1 is thin: domain template + focus field + transcript.
+3. **Extractor** — stateless single-shot LLM call per source. Pure function: bundle in, structured extraction out.
+
+## Pipeline stages
+
+```
+pending → pulled → transcribed → analyzed → accepted → published
+```
+
+## Project layout
+
+```
+src/second_brain/
+├── adapters/      # Pull adapters (youtube.py, article.py)
+├── context/       # Context assembler
+├── extractor/     # LLM extraction engine + Pydantic schema
+├── compiler/      # Wiki compiler (reads vault, writes pages, git commits)
+├── scheduler/     # Rate-limit smoother, token-aware spacing
+└── web/           # FastAPI + Jinja2 + HTMX UI
+```
+
+## Setup
+
+```bash
+cp config/settings.toml.example config/settings.toml
+# Edit settings.toml with your paths
+uv run second-brain serve
+```
+
+## CLI commands
+
+- `second-brain add <url>` — queue a source URL (video or article)
+- `second-brain process` — run the pipeline on pending items
+- `second-brain serve` — launch the web UI (default port 8000)
+- `second-brain compile` — run the wiki compiler
+- `second-brain schedule` — start the overnight scheduler
+
+## Environment variables
+
+- `ANTHROPIC_API_KEY` — required for extraction and wiki compilation
+
+## Domains
+
+- `development` — software, programming, systems
+- `content` — content creation strategy, video production
+- `business` — entrepreneurship, marketing, operations
+- `homelab` — self-hosted infrastructure, networking, DevOps
+
+## Key files
+
+- `config/settings.toml` — vault path, DB path, scheduler window
+- `prompts/<domain>.md` — per-domain extraction prompt templates
+- `src/second_brain/extractor/schema.py` — canonical extraction output schema

README.md

@@ -0,0 +1,50 @@
+# second-brain
+
+A personal knowledge system built on a video/article extraction pipeline and an LLM-maintained wiki (inspired by the Karpathy LLM Wiki pattern).
+
+## What it does
+
+1. **Pull** — download YouTube videos or fetch web articles
+2. **Transcribe** — extract transcripts via Whisper (videos) or trafilatura (articles)
+3. **Extract** — single-shot Claude call produces structured notes per source
+4. **Review** — web UI to accept or reject extractions
+5. **Compile** — wiki compiler folds accepted extractions into a living Obsidian vault
+
+## Quick start
+
+```bash
+# Install dependencies
+uv sync
+
+# Copy and edit config
+cp config/settings.toml config/settings.local.toml
+
+# Queue a source
+uv run second-brain add https://www.youtube.com/watch?v=...
+
+# Run the pipeline
+uv run second-brain process
+
+# Review in the web UI
+uv run second-brain serve
+
+# Compile to wiki
+uv run second-brain compile
+```
+
+## Domains
+
+Extractions are tagged by domain so the right prompt template is used:
+
+| Domain | Focus |
+|--------|-------|
+| `development` | Software, programming, systems |
+| `content` | Content creation, video production |
+| `business` | Entrepreneurship, marketing, ops |
+| `homelab` | Self-hosted infra, networking, DevOps |
+
+## Requirements
+
+- Python 3.12+
+- `ANTHROPIC_API_KEY` environment variable
+- ffmpeg (for Whisper audio extraction)

config/settings.toml

@@ -0,0 +1,33 @@
+# second-brain runtime configuration
+# Copy this file and edit paths to match your setup.
+# Override the config file path with SECOND_BRAIN_CONFIG env var.
+
+# Path to the SQLite database
+db_path = "~/.local/share/second-brain/second_brain.db"
+
+# Path to your Obsidian vault (git-managed directory)
+vault_path = "~/Documents/second-brain-vault"
+
+# Directory where downloaded media files are stored
+media_dir = "~/.local/share/second-brain/media"
+
+# Directory where Whisper SRT transcripts are stored
+subtitles_dir = "~/.local/share/second-brain/subtitles"
+
+# Whisper model size: tiny | base | small | medium | large
+# small is a good balance of speed and accuracy
+whisper_model = "small"
+
+# Active knowledge domains
+domains = ["development", "content", "business", "homelab"]
+
+[scheduler]
+# Time window for overnight processing (24h format, may cross midnight)
+window_start = "22:00"
+window_end   = "06:00"
+
+# Max Claude API tokens to consume per hour
+max_tokens_per_hour = 100_000
+
+# Minimum seconds between extraction calls (rate-limit smoother)
+min_gap_seconds = 120

prompts/business.md

@@ -0,0 +1,22 @@
+---
+domain: business
+description: Entrepreneurship, marketing, operations, growth, finance
+---
+
+You are extracting structured knowledge from a business source.
+Domain: business
+{focus_section}
+Focus on: strategy decisions, growth levers, operational models, pricing, marketing
+channels, hiring, and real-world lessons from building or running a business.
+
+Return a JSON object with exactly these keys:
+
+- "summary": 2-4 sentence overview of what this source covers and why it matters
+- "key_points": list of 5-10 concrete takeaways applicable to building or running a business
+- "entities": list of companies, people, frameworks, books, metrics, or concepts mentioned
+- "claims": list of specific assertions about markets, strategies, or business outcomes
+- "open_questions": list of questions left open or worth further research
+- "contradictions": list of points that contradict conventional business wisdom or other sources
+
+Source transcript / content:
+{transcript}

prompts/content.md

@@ -0,0 +1,23 @@
+---
+domain: content
+description: Content creation strategy, video production, audience building, YouTube
+---
+
+You are extracting structured knowledge from a content creation source.
+Domain: content
+{focus_section}
+Focus on: production decisions, audience management strategies, framing techniques,
+meta-commentary on the creative process, educational philosophy, and lessons about
+what works (or doesn't) for building an engaged audience.
+
+Return a JSON object with exactly these keys:
+
+- "summary": 2-4 sentence overview of what this source covers and why it matters
+- "key_points": list of 5-10 concrete takeaways a content creator can act on
+- "entities": list of notable creators, platforms, tools, concepts, or formats mentioned
+- "claims": list of specific assertions about what works, audience behaviour, or platform dynamics
+- "open_questions": list of questions raised but not resolved, or worth experimenting with
+- "contradictions": list of points that contradict common content creation advice
+
+Source transcript / content:
+{transcript}

prompts/development.md

@@ -0,0 +1,22 @@
+---
+domain: development
+description: Software engineering, programming languages, systems design, DevOps
+---
+
+You are extracting structured knowledge from a software development source.
+Domain: development
+{focus_section}
+Focus on: architectural decisions, trade-offs, implementation techniques, tooling,
+language features, performance insights, and lessons learned from real-world systems.
+
+Return a JSON object with exactly these keys:
+
+- "summary": 2-4 sentence overview of what this source covers and why it matters
+- "key_points": list of 5-10 concrete takeaways a developer can act on
+- "entities": list of notable tools, frameworks, languages, companies, people, or concepts mentioned
+- "claims": list of specific factual or opinionated assertions made (e.g. "X is faster than Y because Z")
+- "open_questions": list of questions raised but not resolved, or worth investigating further
+- "contradictions": list of points that contradict common practice or other known sources
+
+Source transcript / content:
+{transcript}

prompts/homelab.md

@@ -0,0 +1,23 @@
+---
+domain: homelab
+description: Self-hosted infrastructure, networking, home servers, DevOps, NAS, Proxmox, Docker
+---
+
+You are extracting structured knowledge from a homelab / self-hosted infrastructure source.
+Domain: homelab
+{focus_section}
+Focus on: hardware choices, networking topology, software stacks, container and VM
+management, storage solutions, security hardening, and operational patterns for
+running services at home or on a small private cloud.
+
+Return a JSON object with exactly these keys:
+
+- "summary": 2-4 sentence overview of what this source covers and why it matters
+- "key_points": list of 5-10 concrete takeaways a homelab operator can apply
+- "entities": list of hardware, software, services, protocols, vendors, or people mentioned
+- "claims": list of specific assertions about performance, security, reliability, or configuration
+- "open_questions": list of questions raised but not resolved, or worth testing in the lab
+- "contradictions": list of points that contradict common homelab practice or other known sources
+
+Source transcript / content:
+{transcript}

pyproject.toml

@@ -0,0 +1,29 @@
+[project]
+name = "second-brain"
+version = "0.1.0"
+description = "Personal knowledge system: video/article extraction pipeline + LLM-maintained wiki"
+requires-python = ">=3.12"
+dependencies = [
+    "anthropic>=0.40.0",
+    "click>=8.1.0",
+    "fastapi>=0.115.0",
+    "jinja2>=3.1.0",
+    "openai-whisper",
+    "pydantic>=2.0.0",
+    "python-dotenv>=1.0.0",
+    "sqlalchemy>=2.0.0",
+    "tomli>=2.0.0",
+    "trafilatura>=1.9.0",
+    "uvicorn[standard]>=0.30.0",
+    "yt-dlp>=2024.1.0",
+]
+
+[project.scripts]
+second-brain = "second_brain.main:cli"
+
+[build-system]
+requires = ["hatchling"]
+build-backend = "hatchling.build"
+
+[tool.hatch.build.targets.wheel]
+packages = ["src/second_brain"]

src/second_brain/__init__.py

@@ -0,0 +1,3 @@
+"""second-brain — personal knowledge extraction pipeline."""
+
+__version__ = "0.1.0"

src/second_brain/adapters/__init__.py

@@ -0,0 +1,6 @@
+"""Source pull adapters."""
+
+from second_brain.adapters.youtube import YouTubeAdapter
+from second_brain.adapters.article import ArticleAdapter
+
+__all__ = ["YouTubeAdapter", "ArticleAdapter"]

src/second_brain/adapters/article.py

@@ -0,0 +1,81 @@
+"""
+Web article pull adapter (stub).
+
+Uses trafilatura to fetch and clean article text.
+Phase 1: basic fetch + text extraction.
+"""
+
+from __future__ import annotations
+
+from datetime import datetime
+from typing import Optional
+from urllib.parse import urlparse
+
+import trafilatura
+
+from second_brain.config import Config
+from second_brain.models import Source, SourceStatus
+
+
+class ArticleAdapter:
+    """Fetch and extract text from a web article URL."""
+
+    def __init__(self, config: Config) -> None:
+        self.config = config
+
+    def pull(self, source: Source) -> bool:
+        """Fetch the article and extract its text content.
+
+        Sets source.transcript_text, source.title, and source.status.
+        Returns True on success.
+        """
+        try:
+            downloaded = trafilatura.fetch_url(source.url)
+            if not downloaded:
+                source.error_message = f"Could not fetch URL: {source.url}"
+                source.status = SourceStatus.FAILED
+                return False
+
+            text = trafilatura.extract(
+                downloaded,
+                include_comments=False,
+                include_tables=True,
+                no_fallback=False,
+            )
+            if not text:
+                source.error_message = "trafilatura returned empty text"
+                source.status = SourceStatus.FAILED
+                return False
+
+            meta = trafilatura.extract_metadata(downloaded)
+            if meta:
+                source.title = source.title or meta.title
+                source.author = meta.author
+                if meta.date:
+                    try:
+                        source.published_at = datetime.fromisoformat(meta.date)
+                    except ValueError:
+                        pass
+
+            source.transcript_text = text
+            source.status = SourceStatus.TRANSCRIBED  # articles skip PULLED state
+            return True
+
+        except Exception as exc:
+            source.error_message = str(exc)
+            source.status = SourceStatus.FAILED
+            return False
+
+    @staticmethod
+    def looks_like_article(url: str) -> bool:
+        """Heuristic: does this URL look like a web article (not a video)?"""
+        parsed = urlparse(url)
+        video_hosts = {
+            "youtube.com", "www.youtube.com",
+            "youtu.be",
+            "vimeo.com", "www.vimeo.com",
+        }
+        return parsed.netloc not in video_hosts
+
+
+__all__ = ["ArticleAdapter"]

src/second_brain/adapters/youtube.py

@@ -0,0 +1,232 @@
+"""
+YouTube pull + Whisper transcribe adapter.
+
+Ported from xtract/main.py (VideoDownloader + SubtitleExtractor classes).
+Adapted to work with the second-brain Source model and Config.
+"""
+
+from __future__ import annotations
+
+import json
+from datetime import datetime
+from pathlib import Path
+from typing import Optional
+
+import whisper
+import yt_dlp
+
+from second_brain.config import Config
+from second_brain.models import Source, SourceStatus
+
+
+class VideoDownloader:
+    """Download a YouTube video (or playlist) via yt-dlp.
+
+    Ported from xtract VideoDownloader — same core yt-dlp options,
+    adapted to write into config.media_dir.
+    """
+
+    def __init__(self, config: Config) -> None:
+        self.config = config
+
+    def download(self, url: str) -> list[Path]:
+        """Download video(s) and return paths to the downloaded files."""
+        config = self.config
+        config.media_dir.mkdir(parents=True, exist_ok=True)
+
+        ydl_opts = {
+            "format": "bestvideo[height<=720]+bestaudio/best[height<=720]",
+            "outtmpl": str(config.media_dir / "%(id)s.%(ext)s"),
+            "writeinfojson": True,
+            "ignoreerrors": True,
+            "no_warnings": False,
+            "quiet": False,
+        }
+
+        downloaded: list[Path] = []
+
+        with yt_dlp.YoutubeDL(ydl_opts) as ydl:
+            print(f"[yt-dlp] Fetching: {url}")
+            info = ydl.extract_info(url, download=True)
+
+            if info is None:
+                print(f"[yt-dlp] Error: could not extract info from {url}")
+                return downloaded
+
+            if "entries" in info:
+                for entry in info["entries"]:
+                    if entry:
+                        p = self._find_media_file(entry.get("id", ""))
+                        if p:
+                            downloaded.append(p)
+            else:
+                p = self._find_media_file(info.get("id", ""))
+                if p:
+                    downloaded.append(p)
+
+        return downloaded
+
+    def _find_media_file(self, video_id: str) -> Optional[Path]:
+        for f in self.config.media_dir.glob(f"{video_id}.*"):
+            if f.suffix in {".mp4", ".mkv", ".webm", ".m4a"}:
+                return f
+        return None
+
+    def extract_metadata(self, url: str) -> dict:
+        """Fetch video metadata without downloading the media file."""
+        ydl_opts = {"quiet": True, "skip_download": True}
+        with yt_dlp.YoutubeDL(ydl_opts) as ydl:
+            info = ydl.extract_info(url, download=False) or {}
+        return {
+            "title": info.get("title"),
+            "author": info.get("uploader"),
+            "duration_seconds": info.get("duration"),
+            "published_at": _parse_upload_date(info.get("upload_date")),
+        }
+
+
+class SubtitleExtractor:
+    """Transcribe a video file to an SRT subtitle file using OpenAI Whisper.
+
+    Ported from xtract SubtitleExtractor — identical core logic,
+    writes SRTs to config.subtitles_dir.
+    """
+
+    def __init__(self, config: Config) -> None:
+        self.config = config
+        self._model = None
+
+    def _load_model(self) -> None:
+        if self._model is None:
+            print(f"[Whisper] Loading {self.config.whisper_model} model…")
+            self._model = whisper.load_model(self.config.whisper_model)
+            print("[Whisper] Model ready")
+
+    def extract(self, media_file: Path) -> Optional[Path]:
+        """Transcribe *media_file* and return the path to the SRT file."""
+        self.config.subtitles_dir.mkdir(parents=True, exist_ok=True)
+        srt_file = self.config.subtitles_dir / f"{media_file.stem}.srt"
+
+        if srt_file.exists():
+            print(f"[Whisper] Already transcribed: {srt_file.name}")
+            return srt_file
+
+        self._load_model()
+        print(f"[Whisper] Transcribing: {media_file.name}")
+
+        try:
+            result = self._model.transcribe(
+                str(media_file),
+                verbose=False,
+                language=None,
+            )
+            self._write_srt(result, srt_file)
+            print(f"[Whisper] Saved: {srt_file.name}")
+            return srt_file
+        except Exception as exc:
+            print(f"[Whisper] Error transcribing {media_file.name}: {exc}")
+            return None
+
+    def extract_text(self, media_file: Path) -> Optional[str]:
+        """Return the plain-text transcript (no SRT formatting)."""
+        self._load_model()
+        try:
+            result = self._model.transcribe(str(media_file), verbose=False)
+            return " ".join(seg["text"].strip() for seg in result["segments"])
+        except Exception as exc:
+            print(f"[Whisper] Error: {exc}")
+            return None
+
+    @staticmethod
+    def _write_srt(result: dict, output_file: Path) -> None:
+        with open(output_file, "w", encoding="utf-8") as fh:
+            for i, seg in enumerate(result["segments"], start=1):
+                start = _fmt_timestamp(seg["start"])
+                end = _fmt_timestamp(seg["end"])
+                fh.write(f"{i}\n{start} --> {end}\n{seg['text'].strip()}\n\n")
+
+    @staticmethod
+    def read_srt(srt_file: Path) -> str:
+        """Read an SRT file and return its raw text content."""
+        return srt_file.read_text(encoding="utf-8")
+
+
+# ---------------------------------------------------------------------------
+# High-level adapter
+# ---------------------------------------------------------------------------
+
+
+class YouTubeAdapter:
+    """Orchestrates pull + transcribe for a YouTube Source record."""
+
+    def __init__(self, config: Config) -> None:
+        self.config = config
+        self.downloader = VideoDownloader(config)
+        self.transcriber = SubtitleExtractor(config)
+
+    def pull(self, source: Source) -> bool:
+        """Download the video.  Updates source in-place, returns success."""
+        try:
+            meta = self.downloader.extract_metadata(source.url)
+            source.title = source.title or meta.get("title")
+            source.author = meta.get("author")
+            source.duration_seconds = meta.get("duration_seconds")
+            source.published_at = meta.get("published_at")
+
+            files = self.downloader.download(source.url)
+            if not files:
+                source.error_message = "yt-dlp returned no files"
+                source.status = SourceStatus.FAILED
+                return False
+
+            source.media_path = str(files[0])
+            source.status = SourceStatus.PULLED
+            return True
+        except Exception as exc:
+            source.error_message = str(exc)
+            source.status = SourceStatus.FAILED
+            return False
+
+    def transcribe(self, source: Source) -> bool:
+        """Transcribe the downloaded video.  Updates source in-place."""
+        if not source.media_path:
+            source.error_message = "No media_path; run pull first"
+            return False
+
+        media_file = Path(source.media_path)
+        srt_file = self.transcriber.extract(media_file)
+        if srt_file is None:
+            source.error_message = "Whisper transcription failed"
+            source.status = SourceStatus.FAILED
+            return False
+
+        source.transcript_path = str(srt_file)
+        source.transcript_text = SubtitleExtractor.read_srt(srt_file)
+        source.status = SourceStatus.TRANSCRIBED
+        return True
+
+
+# ---------------------------------------------------------------------------
+# Helpers
+# ---------------------------------------------------------------------------
+
+
+def _fmt_timestamp(seconds: float) -> str:
+    h = int(seconds // 3600)
+    m = int((seconds % 3600) // 60)
+    s = int(seconds % 60)
+    ms = int((seconds % 1) * 1000)
+    return f"{h:02d}:{m:02d}:{s:02d},{ms:03d}"
+
+
+def _parse_upload_date(upload_date: Optional[str]) -> Optional[datetime]:
+    """Parse yt-dlp's YYYYMMDD upload_date string."""
+    if not upload_date or len(upload_date) != 8:
+        return None
+    try:
+        return datetime.strptime(upload_date, "%Y%m%d")
+    except ValueError:
+        return None
+
+
+__all__ = ["VideoDownloader", "SubtitleExtractor", "YouTubeAdapter"]

src/second_brain/compiler/__init__.py

@@ -0,0 +1,5 @@
+"""Wiki compiler — folds accepted extractions into the Obsidian vault."""
+
+from second_brain.compiler.wiki import WikiCompiler
+
+__all__ = ["WikiCompiler"]

src/second_brain/compiler/wiki.py

@@ -0,0 +1,244 @@
+"""
+Wiki compiler (Karpathy LLM Wiki pattern).
+
+Reads the accepted extractions from the DB, reads the existing Obsidian vault
+pages, calls Claude to decide what to create/update, then writes the pages and
+git-commits the vault before and after.
+
+Phase 1 implementation:
+  - Generates one Markdown page per accepted source (slug from title/url).
+  - Stubs the git commit logic with clear interfaces.
+  - Does NOT yet do cross-source synthesis (that is Phase 2).
+"""
+
+from __future__ import annotations
+
+import re
+import subprocess
+from datetime import datetime
+from pathlib import Path
+from typing import Optional
+
+import anthropic
+
+from second_brain.config import Config
+from second_brain.database import get_database
+from second_brain.models import Extraction, Source, SourceStatus, WikiPage
+
+
+class WikiCompiler:
+    """
+    Compiles accepted extractions into Obsidian vault Markdown pages.
+
+    Usage::
+
+        compiler = WikiCompiler(config)
+        stats = compiler.run()
+    """
+
+    def __init__(self, config: Config, api_key: Optional[str] = None) -> None:
+        self.config = config
+        self._api_key = api_key or config.anthropic_api_key
+
+    def run(self) -> dict:
+        """
+        Main entry point.  Processes all accepted (not yet published) sources.
+
+        Returns a stats dict: {processed, created, updated, errors}.
+        """
+        vault = self.config.vault_path
+        vault.mkdir(parents=True, exist_ok=True)
+
+        db = get_database()
+        stats = {"processed": 0, "created": 0, "updated": 0, "errors": 0}
+
+        with db.session() as sess:
+            sources = (
+                sess.query(Source)
+                .filter(Source.status == SourceStatus.ACCEPTED)
+                .all()
+            )
+
+            if not sources:
+                print("[Compiler] No accepted sources to compile.")
+                return stats
+
+            sha_before = self._git_sha(vault)
+
+            for source in sources:
+                try:
+                    created = self._compile_source(source, sess, vault)
+                    if created:
+                        stats["created"] += 1
+                    else:
+                        stats["updated"] += 1
+                    source.status = SourceStatus.PUBLISHED
+                    source.updated_at = datetime.utcnow()
+                    stats["processed"] += 1
+                except Exception as exc:
+                    print(f"[Compiler] Error processing source {source.id}: {exc}")
+                    stats["errors"] += 1
+
+            sha_after = self._git_commit(vault, stats["processed"])
+
+            # Update git sha on all written WikiPage rows
+            if sha_after:
+                for page in sess.query(WikiPage).filter(
+                    WikiPage.git_sha_after.is_(None)
+                ).all():
+                    page.git_sha_before = sha_before
+                    page.git_sha_after = sha_after
+
+        return stats
+
+    # ------------------------------------------------------------------
+    # Per-source page generation
+    # ------------------------------------------------------------------
+
+    def _compile_source(
+        self, source: Source, sess, vault: Path
+    ) -> bool:
+        """Write a vault page for *source*.  Returns True if newly created."""
+        extraction: Optional[Extraction] = source.extraction
+        if extraction is None:
+            raise ValueError(f"Source {source.id} has no extraction")
+
+        slug = _slugify(source.title or source.url)
+        domain_dir = vault / source.domain
+        domain_dir.mkdir(exist_ok=True)
+        page_path = domain_dir / f"{slug}.md"
+        relative_path = str(page_path.relative_to(vault))
+
+        content = self._render_page(source, extraction)
+
+        created = not page_path.exists()
+        page_path.write_text(content, encoding="utf-8")
+        print(f"[Compiler] {'Created' if created else 'Updated'}: {relative_path}")
+
+        # Upsert WikiPage tracking row
+        wiki_row = sess.query(WikiPage).filter_by(vault_path=relative_path).first()
+        if wiki_row is None:
+            wiki_row = WikiPage(
+                vault_path=relative_path,
+                domain=source.domain,
+                title=source.title or slug,
+            )
+            sess.add(wiki_row)
+        else:
+            wiki_row.updated_at = datetime.utcnow()
+
+        return created
+
+    @staticmethod
+    def _render_page(source: Source, extraction: Extraction) -> str:
+        """Render the Markdown content for a vault page."""
+        lines: list[str] = []
+
+        title = source.title or source.url
+        lines.append(f"# {title}\n")
+
+        # Front-matter block
+        lines.append("---")
+        lines.append(f"url: {source.url}")
+        lines.append(f"domain: {source.domain}")
+        lines.append(f"source_type: {source.source_type.value}")
+        if source.published_at:
+            lines.append(f"published: {source.published_at.date()}")
+        lines.append(f"ingested: {source.ingested_at.date()}")
+        if source.focus:
+            lines.append(f"focus: {source.focus}")
+        lines.append("---\n")
+
+        # Summary
+        if extraction.summary:
+            lines.append("## Summary\n")
+            lines.append(extraction.summary + "\n")
+
+        # Key points
+        if extraction.key_points:
+            lines.append("## Key Points\n")
+            for pt in extraction.key_points:
+                lines.append(f"- {pt}")
+            lines.append("")
+
+        # Claims
+        if extraction.claims:
+            lines.append("## Claims\n")
+            for c in extraction.claims:
+                lines.append(f"- {c}")
+            lines.append("")
+
+        # Entities
+        if extraction.entities:
+            lines.append("## Entities\n")
+            lines.append(", ".join(f"[[{e}]]" for e in extraction.entities) + "\n")
+
+        # Open questions
+        if extraction.open_questions:
+            lines.append("## Open Questions\n")
+            for q in extraction.open_questions:
+                lines.append(f"- {q}")
+            lines.append("")
+
+        # Contradictions
+        if extraction.contradictions:
+            lines.append("## Contradictions\n")
+            for ct in extraction.contradictions:
+                lines.append(f"- {ct}")
+            lines.append("")
+
+        return "\n".join(lines)
+
+    # ------------------------------------------------------------------
+    # Git helpers (stubbed — interface is clear for later wiring)
+    # ------------------------------------------------------------------
+
+    @staticmethod
+    def _git_sha(vault: Path) -> Optional[str]:
+        """Return HEAD sha of the vault git repo, or None if not a git repo."""
+        try:
+            result = subprocess.run(
+                ["git", "rev-parse", "HEAD"],
+                cwd=vault,
+                capture_output=True,
+                text=True,
+                check=True,
+            )
+            return result.stdout.strip()
+        except (subprocess.CalledProcessError, FileNotFoundError):
+            return None
+
+    @staticmethod
+    def _git_commit(vault: Path, count: int) -> Optional[str]:
+        """Stage all changes and commit.  Returns new HEAD sha or None."""
+        try:
+            subprocess.run(["git", "add", "-A"], cwd=vault, check=True)
+            msg = f"second-brain: compile {count} source(s) [{datetime.utcnow().date()}]"
+            subprocess.run(["git", "commit", "-m", msg], cwd=vault, check=True)
+            result = subprocess.run(
+                ["git", "rev-parse", "HEAD"],
+                cwd=vault,
+                capture_output=True,
+                text=True,
+                check=True,
+            )
+            return result.stdout.strip()
+        except (subprocess.CalledProcessError, FileNotFoundError):
+            return None
+
+
+# ---------------------------------------------------------------------------
+# Helpers
+# ---------------------------------------------------------------------------
+
+
+def _slugify(text: str) -> str:
+    """Convert a title to a filesystem-safe slug."""
+    text = text.lower()
+    text = re.sub(r"[^\w\s-]", "", text)
+    text = re.sub(r"[\s_-]+", "-", text)
+    text = text.strip("-")
+    return text[:100]
+
+
+__all__ = ["WikiCompiler"]

src/second_brain/config.py

@@ -0,0 +1,135 @@
+"""
+Configuration loading for second-brain.
+
+Settings are read from config/settings.toml (or a path set via SECOND_BRAIN_CONFIG).
+All values have sensible defaults so the system works out of the box.
+"""
+
+from __future__ import annotations
+
+import os
+from pathlib import Path
+from typing import Any
+
+try:
+    import tomllib  # Python 3.11+
+except ModuleNotFoundError:
+    import tomli as tomllib  # type: ignore[no-reuse-impl]
+
+# ---------------------------------------------------------------------------
+# Defaults
+# ---------------------------------------------------------------------------
+
+_DEFAULTS: dict[str, Any] = {
+    "db_path": "~/.local/share/second-brain/second_brain.db",
+    "vault_path": "~/Documents/second-brain-vault",
+    "media_dir": "~/.local/share/second-brain/media",
+    "subtitles_dir": "~/.local/share/second-brain/subtitles",
+    "whisper_model": "small",
+    "domains": ["development", "content", "business", "homelab"],
+    "scheduler": {
+        "window_start": "22:00",
+        "window_end": "06:00",
+        "max_tokens_per_hour": 100_000,
+        "min_gap_seconds": 120,
+    },
+}
+
+DOMAINS = ["development", "content", "business", "homelab"]
+
+
+class Config:
+    """Holds resolved configuration values."""
+
+    def __init__(self, raw: dict[str, Any]) -> None:
+        self._raw = raw
+
+    # --- paths ---
+
+    @property
+    def db_path(self) -> Path:
+        return Path(self._raw.get("db_path", _DEFAULTS["db_path"])).expanduser()
+
+    @property
+    def vault_path(self) -> Path:
+        return Path(self._raw.get("vault_path", _DEFAULTS["vault_path"])).expanduser()
+
+    @property
+    def media_dir(self) -> Path:
+        return Path(self._raw.get("media_dir", _DEFAULTS["media_dir"])).expanduser()
+
+    @property
+    def subtitles_dir(self) -> Path:
+        return Path(
+            self._raw.get("subtitles_dir", _DEFAULTS["subtitles_dir"])
+        ).expanduser()
+
+    # --- extraction ---
+
+    @property
+    def whisper_model(self) -> str:
+        return self._raw.get("whisper_model", _DEFAULTS["whisper_model"])
+
+    @property
+    def domains(self) -> list[str]:
+        return self._raw.get("domains", _DEFAULTS["domains"])
+
+    @property
+    def anthropic_api_key(self) -> str | None:
+        return os.getenv("ANTHROPIC_API_KEY")
+
+    # --- scheduler ---
+
+    @property
+    def scheduler(self) -> dict[str, Any]:
+        return {**_DEFAULTS["scheduler"], **self._raw.get("scheduler", {})}
+
+    # --- prompts dir ---
+
+    @property
+    def prompts_dir(self) -> Path:
+        """Directory containing per-domain prompt templates."""
+        here = Path(__file__).parent.parent.parent.parent  # project root
+        return here / "prompts"
+
+    def ensure_dirs(self) -> None:
+        """Create runtime directories if they don't exist."""
+        for d in (self.db_path.parent, self.media_dir, self.subtitles_dir):
+            d.mkdir(parents=True, exist_ok=True)
+
+
+# ---------------------------------------------------------------------------
+# Loader
+# ---------------------------------------------------------------------------
+
+_config_instance: Config | None = None
+
+
+def load_config(path: Path | None = None) -> Config:
+    """Load (and cache) configuration from a TOML file."""
+    global _config_instance
+    if _config_instance is not None:
+        return _config_instance
+
+    if path is None:
+        env_path = os.getenv("SECOND_BRAIN_CONFIG")
+        if env_path:
+            path = Path(env_path)
+        else:
+            # Walk up from CWD looking for config/settings.toml
+            cwd = Path.cwd()
+            for candidate in [cwd / "config" / "settings.toml", cwd / "settings.toml"]:
+                if candidate.exists():
+                    path = candidate
+                    break
+
+    raw: dict[str, Any] = {}
+    if path and path.exists():
+        with open(path, "rb") as fh:
+            raw = tomllib.load(fh)
+
+    _config_instance = Config(raw)
+    return _config_instance
+
+
+__all__ = ["Config", "DOMAINS", "load_config"]

src/second_brain/context/__init__.py

@@ -0,0 +1,5 @@
+"""Context assembly layer."""
+
+from second_brain.context.assembler import ContextAssembler
+
+__all__ = ["ContextAssembler"]

src/second_brain/context/assembler.py

@@ -0,0 +1,88 @@
+"""
+Context assembler — Phase 1 (thin).
+
+Builds the user-turn prompt bundle that gets sent to the extraction engine.
+Phase 1 bundle = domain template + optional focus directive + transcript text.
+
+Future phases can layer in wiki context, related sources, etc.
+"""
+
+from __future__ import annotations
+
+from pathlib import Path
+from typing import Optional
+
+from second_brain.config import Config
+from second_brain.models import Source
+
+
+_FALLBACK_TEMPLATE = """\
+You are extracting structured knowledge from the following source.
+Domain: {domain}
+{focus_section}
+
+Return a JSON object with these keys:
+- summary (string): 2-4 sentence overview
+- key_points (list of strings): 5-10 takeaways
+- entities (list of strings): notable people, tools, products, concepts
+- claims (list of strings): specific factual assertions
+- open_questions (list of strings): questions raised but unanswered
+- contradictions (list of strings): points that contradict known facts
+
+Source transcript / content:
+{transcript}
+"""
+
+
+class ContextAssembler:
+    """Assembles the prompt bundle for a single source."""
+
+    def __init__(self, config: Config) -> None:
+        self.config = config
+        self._template_cache: dict[str, str] = {}
+
+    def assemble(self, source: Source) -> str:
+        """Return the full user-turn prompt string for the extraction engine."""
+        template = self._load_template(source.domain)
+        transcript = source.transcript_text or ""
+
+        focus_section = ""
+        if source.focus:
+            focus_section = f"Focus area: {source.focus}\n"
+
+        return template.format(
+            domain=source.domain,
+            focus_section=focus_section,
+            transcript=transcript,
+        )
+
+    def _load_template(self, domain: str) -> str:
+        """Load and cache the domain prompt template from prompts/<domain>.md."""
+        if domain in self._template_cache:
+            return self._template_cache[domain]
+
+        template_path = self.config.prompts_dir / f"{domain}.md"
+        if template_path.exists():
+            raw = template_path.read_text(encoding="utf-8")
+            template = self._extract_template_body(raw)
+        else:
+            print(
+                f"[Assembler] Warning: no prompt template for domain {domain!r}, "
+                "using fallback"
+            )
+            template = _FALLBACK_TEMPLATE
+
+        self._template_cache[domain] = template
+        return template
+
+    @staticmethod
+    def _extract_template_body(raw: str) -> str:
+        """Strip YAML front-matter if present, return body."""
+        if raw.startswith("---"):
+            parts = raw.split("---", 2)
+            if len(parts) >= 3:
+                return parts[2].strip()
+        return raw.strip()
+
+
+__all__ = ["ContextAssembler"]

src/second_brain/database.py

@@ -0,0 +1,81 @@
+"""
+Database engine and session management for second-brain.
+
+Ported and adapted from xtract/database.py.
+Uses a singleton pattern: call get_database() everywhere.
+"""
+
+from __future__ import annotations
+
+from contextlib import contextmanager
+from pathlib import Path
+from typing import Generator
+
+from sqlalchemy import create_engine
+from sqlalchemy.orm import Session, sessionmaker
+
+from second_brain.models import Base
+
+
+class Database:
+    """Manages the SQLAlchemy engine and session factory."""
+
+    def __init__(self, db_path: Path) -> None:
+        self.db_path = db_path
+        self.db_path.parent.mkdir(parents=True, exist_ok=True)
+
+        self.engine = create_engine(
+            f"sqlite:///{self.db_path}",
+            echo=False,
+            connect_args={"check_same_thread": False},
+        )
+
+        self.SessionLocal = sessionmaker(
+            autocommit=False,
+            autoflush=False,
+            bind=self.engine,
+        )
+
+    def init_db(self) -> None:
+        """Create all tables (idempotent)."""
+        Base.metadata.create_all(bind=self.engine)
+
+    def get_session(self) -> Session:
+        """Return a new bare session (caller must close it)."""
+        return self.SessionLocal()
+
+    @contextmanager
+    def session(self) -> Generator[Session, None, None]:
+        """Context-manager that commits on exit and rolls back on error."""
+        sess = self.SessionLocal()
+        try:
+            yield sess
+            sess.commit()
+        except Exception:
+            sess.rollback()
+            raise
+        finally:
+            sess.close()
+
+
+# ---------------------------------------------------------------------------
+# Singleton
+# ---------------------------------------------------------------------------
+
+_db_instance: Database | None = None
+
+
+def get_database(db_path: Path | None = None) -> Database:
+    """Return the singleton Database, creating and initialising it on first call."""
+    global _db_instance
+    if _db_instance is None:
+        if db_path is None:
+            from second_brain.config import load_config
+
+            db_path = load_config().db_path
+        _db_instance = Database(db_path)
+        _db_instance.init_db()
+    return _db_instance
+
+
+__all__ = ["Database", "get_database"]

src/second_brain/extractor/__init__.py

@@ -0,0 +1,6 @@
+"""Stateless LLM extraction layer."""
+
+from second_brain.extractor.engine import ExtractionEngine
+from second_brain.extractor.schema import ExtractionResult, SourceMeta
+
+__all__ = ["ExtractionEngine", "ExtractionResult", "SourceMeta"]

src/second_brain/extractor/engine.py

@@ -0,0 +1,140 @@
+"""
+Single-shot LLM extraction engine.
+
+Ported/refactored from xtract/analyzer.py.
+Pure function: context bundle in → ExtractionResult out.
+Uses the Anthropic SDK with a single messages.create() call.
+"""
+
+from __future__ import annotations
+
+import json
+import os
+from typing import Optional
+
+import anthropic
+
+from second_brain.extractor.schema import ExtractionResult, SourceMeta
+from second_brain.models import Source
+
+
+class ExtractionEngine:
+    """
+    Stateless LLM extractor.
+
+    Call extract() with a pre-assembled context bundle (prompt + transcript).
+    Returns a validated ExtractionResult or raises on failure.
+    """
+
+    MODEL = "claude-sonnet-4-6"
+    MAX_TOKENS = 4096
+
+    def __init__(self, api_key: Optional[str] = None) -> None:
+        key = api_key or os.getenv("ANTHROPIC_API_KEY")
+        if not key:
+            raise ValueError(
+                "ANTHROPIC_API_KEY is required. Set the environment variable or "
+                "pass api_key to ExtractionEngine."
+            )
+        self.client = anthropic.Anthropic(api_key=key)
+
+    def extract(self, prompt: str, source: Source) -> ExtractionResult:
+        """
+        Run a single-shot extraction call.
+
+        Args:
+            prompt: The fully assembled context bundle (system + transcript).
+            source: The Source row (used to populate SourceMeta).
+
+        Returns:
+            Validated ExtractionResult.
+        """
+        print(f"[Extractor] Calling Claude for source {source.id} ({source.domain})…")
+
+        response = self.client.messages.create(
+            model=self.MODEL,
+            max_tokens=self.MAX_TOKENS,
+            system=self._system_prompt(),
+            messages=[{"role": "user", "content": prompt}],
+        )
+
+        raw_text = response.content[0].text
+        usage = response.usage
+
+        print(
+            f"[Extractor] Done — "
+            f"input={usage.input_tokens} output={usage.output_tokens} tokens"
+        )
+
+        parsed = self._parse(raw_text, source)
+
+        # Attach token counts to the result for the caller to store
+        parsed._input_tokens = usage.input_tokens
+        parsed._output_tokens = usage.output_tokens
+        parsed._raw_response = raw_text
+
+        return parsed
+
+    # ------------------------------------------------------------------
+    # Internal helpers
+    # ------------------------------------------------------------------
+
+    @staticmethod
+    def _system_prompt() -> str:
+        return (
+            "You are a knowledge extraction assistant. "
+            "Given a source (transcript or article text) and a domain-specific prompt, "
+            "extract structured insights and return them as a single JSON object. "
+            "Return ONLY the JSON object — no markdown fences, no preamble."
+        )
+
+    def _parse(self, raw: str, source: Source) -> ExtractionResult:
+        """Parse Claude's JSON response into an ExtractionResult."""
+        text = raw.strip()
+        # Strip accidental markdown fences
+        if text.startswith("```"):
+            text = text.split("\n", 1)[-1]
+            text = text.rsplit("```", 1)[0].strip()
+
+        try:
+            data = json.loads(text)
+        except json.JSONDecodeError as exc:
+            raise ValueError(
+                f"Could not parse JSON from LLM response: {exc}\n"
+                f"Raw response (first 500 chars):\n{raw[:500]}"
+            ) from exc
+
+        meta = SourceMeta(
+            url=source.url,
+            title=source.title,
+            source_type=source.source_type.value,
+            duration_or_length=_fmt_duration(source.duration_seconds),
+            published_at=source.published_at.isoformat() if source.published_at else None,
+            ingested_at=source.ingested_at.isoformat(),
+        )
+
+        return ExtractionResult(
+            source=meta,
+            domain=source.domain,
+            focus=source.focus,
+            summary=data.get("summary", ""),
+            key_points=data.get("key_points", []),
+            entities=data.get("entities", []),
+            claims=data.get("claims", []),
+            open_questions=data.get("open_questions", []),
+            contradictions=data.get("contradictions", []),
+        )
+
+
+def _fmt_duration(seconds: Optional[int]) -> Optional[str]:
+    if seconds is None:
+        return None
+    h = seconds // 3600
+    m = (seconds % 3600) // 60
+    s = seconds % 60
+    if h:
+        return f"{h}:{m:02d}:{s:02d}"
+    return f"{m}:{s:02d}"
+
+
+__all__ = ["ExtractionEngine"]

src/second_brain/extractor/schema.py

@@ -0,0 +1,82 @@
+"""
+Pydantic schema for the extraction output contract.
+
+This is the canonical shape returned by the extractor engine and
+stored in the Extraction model as JSON columns.
+"""
+
+from __future__ import annotations
+
+from datetime import datetime
+from typing import Optional
+
+from pydantic import BaseModel, Field, HttpUrl
+
+
+class SourceMeta(BaseModel):
+    """Metadata about the source (populated from the Source row)."""
+
+    url: str
+    title: Optional[str] = None
+    source_type: str  # "video" | "article"
+    duration_or_length: Optional[str] = None  # e.g. "32:14" or "~1800 words"
+    published_at: Optional[str] = None
+    ingested_at: str
+
+
+class ExtractionResult(BaseModel):
+    """
+    Canonical extraction output.
+
+    Contract between the extractor engine and every downstream consumer
+    (web UI, wiki compiler, scheduler).
+    """
+
+    source: SourceMeta
+
+    domain: str = Field(description="One of: development, content, business, homelab")
+    focus: Optional[str] = Field(
+        default=None,
+        description="Optional focus directive supplied when the source was queued",
+    )
+
+    summary: str = Field(description="2-4 sentence summary of the source")
+
+    key_points: list[str] = Field(
+        default_factory=list,
+        description="Bulleted key takeaways (5-10 items)",
+    )
+
+    entities: list[str] = Field(
+        default_factory=list,
+        description="People, tools, products, companies, concepts mentioned",
+    )
+
+    claims: list[str] = Field(
+        default_factory=list,
+        description="Specific factual claims or assertions made by the source",
+    )
+
+    open_questions: list[str] = Field(
+        default_factory=list,
+        description="Questions raised but not answered, or worth investigating further",
+    )
+
+    contradictions: list[str] = Field(
+        default_factory=list,
+        description="Points that contradict known facts or other sources",
+    )
+
+    def to_db_dict(self) -> dict:
+        """Return the subset of fields stored in the Extraction table."""
+        return {
+            "summary": self.summary,
+            "key_points": self.key_points,
+            "entities": self.entities,
+            "claims": self.claims,
+            "open_questions": self.open_questions,
+            "contradictions": self.contradictions,
+        }
+
+
+__all__ = ["ExtractionResult", "SourceMeta"]

src/second_brain/main.py

@@ -0,0 +1,291 @@
+"""
+second-brain CLI entry point.
+
+Commands:
+  add       — queue a source URL (video or article)
+  process   — run the full pipeline on pending/pulled/transcribed sources
+  serve     — launch the FastAPI web UI
+  compile   — run the wiki compiler on accepted sources
+  schedule  — start the overnight scheduler
+"""
+
+from __future__ import annotations
+
+import sys
+from datetime import datetime
+from pathlib import Path
+from typing import Optional
+from urllib.parse import urlparse
+
+import click
+
+from second_brain.config import DOMAINS, load_config
+from second_brain.database import get_database
+from second_brain.models import Source, SourceStatus, SourceType
+
+
+# ---------------------------------------------------------------------------
+# CLI group
+# ---------------------------------------------------------------------------
+
+
+@click.group()
+def cli() -> None:
+    """second-brain — personal knowledge extraction pipeline."""
+
+
+# ---------------------------------------------------------------------------
+# add
+# ---------------------------------------------------------------------------
+
+
+@cli.command()
+@click.argument("url")
+@click.option(
+    "--domain",
+    "-d",
+    type=click.Choice(DOMAINS),
+    default="development",
+    show_default=True,
+    help="Knowledge domain for extraction.",
+)
+@click.option("--focus", "-f", default=None, help="Optional focus directive.")
+@click.option("--title", "-t", default=None, help="Override title.")
+def add(url: str, domain: str, focus: Optional[str], title: Optional[str]) -> None:
+    """Queue a source URL for processing."""
+    config = load_config()
+    config.ensure_dirs()
+    db = get_database(config.db_path)
+
+    source_type = SourceType.ARTICLE if _is_article(url) else SourceType.VIDEO
+
+    with db.session() as sess:
+        existing = sess.query(Source).filter_by(url=url).first()
+        if existing:
+            click.echo(
+                f"[add] Already queued (id={existing.id}, status={existing.status.value})"
+            )
+            return
+
+        source = Source(
+            url=url,
+            title=title,
+            domain=domain,
+            focus=focus,
+            source_type=source_type,
+            status=SourceStatus.PENDING,
+            ingested_at=datetime.utcnow(),
+        )
+        sess.add(source)
+
+    click.echo(f"[add] Queued {source_type.value}: {url}")
+    click.echo(f"      domain={domain}" + (f"  focus={focus}" if focus else ""))
+
+
+# ---------------------------------------------------------------------------
+# process
+# ---------------------------------------------------------------------------
+
+
+@cli.command()
+@click.option("--limit", "-n", default=None, type=int, help="Max sources to process.")
+@click.option(
+    "--skip-pull", is_flag=True, default=False, help="Skip the pull/download step."
+)
+@click.option(
+    "--skip-transcribe",
+    is_flag=True,
+    default=False,
+    help="Skip the transcription step.",
+)
+@click.option(
+    "--skip-extract", is_flag=True, default=False, help="Skip the LLM extraction step."
+)
+def process(
+    limit: Optional[int],
+    skip_pull: bool,
+    skip_transcribe: bool,
+    skip_extract: bool,
+) -> None:
+    """Run the pipeline on pending sources."""
+    from second_brain.adapters.youtube import YouTubeAdapter
+    from second_brain.adapters.article import ArticleAdapter
+    from second_brain.context.assembler import ContextAssembler
+    from second_brain.extractor.engine import ExtractionEngine
+    from second_brain.models import Extraction
+
+    config = load_config()
+    config.ensure_dirs()
+    db = get_database(config.db_path)
+
+    yt_adapter = YouTubeAdapter(config)
+    art_adapter = ArticleAdapter(config)
+    assembler = ContextAssembler(config)
+
+    try:
+        engine = ExtractionEngine(api_key=config.anthropic_api_key)
+    except ValueError as exc:
+        if not skip_extract:
+            click.echo(f"[process] Warning: {exc}")
+            click.echo("[process] Extraction step will be skipped.")
+            skip_extract = True
+        engine = None
+
+    with db.session() as sess:
+        query = sess.query(Source).filter(
+            Source.status.in_([
+                SourceStatus.PENDING,
+                SourceStatus.PULLED,
+                SourceStatus.TRANSCRIBED,
+            ])
+        )
+        if limit:
+            query = query.limit(limit)
+        sources = query.all()
+
+    if not sources:
+        click.echo("[process] No sources to process.")
+        return
+
+    click.echo(f"[process] Processing {len(sources)} source(s)…")
+
+    for source in sources:
+        click.echo(f"\n  [{source.id}] {source.url[:80]}")
+
+        with db.session() as sess:
+            source = sess.query(Source).get(source.id)
+
+            # Pull step
+            if not skip_pull and source.status == SourceStatus.PENDING:
+                click.echo("    → pull")
+                if source.source_type == SourceType.VIDEO:
+                    ok = yt_adapter.pull(source)
+                else:
+                    ok = art_adapter.pull(source)
+                if not ok:
+                    click.echo(f"    ✗ pull failed: {source.error_message}")
+                    continue
+
+            # Transcribe step (videos only — articles come out transcribed)
+            if (
+                not skip_transcribe
+                and source.status == SourceStatus.PULLED
+                and source.source_type == SourceType.VIDEO
+            ):
+                click.echo("    → transcribe")
+                ok = yt_adapter.transcribe(source)
+                if not ok:
+                    click.echo(f"    ✗ transcribe failed: {source.error_message}")
+                    continue
+
+            # Extract step
+            if not skip_extract and source.status == SourceStatus.TRANSCRIBED and engine:
+                click.echo("    → extract")
+                try:
+                    prompt = assembler.assemble(source)
+                    result = engine.extract(prompt, source)
+
+                    existing_ext = sess.query(Extraction).filter_by(
+                        source_id=source.id
+                    ).first()
+                    if existing_ext:
+                        for k, v in result.to_db_dict().items():
+                            setattr(existing_ext, k, v)
+                        existing_ext.raw_response = getattr(result, "_raw_response", None)
+                        existing_ext.input_tokens = getattr(result, "_input_tokens", None)
+                        existing_ext.output_tokens = getattr(result, "_output_tokens", None)
+                    else:
+                        extraction = Extraction(
+                            source_id=source.id,
+                            raw_response=getattr(result, "_raw_response", None),
+                            input_tokens=getattr(result, "_input_tokens", None),
+                            output_tokens=getattr(result, "_output_tokens", None),
+                            **result.to_db_dict(),
+                        )
+                        sess.add(extraction)
+
+                    source.status = SourceStatus.ANALYZED
+                    source.updated_at = datetime.utcnow()
+                    click.echo("    ✓ analyzed")
+                except Exception as exc:
+                    source.error_message = str(exc)
+                    source.status = SourceStatus.FAILED
+                    click.echo(f"    ✗ extraction failed: {exc}")
+
+    click.echo("\n[process] Done.")
+
+
+# ---------------------------------------------------------------------------
+# serve
+# ---------------------------------------------------------------------------
+
+
+@cli.command()
+@click.option("--host", default="127.0.0.1", show_default=True)
+@click.option("--port", default=8000, show_default=True, type=int)
+@click.option("--reload", is_flag=True, default=False)
+def serve(host: str, port: int, reload: bool) -> None:
+    """Launch the web UI."""
+    import uvicorn
+
+    click.echo(f"[serve] Starting on http://{host}:{port}")
+    uvicorn.run(
+        "second_brain.web.app:app",
+        host=host,
+        port=port,
+        reload=reload,
+    )
+
+
+# ---------------------------------------------------------------------------
+# compile
+# ---------------------------------------------------------------------------
+
+
+@cli.command("compile")
+def compile_wiki() -> None:
+    """Run the wiki compiler on accepted sources."""
+    from second_brain.compiler.wiki import WikiCompiler
+
+    config = load_config()
+    compiler = WikiCompiler(config)
+    stats = compiler.run()
+    click.echo(
+        f"[compile] processed={stats['processed']} "
+        f"created={stats['created']} updated={stats['updated']} "
+        f"errors={stats['errors']}"
+    )
+
+
+# ---------------------------------------------------------------------------
+# schedule
+# ---------------------------------------------------------------------------
+
+
+@cli.command()
+@click.option("--dry-run", is_flag=True, default=False)
+def schedule(dry_run: bool) -> None:
+    """Start the overnight scheduler."""
+    from second_brain.scheduler.runner import Scheduler
+
+    config = load_config()
+    scheduler = Scheduler(config)
+    scheduler.run(dry_run=dry_run)
+
+
+# ---------------------------------------------------------------------------
+# Helpers
+# ---------------------------------------------------------------------------
+
+
+def _is_article(url: str) -> bool:
+    parsed = urlparse(url)
+    video_hosts = {
+        "youtube.com", "www.youtube.com", "youtu.be",
+        "vimeo.com", "www.vimeo.com",
+    }
+    return parsed.netloc not in video_hosts
+
+
+if __name__ == "__main__":
+    cli()

src/second_brain/models.py

@@ -0,0 +1,170 @@
+"""
+SQLAlchemy ORM models for second-brain.
+
+Tables:
+- Source      — a queued/processed URL (video or article)
+- Extraction  — structured LLM output for a source
+- WikiPage    — track which vault pages have been written / updated
+"""
+
+from __future__ import annotations
+
+import enum
+from datetime import datetime
+from typing import Optional
+
+from sqlalchemy import (
+    DateTime,
+    Enum,
+    ForeignKey,
+    Integer,
+    JSON,
+    String,
+    Text,
+)
+from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, relationship
+
+
+class Base(DeclarativeBase):
+    """Shared declarative base."""
+
+
+# ---------------------------------------------------------------------------
+# Enums
+# ---------------------------------------------------------------------------
+
+
+class SourceType(enum.Enum):
+    VIDEO = "video"
+    ARTICLE = "article"
+
+
+class SourceStatus(enum.Enum):
+    PENDING = "pending"
+    PULLED = "pulled"
+    TRANSCRIBED = "transcribed"
+    ANALYZED = "analyzed"
+    ACCEPTED = "accepted"
+    PUBLISHED = "published"
+    FAILED = "failed"
+
+
+# ---------------------------------------------------------------------------
+# Models
+# ---------------------------------------------------------------------------
+
+
+class Source(Base):
+    """A URL queued for processing."""
+
+    __tablename__ = "sources"
+
+    id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
+
+    url: Mapped[str] = mapped_column(String(2000), unique=True, nullable=False)
+    title: Mapped[Optional[str]] = mapped_column(String(500), nullable=True)
+    source_type: Mapped[SourceType] = mapped_column(
+        Enum(SourceType), nullable=False, default=SourceType.VIDEO
+    )
+
+    # Domain and focus
+    domain: Mapped[str] = mapped_column(String(50), nullable=False, default="development")
+    focus: Mapped[Optional[str]] = mapped_column(Text, nullable=True)
+
+    # Pipeline status
+    status: Mapped[SourceStatus] = mapped_column(
+        Enum(SourceStatus), nullable=False, default=SourceStatus.PENDING
+    )
+
+    # File paths (set after pull/transcribe steps)
+    media_path: Mapped[Optional[str]] = mapped_column(String(1000), nullable=True)
+    transcript_path: Mapped[Optional[str]] = mapped_column(String(1000), nullable=True)
+    transcript_text: Mapped[Optional[str]] = mapped_column(Text, nullable=True)
+
+    # Metadata from the source
+    published_at: Mapped[Optional[datetime]] = mapped_column(DateTime, nullable=True)
+    duration_seconds: Mapped[Optional[int]] = mapped_column(Integer, nullable=True)
+    author: Mapped[Optional[str]] = mapped_column(String(200), nullable=True)
+
+    # Timestamps
+    ingested_at: Mapped[datetime] = mapped_column(DateTime, default=datetime.utcnow)
+    updated_at: Mapped[Optional[datetime]] = mapped_column(DateTime, nullable=True)
+    error_message: Mapped[Optional[str]] = mapped_column(Text, nullable=True)
+
+    # Relationships
+    extraction: Mapped[Optional["Extraction"]] = relationship(
+        "Extraction", back_populates="source", uselist=False, cascade="all, delete-orphan"
+    )
+
+    def __repr__(self) -> str:
+        return (
+            f"Source(id={self.id}, domain={self.domain!r}, "
+            f"status={self.status.value}, url={self.url!r})"
+        )
+
+
+class Extraction(Base):
+    """Structured LLM extraction output for a source."""
+
+    __tablename__ = "extractions"
+
+    id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
+    source_id: Mapped[int] = mapped_column(
+        ForeignKey("sources.id", ondelete="CASCADE"), nullable=False, unique=True
+    )
+
+    # Core fields (mirrors the extraction schema contract)
+    summary: Mapped[Optional[str]] = mapped_column(Text, nullable=True)
+    key_points: Mapped[Optional[list]] = mapped_column(JSON, nullable=True)
+    entities: Mapped[Optional[list]] = mapped_column(JSON, nullable=True)
+    claims: Mapped[Optional[list]] = mapped_column(JSON, nullable=True)
+    open_questions: Mapped[Optional[list]] = mapped_column(JSON, nullable=True)
+    contradictions: Mapped[Optional[list]] = mapped_column(JSON, nullable=True)
+
+    # Raw LLM response (for debugging / re-parsing)
+    raw_response: Mapped[Optional[str]] = mapped_column(Text, nullable=True)
+
+    # Token usage
+    input_tokens: Mapped[Optional[int]] = mapped_column(Integer, nullable=True)
+    output_tokens: Mapped[Optional[int]] = mapped_column(Integer, nullable=True)
+
+    created_at: Mapped[datetime] = mapped_column(DateTime, default=datetime.utcnow)
+
+    # Relationships
+    source: Mapped["Source"] = relationship("Source", back_populates="extraction")
+
+    def __repr__(self) -> str:
+        return f"Extraction(id={self.id}, source_id={self.source_id})"
+
+
+class WikiPage(Base):
+    """Tracks vault pages written or updated by the wiki compiler."""
+
+    __tablename__ = "wiki_pages"
+
+    id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
+
+    # Relative path inside the vault
+    vault_path: Mapped[str] = mapped_column(String(500), unique=True, nullable=False)
+    domain: Mapped[str] = mapped_column(String(50), nullable=False)
+    title: Mapped[str] = mapped_column(String(300), nullable=False)
+
+    # Git commit shas before/after last compile
+    git_sha_before: Mapped[Optional[str]] = mapped_column(String(40), nullable=True)
+    git_sha_after: Mapped[Optional[str]] = mapped_column(String(40), nullable=True)
+
+    created_at: Mapped[datetime] = mapped_column(DateTime, default=datetime.utcnow)
+    updated_at: Mapped[Optional[datetime]] = mapped_column(DateTime, nullable=True)
+
+    def __repr__(self) -> str:
+        return f"WikiPage(id={self.id}, vault_path={self.vault_path!r})"
+
+
+__all__ = [
+    "Base",
+    "Source",
+    "SourceStatus",
+    "SourceType",
+    "Extraction",
+    "WikiPage",
+]

src/second_brain/scheduler/__init__.py

@@ -0,0 +1,5 @@
+"""Overnight scheduler — rate-limit smoother with token-aware spacing."""
+
+from second_brain.scheduler.runner import Scheduler
+
+__all__ = ["Scheduler"]

src/second_brain/scheduler/runner.py

@@ -0,0 +1,148 @@
+"""
+Overnight scheduler — rate-limit smoother with token-aware spacing.
+
+Spreads LLM extraction calls across a configurable overnight window,
+enforcing a minimum gap between calls and a max-tokens-per-hour budget.
+Idempotent: safe to restart mid-run.
+"""
+
+from __future__ import annotations
+
+import time
+from datetime import datetime, time as dtime
+from typing import Optional
+
+from second_brain.config import Config
+from second_brain.database import get_database
+from second_brain.models import Source, SourceStatus
+
+
+class Scheduler:
+    """
+    Runs the pipeline on pending sources within a time window.
+
+    Config keys (from settings.toml [scheduler] table):
+      window_start      — "HH:MM"  (default "22:00")
+      window_end        — "HH:MM"  (default "06:00")  may cross midnight
+      max_tokens_per_hour — int    (default 100_000)
+      min_gap_seconds   — int      (default 120)
+    """
+
+    def __init__(self, config: Config) -> None:
+        self.config = config
+        sched = config.scheduler
+        self.window_start: dtime = _parse_time(sched.get("window_start", "22:00"))
+        self.window_end: dtime = _parse_time(sched.get("window_end", "06:00"))
+        self.max_tokens_per_hour: int = int(sched.get("max_tokens_per_hour", 100_000))
+        self.min_gap_seconds: int = int(sched.get("min_gap_seconds", 120))
+
+    def run(self, dry_run: bool = False) -> None:
+        """
+        Process pending sources respecting the window and token budget.
+
+        Runs until the window closes or there are no more pending sources.
+        """
+        from second_brain.context.assembler import ContextAssembler
+        from second_brain.extractor.engine import ExtractionEngine
+        from second_brain.models import Extraction
+
+        db = get_database()
+        assembler = ContextAssembler(self.config)
+        engine = ExtractionEngine(api_key=self.config.anthropic_api_key)
+
+        tokens_this_hour: int = 0
+        hour_start: float = time.time()
+        processed: int = 0
+
+        print(
+            f"[Scheduler] Starting. Window {_fmt(self.window_start)}–{_fmt(self.window_end)}, "
+            f"budget {self.max_tokens_per_hour:,} tokens/hr, "
+            f"gap {self.min_gap_seconds}s"
+        )
+
+        while self._in_window():
+            # Reset hourly token bucket
+            elapsed = time.time() - hour_start
+            if elapsed >= 3600:
+                tokens_this_hour = 0
+                hour_start = time.time()
+
+            with db.session() as sess:
+                source = (
+                    sess.query(Source)
+                    .filter(Source.status == SourceStatus.ANALYZED)
+                    .first()
+                )
+                if source is None:
+                    print("[Scheduler] No pending sources. Done.")
+                    break
+
+                # Token budget check (rough: assume avg 2k tokens per extraction)
+                if tokens_this_hour + 2000 > self.max_tokens_per_hour:
+                    wait = int(3600 - elapsed) + 5
+                    print(f"[Scheduler] Token budget reached. Sleeping {wait}s…")
+                    if not dry_run:
+                        time.sleep(wait)
+                    tokens_this_hour = 0
+                    hour_start = time.time()
+                    continue
+
+                if dry_run:
+                    print(f"[Scheduler] DRY RUN — would process source {source.id}")
+                    break
+
+                # Run extraction
+                try:
+                    prompt = assembler.assemble(source)
+                    result = engine.extract(prompt, source)
+
+                    extraction = Extraction(
+                        source_id=source.id,
+                        raw_response=getattr(result, "_raw_response", None),
+                        input_tokens=getattr(result, "_input_tokens", None),
+                        output_tokens=getattr(result, "_output_tokens", None),
+                        **result.to_db_dict(),
+                    )
+                    sess.add(extraction)
+                    source.status = SourceStatus.ANALYZED
+                    source.updated_at = datetime.utcnow()
+
+                    used = (getattr(result, "_input_tokens", 0) or 0) + (
+                        getattr(result, "_output_tokens", 0) or 0
+                    )
+                    tokens_this_hour += used
+                    processed += 1
+
+                except Exception as exc:
+                    print(f"[Scheduler] Error on source {source.id}: {exc}")
+                    source.error_message = str(exc)
+                    source.status = SourceStatus.FAILED
+
+            # Enforce minimum gap between calls
+            time.sleep(self.min_gap_seconds)
+
+        print(f"[Scheduler] Finished. Processed {processed} source(s).")
+
+    def _in_window(self) -> bool:
+        """Return True if current time is within the scheduler window."""
+        now = datetime.now().time().replace(second=0, microsecond=0)
+        start = self.window_start
+        end = self.window_end
+
+        if start <= end:
+            return start <= now <= end
+        else:
+            # Window crosses midnight (e.g., 22:00 → 06:00)
+            return now >= start or now <= end
+
+
+def _parse_time(s: str) -> dtime:
+    h, m = s.split(":")
+    return dtime(int(h), int(m))
+
+
+def _fmt(t: dtime) -> str:
+    return t.strftime("%H:%M")
+
+
+__all__ = ["Scheduler"]

src/second_brain/web/__init__.py

@@ -0,0 +1 @@
+"""FastAPI web interface for second-brain."""

src/second_brain/web/app.py

@@ -0,0 +1,294 @@
+"""
+FastAPI web interface for second-brain.
+
+Ported and extended from xtract/webapp.py.
+New additions:
+  - Domain and status filter params on the source list
+  - Source detail shows the full ExtractionResult
+  - Accept / reject HTMX actions
+  - Queue view (pending + in-progress sources)
+"""
+
+from __future__ import annotations
+
+from pathlib import Path
+from typing import List, Optional
+
+from fastapi import FastAPI, Form, HTTPException, Request
+from fastapi.responses import HTMLResponse
+from fastapi.templating import Jinja2Templates
+from pydantic import BaseModel
+from sqlalchemy.orm import Session
+
+from second_brain.config import DOMAINS, load_config
+from second_brain.database import get_database
+from second_brain.models import Extraction, Source, SourceStatus, SourceType
+
+# ---------------------------------------------------------------------------
+# App setup
+# ---------------------------------------------------------------------------
+
+app = FastAPI(
+    title="second-brain",
+    description="Personal knowledge extraction pipeline",
+    version="0.1.0",
+)
+
+_templates_dir = Path(__file__).parent / "templates"
+templates = Jinja2Templates(directory=str(_templates_dir))
+
+
+# ---------------------------------------------------------------------------
+# Pydantic response models
+# ---------------------------------------------------------------------------
+
+
+class SourceResponse(BaseModel):
+    id: int
+    url: str
+    title: Optional[str]
+    domain: str
+    status: str
+    source_type: str
+    focus: Optional[str]
+    ingested_at: str
+    has_extraction: bool
+
+    class Config:
+        from_attributes = True
+
+
+class ExtractionResponse(BaseModel):
+    summary: Optional[str]
+    key_points: Optional[list]
+    entities: Optional[list]
+    claims: Optional[list]
+    open_questions: Optional[list]
+    contradictions: Optional[list]
+
+    class Config:
+        from_attributes = True
+
+
+# ---------------------------------------------------------------------------
+# HTML routes
+# ---------------------------------------------------------------------------
+
+
+@app.get("/", response_class=HTMLResponse)
+async def index(
+    request: Request,
+    domain: Optional[str] = None,
+    status: Optional[str] = None,
+):
+    """Main source list with optional domain/status filters."""
+    db = get_database()
+    with db.session() as sess:
+        q = sess.query(Source)
+        if domain and domain in DOMAINS:
+            q = q.filter(Source.domain == domain)
+        if status:
+            try:
+                q = q.filter(Source.status == SourceStatus[status.upper()])
+            except KeyError:
+                pass
+        sources = q.order_by(Source.ingested_at.desc()).all()
+        source_list = [_source_to_dict(s) for s in sources]
+
+    return templates.TemplateResponse(
+        "index.html",
+        {
+            "request": request,
+            "sources": source_list,
+            "domains": DOMAINS,
+            "statuses": [s.value for s in SourceStatus],
+            "selected_domain": domain or "",
+            "selected_status": status or "",
+        },
+    )
+
+
+@app.get("/sources/{source_id}", response_class=HTMLResponse)
+async def source_detail(request: Request, source_id: int):
+    """Source detail page with extraction results and accept/reject actions."""
+    db = get_database()
+    with db.session() as sess:
+        source = sess.query(Source).filter(Source.id == source_id).first()
+        if not source:
+            raise HTTPException(status_code=404, detail="Source not found")
+        source_data = _source_to_dict(source)
+        extraction_data = _extraction_to_dict(source.extraction) if source.extraction else None
+
+    return templates.TemplateResponse(
+        "source_detail.html",
+        {
+            "request": request,
+            "source": source_data,
+            "extraction": extraction_data,
+        },
+    )
+
+
+@app.get("/queue", response_class=HTMLResponse)
+async def queue_view(request: Request):
+    """Queue view: pending and in-flight sources."""
+    db = get_database()
+    in_flight_statuses = [SourceStatus.PENDING, SourceStatus.PULLED, SourceStatus.TRANSCRIBED]
+    with db.session() as sess:
+        sources = (
+            sess.query(Source)
+            .filter(Source.status.in_(in_flight_statuses))
+            .order_by(Source.ingested_at.asc())
+            .all()
+        )
+        source_list = [_source_to_dict(s) for s in sources]
+
+    return templates.TemplateResponse(
+        "index.html",
+        {
+            "request": request,
+            "sources": source_list,
+            "domains": DOMAINS,
+            "statuses": [s.value for s in SourceStatus],
+            "selected_domain": "",
+            "selected_status": "",
+            "page_title": "Queue",
+        },
+    )
+
+
+# ---------------------------------------------------------------------------
+# HTMX action endpoints
+# ---------------------------------------------------------------------------
+
+
+@app.post("/sources/{source_id}/accept")
+async def accept_source(request: Request, source_id: int):
+    """Mark a source as accepted (ready for wiki compilation)."""
+    db = get_database()
+    with db.session() as sess:
+        source = sess.query(Source).filter(Source.id == source_id).first()
+        if not source:
+            raise HTTPException(status_code=404, detail="Source not found")
+        source.status = SourceStatus.ACCEPTED
+
+    return templates.TemplateResponse(
+        "source_detail.html",
+        {
+            "request": request,
+            "source": _source_to_dict(source),
+            "extraction": _extraction_to_dict(source.extraction) if source.extraction else None,
+            "flash": "Accepted — will be included in next wiki compile.",
+        },
+    )
+
+
+@app.post("/sources/{source_id}/reject")
+async def reject_source(request: Request, source_id: int):
+    """Mark a source as failed/rejected (excluded from wiki)."""
+    db = get_database()
+    with db.session() as sess:
+        source = sess.query(Source).filter(Source.id == source_id).first()
+        if not source:
+            raise HTTPException(status_code=404, detail="Source not found")
+        source.status = SourceStatus.FAILED
+
+    return templates.TemplateResponse(
+        "source_detail.html",
+        {
+            "request": request,
+            "source": _source_to_dict(source),
+            "extraction": _extraction_to_dict(source.extraction) if source.extraction else None,
+            "flash": "Rejected.",
+        },
+    )
+
+
+# ---------------------------------------------------------------------------
+# JSON API endpoints
+# ---------------------------------------------------------------------------
+
+
+@app.get("/api/sources", response_model=List[SourceResponse])
+async def api_sources(
+    domain: Optional[str] = None,
+    status: Optional[str] = None,
+):
+    db = get_database()
+    with db.session() as sess:
+        q = sess.query(Source)
+        if domain:
+            q = q.filter(Source.domain == domain)
+        if status:
+            try:
+                q = q.filter(Source.status == SourceStatus[status.upper()])
+            except KeyError:
+                pass
+        sources = q.order_by(Source.ingested_at.desc()).all()
+        return [_source_to_response(s) for s in sources]
+
+
+@app.get("/api/sources/{source_id}/extraction", response_model=ExtractionResponse)
+async def api_extraction(source_id: int):
+    db = get_database()
+    with db.session() as sess:
+        extraction = sess.query(Extraction).filter_by(source_id=source_id).first()
+        if not extraction:
+            raise HTTPException(status_code=404, detail="No extraction for this source")
+        return ExtractionResponse(
+            summary=extraction.summary,
+            key_points=extraction.key_points,
+            entities=extraction.entities,
+            claims=extraction.claims,
+            open_questions=extraction.open_questions,
+            contradictions=extraction.contradictions,
+        )
+
+
+# ---------------------------------------------------------------------------
+# Helpers
+# ---------------------------------------------------------------------------
+
+
+def _source_to_dict(s: Source) -> dict:
+    return {
+        "id": s.id,
+        "url": s.url,
+        "title": s.title or s.url,
+        "domain": s.domain,
+        "status": s.status.value,
+        "source_type": s.source_type.value,
+        "focus": s.focus,
+        "ingested_at": s.ingested_at.strftime("%Y-%m-%d %H:%M") if s.ingested_at else "",
+        "has_extraction": s.extraction is not None,
+        "error_message": s.error_message,
+    }
+
+
+def _extraction_to_dict(e: Optional[Extraction]) -> Optional[dict]:
+    if e is None:
+        return None
+    return {
+        "summary": e.summary,
+        "key_points": e.key_points or [],
+        "entities": e.entities or [],
+        "claims": e.claims or [],
+        "open_questions": e.open_questions or [],
+        "contradictions": e.contradictions or [],
+        "input_tokens": e.input_tokens,
+        "output_tokens": e.output_tokens,
+    }
+
+
+def _source_to_response(s: Source) -> SourceResponse:
+    return SourceResponse(
+        id=s.id,
+        url=s.url,
+        title=s.title,
+        domain=s.domain,
+        status=s.status.value,
+        source_type=s.source_type.value,
+        focus=s.focus,
+        ingested_at=s.ingested_at.isoformat() if s.ingested_at else "",
+        has_extraction=s.extraction is not None,
+    )

src/second_brain/web/templates/base.html

@@ -0,0 +1,30 @@
+<!DOCTYPE html>
+<html lang="en">
+<head>
+    <meta charset="UTF-8">
+    <meta name="viewport" content="width=device-width, initial-scale=1.0">
+    <title>{% block title %}second-brain{% endblock %}</title>
+    <script src="https://cdn.tailwindcss.com"></script>
+    <script src="https://unpkg.com/htmx.org@1.9.10"></script>
+</head>
+<body class="bg-gray-100 min-h-screen">
+
+<header class="bg-gradient-to-r from-indigo-700 to-violet-700 text-white shadow-lg">
+    <div class="container mx-auto px-4 py-5 flex items-center justify-between">
+        <div>
+            <a href="/" class="text-2xl font-bold tracking-tight hover:text-indigo-100">second-brain</a>
+            <p class="text-indigo-200 text-sm mt-0.5">personal knowledge pipeline</p>
+        </div>
+        <nav class="flex gap-4 text-sm font-medium">
+            <a href="/" class="hover:text-indigo-200 transition-colors">Sources</a>
+            <a href="/queue" class="hover:text-indigo-200 transition-colors">Queue</a>
+        </nav>
+    </div>
+</header>
+
+<main class="container mx-auto px-4 py-8">
+    {% block content %}{% endblock %}
+</main>
+
+</body>
+</html>

src/second_brain/web/templates/index.html

@@ -0,0 +1,94 @@
+{% extends "base.html" %}
+
+{% block title %}{% if page_title %}{{ page_title }} — {% endif %}second-brain{% endblock %}
+
+{% block content %}
+<div class="mb-6 flex items-center justify-between">
+    <h2 class="text-2xl font-semibold text-gray-800">
+        {{ page_title or "Sources" }}
+        <span class="text-base font-normal text-gray-500 ml-2">({{ sources|length }})</span>
+    </h2>
+</div>
+
+<!-- Filters -->
+<form method="get" action="/" class="bg-white rounded-lg shadow-sm p-4 mb-6 flex flex-wrap gap-4 items-end">
+    <div>
+        <label class="block text-xs font-medium text-gray-600 mb-1">Domain</label>
+        <select name="domain" class="text-sm border border-gray-300 rounded px-3 py-1.5 focus:outline-none focus:ring-2 focus:ring-indigo-400">
+            <option value="">All domains</option>
+            {% for d in domains %}
+            <option value="{{ d }}" {% if selected_domain == d %}selected{% endif %}>{{ d }}</option>
+            {% endfor %}
+        </select>
+    </div>
+    <div>
+        <label class="block text-xs font-medium text-gray-600 mb-1">Status</label>
+        <select name="status" class="text-sm border border-gray-300 rounded px-3 py-1.5 focus:outline-none focus:ring-2 focus:ring-indigo-400">
+            <option value="">All statuses</option>
+            {% for s in statuses %}
+            <option value="{{ s }}" {% if selected_status == s %}selected{% endif %}>{{ s }}</option>
+            {% endfor %}
+        </select>
+    </div>
+    <button type="submit"
+            class="px-4 py-1.5 bg-indigo-600 text-white text-sm font-medium rounded hover:bg-indigo-700 transition-colors">
+        Filter
+    </button>
+    <a href="/" class="px-4 py-1.5 bg-gray-100 text-gray-700 text-sm font-medium rounded hover:bg-gray-200 transition-colors">
+        Clear
+    </a>
+</form>
+
+<!-- Source list -->
+{% if sources %}
+<div class="grid gap-3">
+    {% for source in sources %}
+    {% set status_colors = {
+        'pending':     'bg-gray-100 text-gray-700',
+        'pulled':      'bg-blue-100 text-blue-700',
+        'transcribed': 'bg-yellow-100 text-yellow-700',
+        'analyzed':    'bg-orange-100 text-orange-700',
+        'accepted':    'bg-green-100 text-green-700',
+        'published':   'bg-purple-100 text-purple-700',
+        'failed':      'bg-red-100 text-red-700'
+    } %}
+    {% set domain_colors = {
+        'development': 'bg-cyan-50 text-cyan-700 border-cyan-200',
+        'content':     'bg-pink-50 text-pink-700 border-pink-200',
+        'business':    'bg-amber-50 text-amber-700 border-amber-200',
+        'homelab':     'bg-teal-50 text-teal-700 border-teal-200'
+    } %}
+    <a href="/sources/{{ source.id }}"
+       class="block bg-white rounded-lg border border-gray-200 p-4 hover:border-indigo-400 hover:shadow-sm transition-all duration-150 group">
+        <div class="flex items-start justify-between gap-4">
+            <div class="flex-1 min-w-0">
+                <h3 class="font-semibold text-gray-900 group-hover:text-indigo-700 truncate">
+                    {{ source.title }}
+                </h3>
+                <p class="text-xs text-gray-400 mt-0.5 truncate">{{ source.url }}</p>
+                {% if source.focus %}
+                <p class="text-xs text-gray-500 mt-1 italic">Focus: {{ source.focus }}</p>
+                {% endif %}
+            </div>
+            <div class="flex flex-col items-end gap-1.5 flex-shrink-0">
+                <span class="text-xs px-2 py-0.5 rounded-full font-medium
+                    {{ status_colors.get(source.status, 'bg-gray-100 text-gray-700') }}">
+                    {{ source.status }}
+                </span>
+                <span class="text-xs px-2 py-0.5 rounded border font-medium
+                    {{ domain_colors.get(source.domain, 'bg-gray-50 text-gray-600 border-gray-200') }}">
+                    {{ source.domain }}
+                </span>
+                <span class="text-xs text-gray-400">{{ source.ingested_at }}</span>
+            </div>
+        </div>
+    </a>
+    {% endfor %}
+</div>
+{% else %}
+<div class="text-center py-20 bg-white rounded-lg border border-dashed border-gray-300">
+    <p class="text-gray-400 text-lg">No sources found.</p>
+    <p class="text-gray-400 text-sm mt-1">Run <code class="bg-gray-100 px-1 rounded">second-brain add &lt;url&gt;</code> to queue one.</p>
+</div>
+{% endif %}
+{% endblock %}

src/second_brain/web/templates/source_detail.html

@@ -0,0 +1,179 @@
+{% extends "base.html" %}
+
+{% block title %}{{ source.title }} — second-brain{% endblock %}
+
+{% block content %}
+<!-- Back + title -->
+<div class="mb-6">
+    <a href="/" class="inline-flex items-center text-sm text-indigo-600 hover:text-indigo-800 mb-3">
+        <svg class="h-4 w-4 mr-1" fill="none" viewBox="0 0 24 24" stroke="currentColor">
+            <path stroke-linecap="round" stroke-linejoin="round" stroke-width="2" d="M10 19l-7-7m0 0l7-7m-7 7h18"/>
+        </svg>
+        Back to sources
+    </a>
+    <h2 class="text-2xl font-semibold text-gray-900">{{ source.title }}</h2>
+    <a href="{{ source.url }}" target="_blank"
+       class="text-sm text-indigo-500 hover:text-indigo-700 break-all">{{ source.url }}</a>
+</div>
+
+{% if flash %}
+<div class="mb-4 px-4 py-3 bg-green-50 border border-green-200 text-green-800 rounded text-sm">
+    {{ flash }}
+</div>
+{% endif %}
+
+<!-- Meta row -->
+<div class="flex flex-wrap gap-2 mb-6">
+    {% set status_colors = {
+        'pending':     'bg-gray-100 text-gray-700',
+        'pulled':      'bg-blue-100 text-blue-700',
+        'transcribed': 'bg-yellow-100 text-yellow-700',
+        'analyzed':    'bg-orange-100 text-orange-700',
+        'accepted':    'bg-green-100 text-green-700',
+        'published':   'bg-purple-100 text-purple-700',
+        'failed':      'bg-red-100 text-red-700'
+    } %}
+    <span class="px-3 py-1 rounded-full text-sm font-medium
+        {{ status_colors.get(source.status, 'bg-gray-100 text-gray-700') }}">
+        {{ source.status }}
+    </span>
+    <span class="px-3 py-1 rounded-full text-sm font-medium bg-indigo-100 text-indigo-700">
+        {{ source.domain }}
+    </span>
+    <span class="px-3 py-1 rounded-full text-sm font-medium bg-gray-100 text-gray-700">
+        {{ source.source_type }}
+    </span>
+    <span class="px-3 py-1 text-sm text-gray-500">
+        added {{ source.ingested_at }}
+    </span>
+</div>
+
+{% if source.focus %}
+<div class="mb-6 bg-indigo-50 border border-indigo-100 rounded p-3 text-sm text-indigo-800">
+    <span class="font-medium">Focus:</span> {{ source.focus }}
+</div>
+{% endif %}
+
+{% if source.error_message %}
+<div class="mb-6 bg-red-50 border border-red-200 rounded p-3 text-sm text-red-800">
+    <span class="font-medium">Error:</span> {{ source.error_message }}
+</div>
+{% endif %}
+
+<!-- Accept / Reject actions (only when analyzed) -->
+{% if source.status == 'analyzed' %}
+<div class="flex gap-3 mb-8">
+    <button hx-post="/sources/{{ source.id }}/accept"
+            hx-target="body"
+            hx-swap="innerHTML"
+            class="px-5 py-2 bg-green-600 text-white text-sm font-semibold rounded hover:bg-green-700 transition-colors">
+        Accept
+    </button>
+    <button hx-post="/sources/{{ source.id }}/reject"
+            hx-target="body"
+            hx-swap="innerHTML"
+            class="px-5 py-2 bg-red-500 text-white text-sm font-semibold rounded hover:bg-red-600 transition-colors">
+        Reject
+    </button>
+</div>
+{% endif %}
+
+<!-- Extraction results -->
+{% if extraction %}
+<div class="space-y-5">
+
+    <!-- Summary -->
+    {% if extraction.summary %}
+    <div class="bg-white rounded-lg border border-gray-200 p-5">
+        <h3 class="text-sm font-semibold text-gray-500 uppercase tracking-wide mb-2">Summary</h3>
+        <p class="text-gray-800 leading-relaxed">{{ extraction.summary }}</p>
+    </div>
+    {% endif %}
+
+    <!-- Key points -->
+    {% if extraction.key_points %}
+    <div class="bg-white rounded-lg border border-gray-200 p-5">
+        <h3 class="text-sm font-semibold text-gray-500 uppercase tracking-wide mb-3">Key Points</h3>
+        <ul class="space-y-2">
+            {% for pt in extraction.key_points %}
+            <li class="flex gap-2 text-gray-800 text-sm">
+                <span class="text-indigo-400 mt-0.5">•</span>
+                <span>{{ pt }}</span>
+            </li>
+            {% endfor %}
+        </ul>
+    </div>
+    {% endif %}
+
+    <!-- Claims -->
+    {% if extraction.claims %}
+    <div class="bg-white rounded-lg border border-gray-200 p-5">
+        <h3 class="text-sm font-semibold text-gray-500 uppercase tracking-wide mb-3">Claims</h3>
+        <ul class="space-y-2">
+            {% for c in extraction.claims %}
+            <li class="flex gap-2 text-gray-800 text-sm">
+                <span class="text-amber-400 mt-0.5">→</span>
+                <span>{{ c }}</span>
+            </li>
+            {% endfor %}
+        </ul>
+    </div>
+    {% endif %}
+
+    <!-- Entities -->
+    {% if extraction.entities %}
+    <div class="bg-white rounded-lg border border-gray-200 p-5">
+        <h3 class="text-sm font-semibold text-gray-500 uppercase tracking-wide mb-3">Entities</h3>
+        <div class="flex flex-wrap gap-2">
+            {% for e in extraction.entities %}
+            <span class="px-2.5 py-1 bg-gray-100 text-gray-700 rounded text-sm">{{ e }}</span>
+            {% endfor %}
+        </div>
+    </div>
+    {% endif %}
+
+    <!-- Open questions -->
+    {% if extraction.open_questions %}
+    <div class="bg-white rounded-lg border border-gray-200 p-5">
+        <h3 class="text-sm font-semibold text-gray-500 uppercase tracking-wide mb-3">Open Questions</h3>
+        <ul class="space-y-2">
+            {% for q in extraction.open_questions %}
+            <li class="flex gap-2 text-gray-800 text-sm">
+                <span class="text-teal-400 mt-0.5">?</span>
+                <span>{{ q }}</span>
+            </li>
+            {% endfor %}
+        </ul>
+    </div>
+    {% endif %}
+
+    <!-- Contradictions -->
+    {% if extraction.contradictions %}
+    <div class="bg-white rounded-lg border border-gray-200 p-5">
+        <h3 class="text-sm font-semibold text-gray-500 uppercase tracking-wide mb-3">Contradictions</h3>
+        <ul class="space-y-2">
+            {% for ct in extraction.contradictions %}
+            <li class="flex gap-2 text-gray-800 text-sm">
+                <span class="text-red-400 mt-0.5">✗</span>
+                <span>{{ ct }}</span>
+            </li>
+            {% endfor %}
+        </ul>
+    </div>
+    {% endif %}
+
+    <!-- Token usage footer -->
+    {% if extraction.input_tokens or extraction.output_tokens %}
+    <p class="text-xs text-gray-400 text-right">
+        tokens: {{ extraction.input_tokens or 0 }} in / {{ extraction.output_tokens or 0 }} out
+    </p>
+    {% endif %}
+
+</div>
+{% else %}
+<div class="text-center py-16 bg-white rounded-lg border border-dashed border-gray-300">
+    <p class="text-gray-400">No extraction yet.</p>
+    <p class="text-gray-400 text-sm mt-1">Run <code class="bg-gray-100 px-1 rounded">second-brain process</code> to analyze this source.</p>
+</div>
+{% endif %}
+{% endblock %}