.gitignore

@@ -1,45 +0,0 @@
-# Python bytecode + caches
-__pycache__/
-*.py[cod]
-*$py.class
-*.so
-
-# Build / packaging
-build/
-dist/
-*.egg
-*.egg-info/
-.eggs/
-
-# Virtual environments
-.venv/
-venv/
-env/
-ENV/
-
-# Tooling caches
-.pytest_cache/
-.mypy_cache/
-.ruff_cache/
-.tox/
-.coverage
-.coverage.*
-htmlcov/
-
-# Secrets / local config
-.env
-.env.local
-config/settings.local.toml
-
-# Editor / OS noise
-.idea/
-.vscode/
-*.swp
-*.swo
-.DS_Store
-Thumbs.db
-
-# Runtime artifacts (only if someone points settings.toml at a repo-local path)
-*.db
-*.sqlite
-*.sqlite3

config/settings.toml

@@ -21,24 +21,13 @@ whisper_model = "small"
 # Active knowledge domains
 domains = ["development", "content", "business", "homelab"]
 
-[extractor]
-# "cli" → runs `claude -p` under your Max OAuth (no per-token billing).
-# "api" → uses Anthropic SDK, requires ANTHROPIC_API_KEY.
-backend = "cli"
-model = "claude-sonnet-4-6"
-cli_binary = "claude"
-timeout_sec = 600
-
 [scheduler]
 # Time window for overnight processing (24h format, may cross midnight)
 window_start = "22:00"
 window_end   = "06:00"
 
-# CLI backend: cap calls per rolling hour (Max sub has 5h-window message caps).
+# Max Claude API tokens to consume per hour
-max_calls_per_hour = 30
+max_tokens_per_hour = 100_000
 
 # Minimum seconds between extraction calls (rate-limit smoother)
 min_gap_seconds = 120
-
-# Legacy API-backend budget (only enforced when extractor.backend = "api").
-max_tokens_per_hour = 100_000

pyproject.toml

@@ -4,6 +4,7 @@ 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",
@@ -17,12 +18,6 @@ dependencies = [
     "yt-dlp>=2024.1.0",
 ]
 
-[project.optional-dependencies]
-# Only required when extractor.backend = "api" in settings.toml.
-# The default "cli" backend shells out to the Claude Code CLI under your
-# Max OAuth subscription and has no Python-side Anthropic dependency.
-api = ["anthropic>=0.40.0"]
-
 [project.scripts]
 second-brain = "second_brain.main:cli"
 

reviews/a-review-2026-05-24.md

@@ -1,42 +0,0 @@
-# a-review — 2026-05-24
-
-**Model:** gemini
-**Diff base:** `HEAD` (9 files changed, 364 insertions(+), 152 deletions(-))
-**Session reference:** none
-**Context:** CLAUDE.md + session notes + expanded diff
-
----
-
-Here is a review of the code changes.
-
-### [WARNING] Potential for brittle path resolution
-- **File:** `src/second_brain/config.py`
-- **Location:** Line 94, `prompts_dir` property
-- **Issue:** The `prompts_dir` property calculates the project's root directory using `Path(__file__).resolve().parent.parent.parent`. This kind of relative path traversal (`../../..`) is brittle and can easily break if `config.py` or the directory structure is refactored.
-- **Suggestion:** For more complex projects, libraries like `importlib.resources` (for Python 3.9+) are a more robust way to locate package data. For the current project size, this is a minor issue, but it's worth being aware of the fragility.
-
-### [WARNING] Potential race condition in `process` command
-- **File:** `src/second_brain/main.py`
-- **Location:** Line 150 (`process` command)
-- **Issue:** The command first queries for a list of all eligible `source_ids`, and then iterates through this list, processing each one. If two instances of `second-brain process` were run concurrently, they would both get the same list of IDs and attempt to process the same sources, leading to duplicated work and potentially errors or race conditions upon writing results to the database.
-- **Suggestion:** This is a low-risk issue for a CLI tool that is likely run as a singleton. However, for a more robust system, processing should claim an item atomically. This could be done by querying for a single processable item within the loop, or by using a database-level lock (`SELECT ... FOR UPDATE`), to ensure a given source is only picked up by one process.
-
-### [INFO] Bug fix in scheduler logic
-- **File:** `src/second_brain/scheduler/runner.py`
-- **Location:** Line 122
-- **Issue:** The comments and code indicate a critical bug was fixed. The scheduler was previously attempting to process sources that were already in the `ANALYZED` state, causing it to do no useful work.
-- **Fix:** The query has been corrected to target sources in the `TRANSCRIBED` state (`.filter(Source.status == SourceStatus.TRANSCRIBED)`), which aligns with the documented pipeline flow.
-
-### [INFO] Robustness improvements in wiki compiler
-- **File:** `src/second_brain/compiler/wiki.py`
-- **Location:** `_ensure_git_repo` and `_git_commit` methods
-- **Issue:** The git integration is now significantly more robust.
-- **Fix:** The compiler now automatically initializes a git repository in the vault (`_ensure_git_repo`) if one doesn't exist, preventing errors on a fresh setup. Furthermore, the `_git_commit` logic now checks for staged changes before attempting a commit, preventing the creation of empty commits. These are high-quality, thoughtful improvements.
-
-### Summary of Changes
-
-Overall, these are high-quality changes that introduce a significant and well-implemented architectural improvement (the CLI-based extractor backend) while also fixing several bugs and improving system robustness.
-
-The new CLI backend is a clever approach that leverages the `claude` tool's schema enforcement to improve the reliability of JSON extraction, while also making the `anthropic` SDK an optional dependency.
-
-The bug fix in the scheduler is critical for the system's core function. The improvements to the wiki compiler's git handling show good attention to detail and user experience. The minor warnings are about potential future issues rather than immediate bugs and do not detract from the overall quality of the submission. The code aligns well with the project's stated architecture and demonstrates a clear, positive evolution of the system.

src/second_brain/compiler/wiki.py

@@ -19,9 +19,11 @@ 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, utcnow
+from second_brain.models import Extraction, Source, SourceStatus, WikiPage
 
 
 class WikiCompiler:
@@ -46,7 +48,6 @@ class WikiCompiler:
         """
         vault = self.config.vault_path
         vault.mkdir(parents=True, exist_ok=True)
-        self._ensure_git_repo(vault)
 
         db = get_database()
         stats = {"processed": 0, "created": 0, "updated": 0, "errors": 0}
@@ -72,7 +73,7 @@ class WikiCompiler:
                     else:
                         stats["updated"] += 1
                     source.status = SourceStatus.PUBLISHED
-                    source.updated_at = utcnow()
+                    source.updated_at = datetime.utcnow()
                     stats["processed"] += 1
                 except Exception as exc:
                     print(f"[Compiler] Error processing source {source.id}: {exc}")
@@ -124,7 +125,7 @@ class WikiCompiler:
             )
             sess.add(wiki_row)
         else:
-            wiki_row.updated_at = utcnow()
+            wiki_row.updated_at = datetime.utcnow()
 
         return created
 
@@ -192,30 +193,6 @@ class WikiCompiler:
     # Git helpers (stubbed — interface is clear for later wiring)
     # ------------------------------------------------------------------
 
-    @staticmethod
-    def _ensure_git_repo(vault: Path) -> None:
-        """Initialise a git repo in the vault if one doesn't already exist.
-
-        Idempotent. Creates the repo with an initial empty commit so that
-        subsequent `git rev-parse HEAD` calls succeed even before the first
-        compile writes a page.
-        """
-        if (vault / ".git").exists():
-            return
-        try:
-            subprocess.run(
-                ["git", "init", "--initial-branch=main", "--quiet"],
-                cwd=vault, check=True,
-            )
-            # Allow the empty initial commit so HEAD exists immediately.
-            subprocess.run(
-                ["git", "commit", "--allow-empty", "-m", "second-brain: vault init", "--quiet"],
-                cwd=vault, check=True,
-            )
-            print(f"[Compiler] Initialised vault git repo at {vault}")
-        except (subprocess.CalledProcessError, FileNotFoundError) as exc:
-            print(f"[Compiler] Warning: could not init git repo in {vault}: {exc}")
-
     @staticmethod
     def _git_sha(vault: Path) -> Optional[str]:
         """Return HEAD sha of the vault git repo, or None if not a git repo."""
@@ -233,22 +210,11 @@ class WikiCompiler:
 
     @staticmethod
     def _git_commit(vault: Path, count: int) -> Optional[str]:
-        """Stage all changes and commit.  Returns new HEAD sha or None.
+        """Stage all changes and commit.  Returns new HEAD sha or None."""
-
-        Returns None (without raising) if there's nothing to commit — that's
-        the normal case when a compile run produced no actual file changes.
-        """
         try:
             subprocess.run(["git", "add", "-A"], cwd=vault, check=True)
-            # Bail out cleanly if there's nothing staged.
+            msg = f"second-brain: compile {count} source(s) [{datetime.utcnow().date()}]"
-            diff = subprocess.run(
+            subprocess.run(["git", "commit", "-m", msg], cwd=vault, check=True)
-                ["git", "diff", "--cached", "--quiet"],
-                cwd=vault, check=False,
-            )
-            if diff.returncode == 0:
-                return None  # nothing to commit
-            msg = f"second-brain: compile {count} source(s) [{utcnow().date()}]"
-            subprocess.run(["git", "commit", "-m", msg, "--quiet"], cwd=vault, check=True)
             result = subprocess.run(
                 ["git", "rev-parse", "HEAD"],
                 cwd=vault,

src/second_brain/config.py

@@ -27,22 +27,11 @@ _DEFAULTS: dict[str, Any] = {
     "subtitles_dir": "~/.local/share/second-brain/subtitles",
     "whisper_model": "small",
     "domains": ["development", "content", "business", "homelab"],
-    "extractor": {
-        # "cli" runs `claude -p` under the Max OAuth subscription (default).
-        # "api" uses the Anthropic SDK and requires ANTHROPIC_API_KEY.
-        "backend": "cli",
-        "model": "claude-sonnet-4-6",
-        "cli_binary": "claude",
-        "timeout_sec": 600,
-    },
     "scheduler": {
         "window_start": "22:00",
         "window_end": "06:00",
-        # CLI backend: throttle by request rate (no per-token billing).
-        "max_calls_per_hour": 30,
-        "min_gap_seconds": 120,
-        # Legacy API-backend budget; kept for backwards compat.
         "max_tokens_per_hour": 100_000,
+        "min_gap_seconds": 120,
     },
 }
 
@@ -89,12 +78,6 @@ class Config:
     def anthropic_api_key(self) -> str | None:
         return os.getenv("ANTHROPIC_API_KEY")
 
-    # --- extractor ---
-
-    @property
-    def extractor(self) -> dict[str, Any]:
-        return {**_DEFAULTS["extractor"], **self._raw.get("extractor", {})}
-
     # --- scheduler ---
 
     @property
@@ -106,8 +89,7 @@ class Config:
     @property
     def prompts_dir(self) -> Path:
         """Directory containing per-domain prompt templates."""
-        # src/second_brain/config.py → ../../.. = project root
+        here = Path(__file__).parent.parent.parent.parent  # project root
-        here = Path(__file__).resolve().parent.parent.parent
         return here / "prompts"
 
     def ensure_dirs(self) -> None:

src/second_brain/extractor/engine.py

@@ -1,14 +1,9 @@
 """
 Single-shot LLM extraction engine.
 
-Two backends:
+Ported/refactored from xtract/analyzer.py.
-- "cli" (default): subprocess call to `claude -p`, runs under Max OAuth.
+Pure function: context bundle in → ExtractionResult out.
-                   Uses --json-schema for guaranteed valid JSON.
+Uses the Anthropic SDK with a single messages.create() call.
-- "api":           Anthropic SDK call. Requires ANTHROPIC_API_KEY.
-
-The CLI backend is preferred — costs nothing against the subscription and
-gets us schema-enforced output for free. The API backend is kept as a
-fallback (e.g. for CI, headless servers without an OAuth session, etc.).
 """
 
 from __future__ import annotations
@@ -17,141 +12,86 @@ import json
 import os
 from typing import Optional
 
-from second_brain.extractor.schema import ExtractionResult, LLMExtraction, SourceMeta
+import anthropic
-from second_brain.llm.claude_cli import ClaudeCLI, ClaudeCLIError, DEFAULT_MODEL
+
+from second_brain.extractor.schema import ExtractionResult, SourceMeta
 from second_brain.models import Source
 
 
-SYSTEM_PROMPT = (
-    "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 matching the provided schema. "
-    "Return ONLY the JSON object — no markdown fences, no preamble."
-)
-
-
 class ExtractionEngine:
     """
-    Stateless LLM extractor. Default backend: Claude CLI under OAuth.
+    Stateless LLM extractor.
 
-    Usage:
+    Call extract() with a pre-assembled context bundle (prompt + transcript).
-        engine = ExtractionEngine()                    # cli backend
+    Returns a validated ExtractionResult or raises on failure.
-        engine = ExtractionEngine(backend="api", ...)  # api backend
-        result = engine.extract(prompt, source)
     """
 
-    def __init__(
+    MODEL = "claude-sonnet-4-6"
-        self,
+    MAX_TOKENS = 4096
-        backend: str = "cli",
-        model: str = DEFAULT_MODEL,
-        api_key: Optional[str] = None,
-        cli_binary: str = "claude",
-        timeout_sec: int = 600,
-    ) -> None:
-        self.backend = backend
-        self.model = model
-
-        if backend == "cli":
-            self._cli = ClaudeCLI(
-                binary=cli_binary, model=model, timeout_sec=timeout_sec
-            )
-            self._client = None
-        elif backend == "api":
-            import anthropic  # lazy import — only required for the api path
 
+    def __init__(self, api_key: Optional[str] = None) -> None:
         key = api_key or os.getenv("ANTHROPIC_API_KEY")
         if not key:
             raise ValueError(
-                    "backend='api' requires ANTHROPIC_API_KEY (env var or constructor arg)."
+                "ANTHROPIC_API_KEY is required. Set the environment variable or "
+                "pass api_key to ExtractionEngine."
             )
-            self._client = anthropic.Anthropic(api_key=key)
+        self.client = anthropic.Anthropic(api_key=key)
-            self._cli = None
-        else:
-            raise ValueError(f"Unknown backend: {backend!r} (expected 'cli' or 'api')")
-
-    # ------------------------------------------------------------------
-    # Public API
-    # ------------------------------------------------------------------
 
     def extract(self, prompt: str, source: Source) -> ExtractionResult:
-        """Run a single-shot extraction call. Returns a validated ExtractionResult."""
+        """
-        print(
+        Run a single-shot extraction call.
-            f"[Extractor] [{self.backend}] source={source.id} domain={source.domain} "
-            f"model={self.model}"
-        )
 
-        if self.backend == "cli":
+        Args:
-            text, structured, input_tokens, output_tokens, raw_response = (
+            prompt: The fully assembled context bundle (system + transcript).
-                self._extract_cli(prompt)
+            source: The Source row (used to populate SourceMeta).
-            )
-        else:
-            text, input_tokens, output_tokens, raw_response = self._extract_api(prompt)
-            structured = None
 
-        print(
+        Returns:
-            f"[Extractor] Done — input={input_tokens} output={output_tokens} tokens"
+            Validated ExtractionResult.
-        )
+        """
+        print(f"[Extractor] Calling Claude for source {source.id} ({source.domain})…")
 
-        # Prefer the schema-validated structured payload when available.
+        response = self.client.messages.create(
-        if structured is not None:
+            model=self.MODEL,
-            llm = LLMExtraction.model_validate(structured)
+            max_tokens=self.MAX_TOKENS,
-        else:
+            system=self._system_prompt(),
-            llm = self._parse(text)
-        result = self._compose(llm, source)
-
-        # Attach for the caller to persist
-        result._raw_response = raw_response
-        result._input_tokens = input_tokens
-        result._output_tokens = output_tokens
-
-        return result
-
-    # ------------------------------------------------------------------
-    # Backend: CLI
-    # ------------------------------------------------------------------
-
-    def _extract_cli(
-        self, prompt: str
-    ) -> tuple[str, Optional[dict], Optional[int], Optional[int], str]:
-        schema = LLMExtraction.model_json_schema()
-        res = self._cli.complete(
-            prompt,
-            system_prompt=SYSTEM_PROMPT,
-            json_schema=schema,
-        )
-        return (
-            res.text,
-            res.structured,
-            res.input_tokens,
-            res.output_tokens,
-            json.dumps(res.raw),
-        )
-
-    # ------------------------------------------------------------------
-    # Backend: API
-    # ------------------------------------------------------------------
-
-    def _extract_api(self, prompt: str) -> tuple[str, Optional[int], Optional[int], str]:
-        response = self._client.messages.create(
-            model=self.model,
-            max_tokens=4096,
-            system=SYSTEM_PROMPT,
             messages=[{"role": "user", "content": prompt}],
         )
-        raw = response.content[0].text
+
+        raw_text = response.content[0].text
         usage = response.usage
-        return raw, usage.input_tokens, usage.output_tokens, raw
+
+        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
 
     # ------------------------------------------------------------------
-    # Parse + compose
+    # Internal helpers
     # ------------------------------------------------------------------
 
     @staticmethod
-    def _parse(raw: str) -> LLMExtraction:
+    def _system_prompt() -> str:
-        """Parse the model's JSON output into an LLMExtraction."""
+        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()
-        # CLI with --json-schema should always be clean, but the API path
+        # Strip accidental markdown fences
-        # occasionally wraps in ```json fences. Strip defensively.
         if text.startswith("```"):
             text = text.split("\n", 1)[-1]
             text = text.rsplit("```", 1)[0].strip()
@@ -161,13 +101,9 @@ class ExtractionEngine:
         except json.JSONDecodeError as exc:
             raise ValueError(
                 f"Could not parse JSON from LLM response: {exc}\n"
-                f"Raw (first 500 chars):\n{raw[:500]}"
+                f"Raw response (first 500 chars):\n{raw[:500]}"
             ) from exc
 
-        return LLMExtraction.model_validate(data)
-
-    @staticmethod
-    def _compose(llm: LLMExtraction, source: Source) -> ExtractionResult:
         meta = SourceMeta(
             url=source.url,
             title=source.title,
@@ -176,16 +112,17 @@ class ExtractionEngine:
             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=llm.summary,
+            summary=data.get("summary", ""),
-            key_points=llm.key_points,
+            key_points=data.get("key_points", []),
-            entities=llm.entities,
+            entities=data.get("entities", []),
-            claims=llm.claims,
+            claims=data.get("claims", []),
-            open_questions=llm.open_questions,
+            open_questions=data.get("open_questions", []),
-            contradictions=llm.contradictions,
+            contradictions=data.get("contradictions", []),
         )
 
 
@@ -200,4 +137,4 @@ def _fmt_duration(seconds: Optional[int]) -> Optional[str]:
     return f"{m}:{s:02d}"
 
 
-__all__ = ["ExtractionEngine", "SYSTEM_PROMPT"]
+__all__ = ["ExtractionEngine"]

src/second_brain/extractor/schema.py

@@ -13,38 +13,6 @@ from typing import Optional
 from pydantic import BaseModel, Field, HttpUrl
 
 
-class LLMExtraction(BaseModel):
-    """
-    The subset of fields that the LLM is responsible for producing.
-
-    Used as the contract for `claude -p --json-schema` so the CLI enforces
-    valid structured output. Source metadata (url, title, etc.) is filled
-    in by the engine from the Source row, not by the model.
-    """
-
-    summary: str = Field(description="2-4 sentence overview of the source")
-    key_points: list[str] = Field(
-        default_factory=list,
-        description="5-10 bulleted takeaways",
-    )
-    entities: list[str] = Field(
-        default_factory=list,
-        description="People, tools, products, companies, concepts mentioned",
-    )
-    claims: list[str] = Field(
-        default_factory=list,
-        description="Specific factual assertions made by the source",
-    )
-    open_questions: list[str] = Field(
-        default_factory=list,
-        description="Questions raised but not answered",
-    )
-    contradictions: list[str] = Field(
-        default_factory=list,
-        description="Points that contradict known facts or other sources",
-    )
-
-
 class SourceMeta(BaseModel):
     """Metadata about the source (populated from the Source row)."""
 
@@ -111,4 +79,4 @@ class ExtractionResult(BaseModel):
         }
 
 
-__all__ = ["ExtractionResult", "LLMExtraction", "SourceMeta"]
+__all__ = ["ExtractionResult", "SourceMeta"]

src/second_brain/llm/__init__.py

@@ -1,5 +0,0 @@
-"""LLM backend wrappers (Claude CLI, Anthropic API)."""
-
-from second_brain.llm.claude_cli import ClaudeCLI, ClaudeCLIError
-
-__all__ = ["ClaudeCLI", "ClaudeCLIError"]

src/second_brain/llm/claude_cli.py

@@ -1,185 +0,0 @@
-"""
-Subprocess wrapper around the Claude Code CLI in headless (`-p`) mode.
-
-Runs under the user's OAuth subscription (Max), so it does NOT consume
-ANTHROPIC_API_KEY tokens. The same pattern Travis uses for the Gemini CLI:
-spawn in a throwaway working directory so CLAUDE.md / project settings /
-hooks don't leak into the extraction prompt.
-
-Returns a small `ClaudeResult` dataclass: `(text, usage, cost_usd, raw)`.
-Callers parse `text` themselves (or pre-validate via `--json-schema`).
-"""
-
-from __future__ import annotations
-
-import json
-import os
-import subprocess
-import tempfile
-from dataclasses import dataclass, field
-from typing import Any, Optional
-
-
-DEFAULT_MODEL = "claude-sonnet-4-6"
-DEFAULT_TIMEOUT_SEC = 600
-
-
-class ClaudeCLIError(RuntimeError):
-    """Raised when the `claude` subprocess fails or returns unparseable output."""
-
-
-@dataclass
-class ClaudeResult:
-    text: str
-    structured: Optional[Any] = None  # populated when --json-schema is used
-    usage: dict[str, Any] = field(default_factory=dict)
-    cost_usd: Optional[float] = None
-    raw: dict[str, Any] = field(default_factory=dict)
-
-    @property
-    def input_tokens(self) -> Optional[int]:
-        return self.usage.get("input_tokens")
-
-    @property
-    def output_tokens(self) -> Optional[int]:
-        return self.usage.get("output_tokens")
-
-
-class ClaudeCLI:
-    """
-    Thin wrapper over `claude -p --output-format json ...`.
-
-    Key design choices:
-    - Prompt passed via stdin (argv has ~128KB limit; transcripts can be huge).
-    - `cwd` defaults to a fresh tempdir so the host project's CLAUDE.md,
-      `.claude/settings.json`, hooks, and `.mcp.json` don't get picked up.
-    - `--setting-sources ""` defensively skips user/project/local settings.
-    - `--tools ""` disables tool use (pure completion).
-    - `--no-session-persistence` so we don't accrete session files for batch runs.
-    - `--bare` is deliberately NOT used — it forces ANTHROPIC_API_KEY auth and
-      ignores OAuth, defeating the whole point of running under Max.
-    """
-
-    def __init__(
-        self,
-        binary: str = "claude",
-        model: str = DEFAULT_MODEL,
-        timeout_sec: int = DEFAULT_TIMEOUT_SEC,
-    ) -> None:
-        self.binary = binary
-        self.model = model
-        self.timeout_sec = timeout_sec
-
-    def complete(
-        self,
-        prompt: str,
-        *,
-        system_prompt: Optional[str] = None,
-        json_schema: Optional[dict[str, Any]] = None,
-        model: Optional[str] = None,
-        extra_args: Optional[list[str]] = None,
-    ) -> ClaudeResult:
-        """
-        Run one headless completion. Returns ClaudeResult.
-
-        Args:
-            prompt: User-turn content. Sent via stdin.
-            system_prompt: Replaces the default system prompt entirely
-                           (uses --system-prompt, not --append-system-prompt).
-            json_schema: If set, passes --json-schema so the CLI enforces
-                        valid structured output.
-            model: Override the default model for this call.
-            extra_args: Additional CLI flags to append (escape hatch).
-        """
-        argv: list[str] = [
-            self.binary,
-            "-p",
-            "--output-format", "json",
-            "--model", model or self.model,
-            "--no-session-persistence",
-            "--disable-slash-commands",
-            "--tools", "",
-            "--setting-sources", "",
-        ]
-        if system_prompt is not None:
-            argv += ["--system-prompt", system_prompt]
-        if json_schema is not None:
-            argv += ["--json-schema", json.dumps(json_schema)]
-        if extra_args:
-            argv += extra_args
-
-        # Hermetic working dir: avoids loading the host project's CLAUDE.md,
-        # hooks, MCP config, etc. Auto-cleaned on exit.
-        with tempfile.TemporaryDirectory(prefix="sb-claude-cli-") as workdir:
-            try:
-                proc = subprocess.run(
-                    argv,
-                    input=prompt,
-                    cwd=workdir,
-                    capture_output=True,
-                    text=True,
-                    timeout=self.timeout_sec,
-                    env=self._clean_env(),
-                )
-            except subprocess.TimeoutExpired as exc:
-                raise ClaudeCLIError(
-                    f"claude CLI timed out after {self.timeout_sec}s"
-                ) from exc
-            except FileNotFoundError as exc:
-                raise ClaudeCLIError(
-                    f"`{self.binary}` not found on PATH. "
-                    "Install Claude Code or set extractor.cli_binary in settings.toml."
-                ) from exc
-
-        if proc.returncode != 0:
-            raise ClaudeCLIError(
-                f"claude CLI exited {proc.returncode}.\n"
-                f"stderr: {proc.stderr.strip()[:2000]}\n"
-                f"stdout: {proc.stdout.strip()[:500]}"
-            )
-
-        try:
-            payload = json.loads(proc.stdout)
-        except json.JSONDecodeError as exc:
-            raise ClaudeCLIError(
-                f"Could not parse JSON envelope from claude CLI: {exc}\n"
-                f"stdout (first 500): {proc.stdout[:500]}"
-            ) from exc
-
-        # `--output-format json` envelope fields (Claude Code 2.x):
-        #   result            - text output (empty when --json-schema is used)
-        #   structured_output - parsed object, set when --json-schema is used
-        #   usage             - {input_tokens, output_tokens, cache_*_tokens, ...}
-        #   total_cost_usd    - float (informational on Max sub)
-        # Older versions used `response` instead of `result`; tolerate both.
-        text = payload.get("result") or payload.get("response") or ""
-        structured = payload.get("structured_output")
-        # If the schema produced a structured object but `result` is empty,
-        # surface the structured payload as JSON text so legacy parsers still work.
-        if structured is not None and not text:
-            text = json.dumps(structured)
-
-        return ClaudeResult(
-            text=text,
-            structured=structured,
-            usage=payload.get("usage", {}) or {},
-            cost_usd=payload.get("total_cost_usd"),
-            raw=payload,
-        )
-
-    @staticmethod
-    def _clean_env() -> dict[str, str]:
-        """
-        Build an env that won't pollute the hermetic call.
-
-        Strips ANTHROPIC_API_KEY so the CLI falls through to OAuth/keychain
-        (the whole point of using the subscription). Preserves PATH, HOME,
-        and the XDG/Claude config dirs that hold the OAuth credentials.
-        """
-        env = os.environ.copy()
-        env.pop("ANTHROPIC_API_KEY", None)
-        env.pop("ANTHROPIC_AUTH_TOKEN", None)
-        return env
-
-
-__all__ = ["ClaudeCLI", "ClaudeCLIError", "ClaudeResult", "DEFAULT_MODEL"]

src/second_brain/main.py

@@ -12,6 +12,7 @@ Commands:
 from __future__ import annotations
 
 import sys
+from datetime import datetime
 from pathlib import Path
 from typing import Optional
 from urllib.parse import urlparse
@@ -20,7 +21,7 @@ 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, utcnow
+from second_brain.models import Source, SourceStatus, SourceType
 
 
 # ---------------------------------------------------------------------------
@@ -73,7 +74,7 @@ def add(url: str, domain: str, focus: Optional[str], title: Optional[str]) -> No
             focus=focus,
             source_type=source_type,
             status=SourceStatus.PENDING,
-            ingested_at=utcnow(),
+            ingested_at=datetime.utcnow(),
         )
         sess.add(source)
 
@@ -121,24 +122,17 @@ def process(
     art_adapter = ArticleAdapter(config)
     assembler = ContextAssembler(config)
 
-    engine: Optional[ExtractionEngine] = None
-    if not skip_extract:
-        ex_cfg = config.extractor
     try:
-            engine = ExtractionEngine(
+        engine = ExtractionEngine(api_key=config.anthropic_api_key)
-                backend=ex_cfg["backend"],
-                model=ex_cfg["model"],
-                api_key=config.anthropic_api_key,
-                cli_binary=ex_cfg["cli_binary"],
-                timeout_sec=ex_cfg["timeout_sec"],
-            )
     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.id).filter(
+        query = sess.query(Source).filter(
             Source.status.in_([
                 SourceStatus.PENDING,
                 SourceStatus.PULLED,
@@ -147,19 +141,20 @@ def process(
         )
         if limit:
             query = query.limit(limit)
-        source_ids = [row[0] for row in query.all()]
+        sources = query.all()
 
-    if not source_ids:
+    if not sources:
         click.echo("[process] No sources to process.")
         return
 
-    click.echo(f"[process] Processing {len(source_ids)} source(s)…")
+    click.echo(f"[process] Processing {len(sources)} source(s)…")
 
-    for source_id in source_ids:
+    for source in sources:
-        with db.session() as sess:
-            source = sess.get(Source, source_id)
         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")
@@ -210,7 +205,7 @@ def process(
                         sess.add(extraction)
 
                     source.status = SourceStatus.ANALYZED
-                    source.updated_at = utcnow()
+                    source.updated_at = datetime.utcnow()
                     click.echo("    ✓ analyzed")
                 except Exception as exc:
                     source.error_message = str(exc)

src/second_brain/models.py

@@ -10,14 +10,9 @@ Tables:
 from __future__ import annotations
 
 import enum
-from datetime import datetime, timezone
+from datetime import datetime
 from typing import Optional
 
-
-def utcnow() -> datetime:
-    """Naive UTC timestamp. Replaces `datetime.utcnow()` (deprecated in 3.12)."""
-    return datetime.now(timezone.utc).replace(tzinfo=None)
-
 from sqlalchemy import (
     DateTime,
     Enum,
@@ -92,7 +87,7 @@ class Source(Base):
     author: Mapped[Optional[str]] = mapped_column(String(200), nullable=True)
 
     # Timestamps
-    ingested_at: Mapped[datetime] = mapped_column(DateTime, default=utcnow)
+    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)
 
@@ -133,7 +128,7 @@ class Extraction(Base):
     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=utcnow)
+    created_at: Mapped[datetime] = mapped_column(DateTime, default=datetime.utcnow)
 
     # Relationships
     source: Mapped["Source"] = relationship("Source", back_populates="extraction")
@@ -158,7 +153,7 @@ class WikiPage(Base):
     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=utcnow)
+    created_at: Mapped[datetime] = mapped_column(DateTime, default=datetime.utcnow)
     updated_at: Mapped[Optional[datetime]] = mapped_column(DateTime, nullable=True)
 
     def __repr__(self) -> str:
@@ -172,5 +167,4 @@ __all__ = [
     "SourceType",
     "Extraction",
     "WikiPage",
-    "utcnow",
 ]

src/second_brain/scheduler/runner.py

@@ -1,19 +1,8 @@
 """
-Overnight scheduler — rate-limit smoother.
+Overnight scheduler — rate-limit smoother with token-aware spacing.
 
-Spreads extraction calls across a configurable overnight window, enforcing
+Spreads LLM extraction calls across a configurable overnight window,
-a minimum gap between calls and a cap on calls-per-hour.
+enforcing a minimum gap between calls and a max-tokens-per-hour budget.
-
-Two throttle modes, picked from `config.extractor.backend`:
-- CLI backend (default): throttle by request count. Max OAuth doesn't bill
-  per-token; the relevant limit is the 5h-window message cap and the weekly
-  cap, both of which are call-shaped. `scheduler.max_calls_per_hour` is the
-  knob.
-- API backend: keep the legacy token budget — `scheduler.max_tokens_per_hour`.
-
-Source-status fix vs the previous version: this loop processes TRANSCRIBED
-sources and transitions them to ANALYZED (the prior code looked for ANALYZED
-sources and re-ran extraction on them, which was a no-op + drift bug).
 Idempotent: safe to restart mid-run.
 """
 
@@ -25,121 +14,99 @@ from typing import Optional
 
 from second_brain.config import Config
 from second_brain.database import get_database
-from second_brain.models import Source, SourceStatus, utcnow
+from second_brain.models import Source, SourceStatus
 
 
 class Scheduler:
-    """Process pending sources inside a time window, respecting rate caps."""
+    """
+    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))
 
-        ex_cfg = config.extractor
-        self.backend: str = ex_cfg["backend"]
-
-        # Throttle knobs (only the one matching the backend gets enforced)
-        self.max_calls_per_hour: int = int(sched.get("max_calls_per_hour", 30))
-        self.max_tokens_per_hour: int = int(sched.get("max_tokens_per_hour", 100_000))
-
     def run(self, dry_run: bool = False) -> None:
-        """Process TRANSCRIBED sources until window closes or queue empties."""
+        """
+        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)
-        ex_cfg = self.config.extractor
+        engine = ExtractionEngine(api_key=self.config.anthropic_api_key)
-        engine = ExtractionEngine(
-            backend=ex_cfg["backend"],
-            model=ex_cfg["model"],
-            api_key=self.config.anthropic_api_key,
-            cli_binary=ex_cfg["cli_binary"],
-            timeout_sec=ex_cfg["timeout_sec"],
-        )
 
-        calls_this_hour: int = 0
         tokens_this_hour: int = 0
         hour_start: float = time.time()
         processed: int = 0
 
-        cap_desc = (
-            f"{self.max_calls_per_hour} calls/hr"
-            if self.backend == "cli"
-            else f"{self.max_tokens_per_hour:,} tokens/hr"
-        )
         print(
-            f"[Scheduler] backend={self.backend} window "
+            f"[Scheduler] Starting. Window {_fmt(self.window_start)}–{_fmt(self.window_end)}, "
-            f"{_fmt(self.window_start)}–{_fmt(self.window_end)} cap {cap_desc} "
+            f"budget {self.max_tokens_per_hour:,} tokens/hr, "
             f"gap {self.min_gap_seconds}s"
         )
 
         while self._in_window():
-            # Reset hourly buckets
+            # Reset hourly token bucket
             elapsed = time.time() - hour_start
             if elapsed >= 3600:
-                calls_this_hour = 0
                 tokens_this_hour = 0
                 hour_start = time.time()
-                elapsed = 0
-
-            # Pre-call budget check
-            if self._over_budget(calls_this_hour, tokens_this_hour):
-                wait = int(3600 - elapsed) + 5
-                print(f"[Scheduler] Hourly cap reached. Sleeping {wait}s…")
-                if not dry_run:
-                    time.sleep(wait)
-                calls_this_hour = 0
-                tokens_this_hour = 0
-                hour_start = time.time()
-                continue
 
             with db.session() as sess:
                 source = (
                     sess.query(Source)
-                    .filter(Source.status == SourceStatus.TRANSCRIBED)
+                    .filter(Source.status == SourceStatus.ANALYZED)
-                    .order_by(Source.ingested_at.asc())
                     .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)
 
-                    existing = sess.query(Extraction).filter_by(
+                    extraction = Extraction(
-                        source_id=source.id
-                    ).first()
-                    fields = result.to_db_dict()
-                    if existing:
-                        for k, v in fields.items():
-                            setattr(existing, k, v)
-                        existing.raw_response = getattr(result, "_raw_response", None)
-                        existing.input_tokens = getattr(result, "_input_tokens", None)
-                        existing.output_tokens = getattr(result, "_output_tokens", None)
-                    else:
-                        sess.add(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),
-                            **fields,
+                        **result.to_db_dict(),
-                        ))
+                    )
-
+                    sess.add(extraction)
                     source.status = SourceStatus.ANALYZED
-                    source.updated_at = utcnow()
+                    source.updated_at = datetime.utcnow()
 
-                    calls_this_hour += 1
                     used = (getattr(result, "_input_tokens", 0) or 0) + (
                         getattr(result, "_output_tokens", 0) or 0
                     )
@@ -151,26 +118,21 @@ class Scheduler:
                     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).")
 
-    # ------------------------------------------------------------------
-    # Helpers
-    # ------------------------------------------------------------------
-
-    def _over_budget(self, calls: int, tokens: int) -> bool:
-        if self.backend == "cli":
-            return calls + 1 > self.max_calls_per_hour
-        # API backend — assume ~2k tokens per call when projecting
-        return tokens + 2000 > self.max_tokens_per_hour
-
     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, end = self.window_start, self.window_end
+        start = self.window_start
+        end = self.window_end
+
         if start <= end:
             return start <= now <= end
-        # crosses midnight
+        else:
+            # Window crosses midnight (e.g., 22:00 → 06:00)
             return now >= start or now <= end