petal-power/second-brain
3
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

switch extractor to Claude CLI backend + pipeline fixes

Browse Source 
Run extraction under the Max OAuth subscription via `claude -p` instead
of the per-token Anthropic API. The new src/second_brain/llm/claude_cli.py
spawns the CLI in a hermetic tempdir so the host project's CLAUDE.md,
hooks, MCP config, and settings don't leak into the prompt. Uses
--json-schema with LLMExtraction.model_json_schema() so the CLI guarantees
valid structured output — replaces the brittle markdown-fence stripping
in the old engine. The Anthropic SDK is preserved as an optional "api"
backend selectable via config.

While in here, fix a handful of blockers that the smoke test surfaced:
- scheduler filtered ANALYZED instead of TRANSCRIBED, so it never
  actually advanced any sources
- process command read sources in a closed session, raising
  DetachedInstanceError before any work happened
- config.prompts_dir walked one parent too many, resolving outside
  the project and forcing the fallback prompt for every domain
- compiler called git rev-parse against a vault that was never
  git-init'd; now auto-inits with an empty seed commit and skips empty
  commits cleanly
- datetime.utcnow() deprecated in 3.12+ — single utcnow() helper in
  models.py keeps naive UTC semantics so no DB migration is needed
- sess.query(...).get() deprecated in SA 2.x → sess.get(...)
- dead `import anthropic` removed from compiler

Smoke test (article → process → accept → compile) succeeds end-to-end
with ANTHROPIC_API_KEY unset. a-review run saved under reviews/.

Co-Authored-By: Claude Opus 4 <noreply@anthropic.com>
This commit is contained in:
Travis Herbranson 2026-05-24 21:09:20 -04:00
parent 250ca9fd2f
commit c4793f3a7f
13 changed files with 2470 additions and 164 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

15
config/settings.toml
View File

@ -21,13 +21,24 @@ 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"
# Max Claude API tokens to consume per hour
max_tokens_per_hour = 100_000
# CLI backend: cap calls per rolling hour (Max sub has 5h-window message caps).
max_calls_per_hour = 30
# 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

7
pyproject.toml
View File

@ -4,7 +4,6 @@ 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",
@ -18,6 +17,12 @@ 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"

42
reviews/a-review-2026-05-24.md Normal file
View File

@ -0,0 +1,42 @@
# 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.

50
src/second_brain/compiler/wiki.py
View File

@ -19,11 +19,9 @@ 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
from second_brain.models import Extraction, Source, SourceStatus, WikiPage, utcnow
class WikiCompiler:
@ -48,6 +46,7 @@ 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}
@ -73,7 +72,7 @@ class WikiCompiler:
else:
stats["updated"] += 1
source.status = SourceStatus.PUBLISHED
source.updated_at = datetime.utcnow()
source.updated_at = utcnow()
stats["processed"] += 1
except Exception as exc:
print(f"[Compiler] Error processing source {source.id}: {exc}")
@ -125,7 +124,7 @@ class WikiCompiler:
)
sess.add(wiki_row)
else:
wiki_row.updated_at = datetime.utcnow()
wiki_row.updated_at = utcnow()
return created
@ -193,6 +192,30 @@ 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."""
@ -210,11 +233,22 @@ 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)
msg = f"second-brain: compile {count} source(s) [{datetime.utcnow().date()}]"
subprocess.run(["git", "commit", "-m", msg], cwd=vault, check=True)
# Bail out cleanly if there's nothing staged.
diff = subprocess.run(
["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,

22
src/second_brain/config.py
View File

@ -27,11 +27,22 @@ _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",
"max_tokens_per_hour": 100_000,
# 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,
},
}
@ -78,6 +89,12 @@ 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
@ -89,7 +106,8 @@ class Config:
@property
def prompts_dir(self) -> Path:
"""Directory containing per-domain prompt templates."""
here = Path(__file__).parent.parent.parent.parent # project root
# src/second_brain/config.py → ../../.. = project root
here = Path(__file__).resolve().parent.parent.parent
return here / "prompts"
def ensure_dirs(self) -> None:

209
src/second_brain/extractor/engine.py
View File

@ -1,9 +1,14 @@
"""
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.
Two backends:
- "cli" (default): subprocess call to `claude -p`, runs under Max OAuth.
Uses --json-schema for guaranteed valid JSON.
- "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
@ -12,86 +17,141 @@ import json
import os
from typing import Optional
import anthropic
from second_brain.extractor.schema import ExtractionResult, SourceMeta
from second_brain.extractor.schema import ExtractionResult, LLMExtraction, SourceMeta
from second_brain.llm.claude_cli import ClaudeCLI, ClaudeCLIError, DEFAULT_MODEL
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.
Stateless LLM extractor. Default backend: Claude CLI under OAuth.
Call extract() with a pre-assembled context bundle (prompt + transcript).
Returns a validated ExtractionResult or raises on failure.
Usage:
engine = ExtractionEngine() # cli backend
engine = ExtractionEngine(backend="api", ...) # api backend
result = engine.extract(prompt, source)
"""
MODEL = "claude-sonnet-4-6"
MAX_TOKENS = 4096
def __init__(
self,
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(
"ANTHROPIC_API_KEY is required. Set the environment variable or "
"pass api_key to ExtractionEngine."
"backend='api' requires ANTHROPIC_API_KEY (env var or constructor arg)."
)
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
self._client = anthropic.Anthropic(api_key=key)
self._cli = None
else:
raise ValueError(f"Unknown backend: {backend!r} (expected 'cli' or 'api')")
# ------------------------------------------------------------------
# Internal helpers
# Public API
# ------------------------------------------------------------------
def extract(self, prompt: str, source: Source) -> ExtractionResult:
"""Run a single-shot extraction call. Returns a validated ExtractionResult."""
print(
f"[Extractor] [{self.backend}] source={source.id} domain={source.domain} "
f"model={self.model}"
)
if self.backend == "cli":
text, structured, input_tokens, output_tokens, raw_response = (
self._extract_cli(prompt)
)
else:
text, input_tokens, output_tokens, raw_response = self._extract_api(prompt)
structured = None
print(
f"[Extractor] Done — input={input_tokens} output={output_tokens} tokens"
)
# Prefer the schema-validated structured payload when available.
if structured is not None:
llm = LLMExtraction.model_validate(structured)
else:
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
usage = response.usage
return raw, usage.input_tokens, usage.output_tokens, raw
# ------------------------------------------------------------------
# Parse + compose
# ------------------------------------------------------------------
@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."""
def _parse(raw: str) -> LLMExtraction:
"""Parse the model's JSON output into an LLMExtraction."""
text = raw.strip()
# Strip accidental markdown fences
# CLI with --json-schema should always be clean, but the API path
# occasionally wraps in ```json fences. Strip defensively.
if text.startswith("```"):
text = text.split("\n", 1)[-1]
text = text.rsplit("```", 1)[0].strip()
@ -101,9 +161,13 @@ class ExtractionEngine:
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]}"
f"Raw (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,
@ -112,17 +176,16 @@ 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=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", []),
summary=llm.summary,
key_points=llm.key_points,
entities=llm.entities,
claims=llm.claims,
open_questions=llm.open_questions,
contradictions=llm.contradictions,
)
@ -137,4 +200,4 @@ def _fmt_duration(seconds: Optional[int]) -> Optional[str]:
return f"{m}:{s:02d}"
__all__ = ["ExtractionEngine"]
__all__ = ["ExtractionEngine", "SYSTEM_PROMPT"]

34
src/second_brain/extractor/schema.py
View File

@ -13,6 +13,38 @@ 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)."""
@ -79,4 +111,4 @@ class ExtractionResult(BaseModel):
}
__all__ = ["ExtractionResult", "SourceMeta"]
__all__ = ["ExtractionResult", "LLMExtraction", "SourceMeta"]

5
src/second_brain/llm/__init__.py Normal file
View File

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

185
src/second_brain/llm/claude_cli.py Normal file
View File

@ -0,0 +1,185 @@
"""
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"]

37
src/second_brain/main.py
View File

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

14
src/second_brain/models.py
View File

@ -10,9 +10,14 @@ Tables:
from __future__ import annotations
import enum
from datetime import datetime
from datetime import datetime, timezone
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,
@ -87,7 +92,7 @@ class Source(Base):
author: Mapped[Optional[str]] = mapped_column(String(200), nullable=True)
# Timestamps
ingested_at: Mapped[datetime] = mapped_column(DateTime, default=datetime.utcnow)
ingested_at: Mapped[datetime] = mapped_column(DateTime, default=utcnow)
updated_at: Mapped[Optional[datetime]] = mapped_column(DateTime, nullable=True)
error_message: Mapped[Optional[str]] = mapped_column(Text, nullable=True)
@ -128,7 +133,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=datetime.utcnow)
created_at: Mapped[datetime] = mapped_column(DateTime, default=utcnow)
# Relationships
source: Mapped["Source"] = relationship("Source", back_populates="extraction")
@ -153,7 +158,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=datetime.utcnow)
created_at: Mapped[datetime] = mapped_column(DateTime, default=utcnow)
updated_at: Mapped[Optional[datetime]] = mapped_column(DateTime, nullable=True)
def __repr__(self) -> str:
@ -167,4 +172,5 @@ __all__ = [
"SourceType",
"Extraction",
"WikiPage",
"utcnow",
]

136
src/second_brain/scheduler/runner.py
View File

@ -1,8 +1,19 @@
"""
Overnight scheduler — rate-limit smoother with token-aware spacing.
Overnight scheduler — rate-limit smoother.
Spreads LLM extraction calls across a configurable overnight window,
enforcing a minimum gap between calls and a max-tokens-per-hour budget.
Spreads extraction calls across a configurable overnight window, enforcing
a minimum gap between calls and a cap on calls-per-hour.
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.
"""
@ -14,99 +25,121 @@ from typing import Optional
from second_brain.config import Config
from second_brain.database import get_database
from second_brain.models import Source, SourceStatus
from second_brain.models import Source, SourceStatus, utcnow
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)
"""
"""Process pending sources inside a time window, respecting rate caps."""
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.
ex_cfg = config.extractor
self.backend: str = ex_cfg["backend"]
Runs until the window closes or there are no more pending sources.
"""
# 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."""
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)
ex_cfg = self.config.extractor
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] Starting. Window {_fmt(self.window_start)}–{_fmt(self.window_end)}, "
f"budget {self.max_tokens_per_hour:,} tokens/hr, "
f"[Scheduler] backend={self.backend} window "
f"{_fmt(self.window_start)}–{_fmt(self.window_end)} cap {cap_desc} "
f"gap {self.min_gap_seconds}s"
)
while self._in_window():
# Reset hourly token bucket
# Reset hourly buckets
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.ANALYZED)
.filter(Source.status == SourceStatus.TRANSCRIBED)
.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)
extraction = Extraction(
existing = sess.query(Extraction).filter_by(
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),
**result.to_db_dict(),
)
sess.add(extraction)
source.status = SourceStatus.ANALYZED
source.updated_at = datetime.utcnow()
**fields,
))
source.status = SourceStatus.ANALYZED
source.updated_at = utcnow()
calls_this_hour += 1
used = (getattr(result, "_input_tokens", 0) or 0) + (
getattr(result, "_output_tokens", 0) or 0
)
@ -118,21 +151,26 @@ 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).")
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
# ------------------------------------------------------------------
# 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:
now = datetime.now().time().replace(second=0, microsecond=0)
start, end = self.window_start, self.window_end
if start <= end:
return start <= now <= end
else:
# Window crosses midnight (e.g., 22:00 → 06:00)
# crosses midnight
return now >= start or now <= end

1862
uv.lock generated Normal file
View File

File diff suppressed because it is too large Load Diff