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

Compare commits

base: petal-power:fbbb1a13a0d95cf31082ba174138fc1112355faa
Branches Tags
petal-power:main
petal-power:claude/config-local-overlay
petal-power:claude/source-edit-and-retry
petal-power:claude/laughing-shirley-52f4a6
..
compare: petal-power:4696d7f01548f5b885a56b317fb4ffb81ebb1460
Branches Tags
petal-power:main
petal-power:claude/config-local-overlay
petal-power:claude/source-edit-and-retry
petal-power:claude/laughing-shirley-52f4a6

No commits in common. "fbbb1a13a0d95cf31082ba174138fc1112355faa" and "4696d7f01548f5b885a56b317fb4ffb81ebb1460" have entirely different histories.
fbbb1a13a0 ... 4696d7f015

46 changed files with 644 additions and 4226 deletions
Show Stats Expand all files Collapse all files

41
CLAUDE.md
View File

@ -22,36 +22,18 @@ pending → pulled → transcribed → analyzed → accepted → published
src/second_brain/
├── adapters/ # Pull adapters (youtube.py, article.py)
├── context/ # Context assembler
├── embeddings/ # Chunk + nomic-embed-text + upsert into public.embeddings
├── extractor/ # LLM extraction engine + Pydantic schema
├── llm/ # claude_cli.py — thin subprocess wrapper around `claude -p`
├── compiler/ # Wiki compiler (reads vault, writes pages, git commits)
├── scheduler/ # Rate-limit smoother (call-count for CLI, token-budget for API)
└── web/ # FastAPI + Jinja2 + HTMX UI
alembic/ # Schema migrations for the second_brain Postgres schema
```
## Setup
```bash
cp config/settings.toml.example config/settings.toml
# Edit settings.toml — at minimum, point the paths and set SECOND_BRAIN_DATABASE_URL.
# Connection string format. Containerised (default):
# postgresql+psycopg://lovebug:<password>@homelab-postgres:5432/petalbrain
# Host-side dev runs:
# postgresql+psycopg://lovebug:<password>@127.0.0.1:5433/petalbrain
export SECOND_BRAIN_DATABASE_URL=...
# One-time, before first migration: bootstrap the schema as postgres
# superuser. lovebug doesn't have CREATE on petalbrain, so this can't be
# done from the app side — same pattern as wiki/trellis/ob1.
# CREATE SCHEMA second_brain AUTHORIZATION lovebug;
# GRANT ALL PRIVILEGES ON SCHEMA second_brain TO lovebug;
uv sync
uv run alembic upgrade head
# Edit settings.toml with your paths
uv run second-brain serve
```
@ -76,10 +58,6 @@ uv run second-brain serve
- `ANTHROPIC_API_KEY` — only required when `extractor.backend = "api"`. The default `cli` backend uses your Max OAuth and does not need it.
- `SECOND_BRAIN_CONFIG` — optional override for the settings file path.
- `SECOND_BRAIN_DATABASE_URL` — Postgres connection string. Falls back to `HERBYLAB_DATABASE_URL` if unset. Required at runtime; both `second_brain.database` and `second_brain.embeddings` read it. Use the SQLAlchemy `postgresql+psycopg://` form; the embeddings module strips the `+psycopg` prefix internally for raw psycopg.
- `OLLAMA_URL` — overrides `[embeddings].ollama_url`. Default `http://ollama:11434` (homelab docker network); for host-side runs use `http://127.0.0.1:11434`.
- `EMBEDDING_MODEL` — overrides `[embeddings].model`. Default `nomic-embed-text` (768-d).
- `SECOND_BRAIN_DB_POOL_MIN` / `SECOND_BRAIN_DB_POOL_MAX` — SQLAlchemy pool sizing knobs (defaults 1 / 9 overflow).
## Domains
@ -90,31 +68,24 @@ uv run second-brain serve
## Key files
- `config/settings.toml` — vault path, `[database]`, `[extractor]`, `[embeddings]`, `[scheduler]`
- `config/settings.toml` — vault path, DB path, scheduler window, `[extractor]` backend config
- `prompts/<domain>.md` — per-domain extraction prompt templates
- `src/second_brain/extractor/schema.py` — `LLMExtraction` (LLM-only fields, drives the `--json-schema` contract) + `ExtractionResult` (LLMExtraction + SourceMeta filled from the DB)
- `src/second_brain/llm/claude_cli.py` — CLI subprocess wrapper
- `src/second_brain/models.py` — SQLAlchemy models (all tables live in the `second_brain` schema) + `utcnow()` helper
- `src/second_brain/database.py` — Postgres engine + URL resolution
- `src/second_brain/embeddings/__init__.py` — chunk → Ollama → upsert into `public.embeddings`; best-effort with graceful degradation
- `alembic/` + `alembic.ini` — schema migrations for the `second_brain` schema (Alembic does **not** manage `public.embeddings`)
- `postgres-migration-planning.md` — original planning doc; superseded by this file once the migration shipped
- `src/second_brain/models.py` — SQLAlchemy models + `utcnow()` helper (replaces deprecated `datetime.utcnow`)
- `postgres-migration-planning.md` — open questions blocking the SQLite → Postgres+pgvector migration
## Conventions / gotchas
- **Naive UTC timestamps everywhere.** Use `from second_brain.models import utcnow` rather than `datetime.utcnow()` (deprecated in 3.12) or `datetime.now()` (local time). Postgres columns are `timestamp without time zone` — do not switch to `timestamptz`.
- **PK style.** Int autoincrement (`SERIAL`) — locked for compatibility with the existing schema shape. Do not switch to `BIGINT GENERATED ALWAYS AS IDENTITY` or UUID.
- **Naive UTC timestamps everywhere.** Use `from second_brain.models import utcnow` rather than `datetime.utcnow()` (deprecated in 3.12) or `datetime.now()` (local time).
- **Re-query inside the session.** SQLAlchemy ORM attribute access after the session closes raises `DetachedInstanceError`. The `process` command and the web accept/reject endpoints both collect IDs first, then re-query inside the per-iteration / per-request session block, and pre-compute any dict views needed for templates *inside* the `with db.session()` block.
- **Starlette `TemplateResponse` signature.** Starlette ≥ 0.36 requires `templates.TemplateResponse(request, name, context)`. The old `(name, context)` form raises `TypeError: unhashable type: 'dict'`. Do not put `"request": request` into the context dict — it's auto-injected now.
- **Vault git repo auto-init.** The wiki compiler calls `_ensure_git_repo()` on every run, which `git init`s the vault and creates an empty seed commit if no `.git` is present. `_git_commit()` also checks `git diff --cached --quiet` before committing to avoid empty commits.
- **Scheduler throttling shape.** The CLI backend is throttled by `max_calls_per_hour` (Max subscription limits are call-shaped, not token-shaped). The API backend uses `max_tokens_per_hour`. Both share `min_gap_seconds`. The scheduler targets sources in `TRANSCRIBED` state — earlier code mistakenly filtered on `ANALYZED`, making it a no-op.
- **Alembic + lovebug privilege gap.** `lovebug` owns the `second_brain` schema but does **not** have `CREATE` on the `petalbrain` database. `alembic/env.py` therefore guards `CREATE SCHEMA` behind a `pg_namespace` lookup, and the v1 migration intentionally drops the `IF NOT EXISTS` form (a bare `CREATE SCHEMA IF NOT EXISTS` raises permission denied even when the schema is already present). The schema is bootstrapped once, out-of-band, by `postgres` superuser.
- **public.embeddings is shared, not ours.** Alembic only manages the `second_brain` schema. The vector table itself is owned by `postgres` and shared across vault-mcp / ob1 / second-brain — we just INSERT into it. Rows are keyed by `(source_schema='second_brain', source_table='extractions', source_id, embedding_model='nomic-embed-text')`.
- **Embedding is best-effort.** The pipeline writes the relational `extractions` row first and `sess.flush()`es; only then calls `embed_extraction()`. Any failure (no DB URL, no Ollama, embedding 500) logs at WARNING and returns None — the relational row and file vault stay canonical.
- **File-based wiki, not petalbrain.wiki.** The wiki compiler still writes markdown into the configured `vault_path` and commits via git. We do **not** dual-write into `petalbrain.wiki` — that's vault-mcp's territory and intentionally out of scope.
## In-flight work
- **Postgres + pgvector migration.** SQLite is the current backing store. Travis wants to move to his existing Postgres + pgvector instance and use the shared `/projects/shared/python/embedding_chunking` module for vector indexing. Planning questions are enumerated in `postgres-migration-planning.md` at the project root; no migration work has started yet.
- **Vault location.** Recommended `/opt/projects/second-brain-vault/` as the production location (peer to the project dir). Not yet confirmed.
## Validation

52
alembic.ini
View File

@ -1,52 +0,0 @@
# Alembic config for the second_brain schema migrations.
#
# Mirrors vault-mcp/alembic.ini. The sqlalchemy.url placeholder lets the
# `alembic` CLI parse this file without env vars at import time; env.py
# overrides it from SECOND_BRAIN_DATABASE_URL / HERBYLAB_DATABASE_URL at
# runtime.
[alembic]
script_location = %(here)s/alembic
prepend_sys_path = .
path_separator = os
sqlalchemy.url = postgresql+psycopg://placeholder:placeholder@localhost/placeholder
[post_write_hooks]
[loggers]
keys = root,sqlalchemy,alembic
[handlers]
keys = console
[formatters]
keys = generic
[logger_root]
level = WARNING
handlers = console
qualname =
[logger_sqlalchemy]
level = WARNING
handlers =
qualname = sqlalchemy.engine
[logger_alembic]
level = INFO
handlers =
qualname = alembic
[handler_console]
class = StreamHandler
args = (sys.stderr,)
level = NOTSET
formatter = generic
[formatter_generic]
format = %(levelname)-5.5s [%(name)s] %(message)s
datefmt = %H:%M:%S

119
alembic/env.py
View File

@ -1,119 +0,0 @@
"""Alembic environment for second-brain — second_brain schema.
Mirrors vault-mcp/alembic/env.py. Reads SECOND_BRAIN_DATABASE_URL (with
HERBYLAB_DATABASE_URL as a fallback) so the same migration tree works
across local dev, CI, and production.
The connection bootstraps the `second_brain` schema and pins alembic's
own version table to that schema; migration scripts themselves set
search_path before issuing unqualified DDL.
The bootstrap-then-commit pattern below is the fix for the "autobegin
trap": running any further statement before context.begin_transaction()
would autobegin a new transaction, causing alembic's begin_transaction()
to nest as a savepoint that silently rolls back when the connection
closes. Do not regress this.
"""
from __future__ import annotations
import os
from logging.config import fileConfig
from dotenv import load_dotenv
from sqlalchemy import engine_from_config, pool
from alembic import context
# Make the project's `src/` importable so `second_brain.models` resolves
# without a `pip install -e .`.
import sys
from pathlib import Path
_PROJECT_ROOT = Path(__file__).resolve().parent.parent
sys.path.insert(0, str(_PROJECT_ROOT / "src"))
from second_brain.models import Base, SCHEMA # noqa: E402
load_dotenv()
config = context.config
if config.config_file_name is not None:
fileConfig(config.config_file_name)
def _resolve_db_url() -> str | None:
"""Pick the first usable URL out of the two homelab conventions."""
for var in ("SECOND_BRAIN_DATABASE_URL", "HERBYLAB_DATABASE_URL"):
v = os.environ.get(var)
if v:
return v
return None
_db_url = _resolve_db_url()
if _db_url:
config.set_main_option("sqlalchemy.url", _db_url)
target_metadata = Base.metadata
def run_migrations_offline() -> None:
url = config.get_main_option("sqlalchemy.url")
context.configure(
url=url,
target_metadata=target_metadata,
literal_binds=True,
dialect_opts={"paramstyle": "named"},
version_table_schema=SCHEMA,
include_schemas=True,
)
with context.begin_transaction():
context.run_migrations()
def run_migrations_online() -> None:
connectable = engine_from_config(
config.get_section(config.config_ini_section, {}),
prefix="sqlalchemy.",
poolclass=pool.NullPool,
)
with connectable.connect() as connection:
# Bootstrap the schema before alembic looks for alembic_version.
# The runtime role (lovebug) does NOT have CREATE on the petalbrain
# database — bare `CREATE SCHEMA IF NOT EXISTS` would fail with
# permission denied even though the schema already exists. So we
# only attempt to create it when the schema is genuinely missing
# (operator must have bootstrapped it once as postgres superuser:
# `CREATE SCHEMA second_brain AUTHORIZATION lovebug`).
# Commit explicitly so any DDL persists; otherwise the implicit
# transaction is rolled back when the connection closes. Avoid
# running any further statement here — it would autobegin a new
# transaction and cause alembic's `begin_transaction()` to nest as
# a savepoint, which silently rolls back when the connection closes.
# The migration script sets search_path itself.
row = connection.exec_driver_sql(
"SELECT 1 FROM pg_namespace WHERE nspname = %s",
(SCHEMA,),
).fetchone()
if row is None:
connection.exec_driver_sql(f"CREATE SCHEMA {SCHEMA}")
connection.commit()
context.configure(
connection=connection,
target_metadata=target_metadata,
version_table_schema=SCHEMA,
include_schemas=True,
)
with context.begin_transaction():
context.run_migrations()
if context.is_offline_mode():
run_migrations_offline()
else:
run_migrations_online()

27
alembic/script.py.mako
View File

@ -1,27 +0,0 @@
"""${message}
Revision ID: ${up_revision}
Revises: ${down_revision | comma,n}
Create Date: ${create_date}
"""
from __future__ import annotations
from collections.abc import Sequence
from alembic import op
import sqlalchemy as sa
${imports if imports else ""}
revision: str = ${repr(up_revision)}
down_revision: str | Sequence[str] | None = ${repr(down_revision)}
branch_labels: str | Sequence[str] | None = ${repr(branch_labels)}
depends_on: str | Sequence[str] | None = ${repr(depends_on)}
def upgrade() -> None:
${upgrades if upgrades else "pass"}
def downgrade() -> None:
${downgrades if downgrades else "pass"}

120
alembic/versions/4a1e2f6c9d10_v1_second_brain_pipeline_tables.py
View File

@ -1,120 +0,0 @@
"""v1 second_brain: sources, extractions, wiki_pages
Revision ID: 4a1e2f6c9d10
Revises:
Create Date: 2026-05-24 00:00:00.000000
Emits the v1 second_brain schema: the relational backbone of the
extraction pipeline. Enum types (`source_type`, `source_status`) live
inside the schema so they don't collide with anything else in the
petalbrain cluster.
env.py creates the second_brain schema and pins alembic_version to it
before this migration runs. Below we re-issue CREATE SCHEMA IF NOT
EXISTS defensively and SET search_path so unqualified names resolve.
Vector indexing lives in public.embeddings — managed out-of-band, not
by this migration tree.
"""
from __future__ import annotations
from collections.abc import Sequence
from alembic import op
revision: str = "4a1e2f6c9d10"
down_revision: str | Sequence[str] | None = None
branch_labels: str | Sequence[str] | None = None
depends_on: str | Sequence[str] | None = None
def upgrade() -> None:
"""Create the v1 second_brain schema."""
# The schema is bootstrapped by env.py (only if missing); skip the
# IF NOT EXISTS form here because lovebug doesn't have CREATE on the
# petalbrain database and the bare statement errors out even when the
# schema already exists. Just SET search_path and emit DDL inside it.
op.execute("SET search_path TO second_brain")
op.execute(
"CREATE TYPE source_type AS ENUM ('video', 'article')"
)
op.execute(
"""
CREATE TYPE source_status AS ENUM (
'pending', 'pulled', 'transcribed', 'analyzed',
'accepted', 'published', 'failed'
)
"""
)
op.execute(
"""
CREATE TABLE sources (
id SERIAL PRIMARY KEY,
url VARCHAR(2000) NOT NULL UNIQUE,
title VARCHAR(500),
source_type source_type NOT NULL DEFAULT 'video',
domain VARCHAR(50) NOT NULL DEFAULT 'development',
focus TEXT,
status source_status NOT NULL DEFAULT 'pending',
media_path VARCHAR(1000),
transcript_path VARCHAR(1000),
transcript_text TEXT,
published_at TIMESTAMP,
duration_seconds INTEGER,
author VARCHAR(200),
ingested_at TIMESTAMP NOT NULL DEFAULT (now() AT TIME ZONE 'UTC'),
updated_at TIMESTAMP,
error_message TEXT
)
"""
)
op.execute("CREATE INDEX idx_sources_status ON sources(status)")
op.execute("CREATE INDEX idx_sources_domain ON sources(domain)")
op.execute(
"""
CREATE TABLE extractions (
id SERIAL PRIMARY KEY,
source_id INTEGER NOT NULL UNIQUE
REFERENCES sources(id) ON DELETE CASCADE,
summary TEXT,
key_points JSON,
entities JSON,
claims JSON,
open_questions JSON,
contradictions JSON,
raw_response TEXT,
input_tokens INTEGER,
output_tokens INTEGER,
created_at TIMESTAMP NOT NULL DEFAULT (now() AT TIME ZONE 'UTC')
)
"""
)
op.execute(
"""
CREATE TABLE wiki_pages (
id SERIAL PRIMARY KEY,
vault_path VARCHAR(500) NOT NULL UNIQUE,
domain VARCHAR(50) NOT NULL,
title VARCHAR(300) NOT NULL,
git_sha_before VARCHAR(40),
git_sha_after VARCHAR(40),
created_at TIMESTAMP NOT NULL DEFAULT (now() AT TIME ZONE 'UTC'),
updated_at TIMESTAMP
)
"""
)
def downgrade() -> None:
"""Drop the v1 second_brain tables and enums (schema is left intact)."""
op.execute("SET search_path TO second_brain")
op.execute("DROP TABLE IF EXISTS wiki_pages")
op.execute("DROP TABLE IF EXISTS extractions")
op.execute("DROP TABLE IF EXISTS sources")
op.execute("DROP TYPE IF EXISTS source_status")
op.execute("DROP TYPE IF EXISTS source_type")

70
alembic/versions/7b3e8a5c1f29_v2_pipeline_settings.py
View File

@ -1,70 +0,0 @@
"""v2 second_brain: pipeline_settings table
Revision ID: 7b3e8a5c1f29
Revises: 4a1e2f6c9d10
Create Date: 2026-05-25 00:00:00.000000
Adds a single-row `pipeline_settings` table that the web dashboard edits
and the workers (current dev-side scheduler + future tower-side transcribe
worker) read at the start of each run. The transcription_* fields persist
now but are only consumed once the transcribe worker exists — the
extraction_* fields are wired into the existing scheduler in this commit.
Single-row enforcement: PK is fixed at `id=1` with a CHECK constraint, so
INSERT-as-upsert-by-PK keeps the table free of stale rows.
Schema/permission notes mirror the v1 migration: lovebug lacks CREATE on
the petalbrain database, so env.py has already bootstrapped the schema —
this migration just SETs search_path and emits DDL inside it.
"""
from __future__ import annotations
from collections.abc import Sequence
from alembic import op
revision: str = "7b3e8a5c1f29"
down_revision: str | Sequence[str] | None = "4a1e2f6c9d10"
branch_labels: str | Sequence[str] | None = None
depends_on: str | Sequence[str] | None = None
def upgrade() -> None:
op.execute("SET search_path TO second_brain")
op.execute(
"""
CREATE TABLE pipeline_settings (
id INTEGER PRIMARY KEY DEFAULT 1,
transcription_enabled BOOLEAN NOT NULL DEFAULT TRUE,
transcription_active_hours_start TIME,
transcription_active_hours_end TIME,
transcription_max_concurrent_gpu_jobs INTEGER NOT NULL DEFAULT 1,
transcription_max_video_length_seconds INTEGER,
transcription_max_items_per_run INTEGER NOT NULL DEFAULT 10,
extraction_enabled BOOLEAN NOT NULL DEFAULT TRUE,
extraction_active_hours_start TIME,
extraction_active_hours_end TIME,
extraction_max_items_per_run INTEGER NOT NULL DEFAULT 20,
updated_at TIMESTAMP NOT NULL
DEFAULT (now() AT TIME ZONE 'UTC'),
CONSTRAINT pipeline_settings_singleton CHECK (id = 1)
)
"""
)
# Seed the singleton row with the table's column defaults. Idempotent
# via ON CONFLICT so re-running this migration on a partly-applied DB
# (or a manual cherry-pick) is safe.
op.execute(
"""
INSERT INTO pipeline_settings (id) VALUES (1)
ON CONFLICT (id) DO NOTHING
"""
)
def downgrade() -> None:
op.execute("SET search_path TO second_brain")
op.execute("DROP TABLE IF EXISTS pipeline_settings")

50
alembic/versions/c8f1d9e34b7a_v3_source_claim_columns.py
View File

@ -1,50 +0,0 @@
"""v3 second_brain: sources.claimed_by / claimed_at
Revision ID: c8f1d9e34b7a
Revises: 7b3e8a5c1f29
Create Date: 2026-05-25 12:30:00.000000
Adds a lightweight claim mechanism so the tower and dev sides never grab
the same row off the queue. The claim flow used by both workers is:
BEGIN;
SELECT id, status, source_type
FROM second_brain.sources
WHERE <filter>
AND claimed_by IS NULL
ORDER BY ingested_at ASC
LIMIT 1
FOR UPDATE SKIP LOCKED;
UPDATE second_brain.sources
SET claimed_by = :worker, claimed_at = now() AT TIME ZONE 'UTC'
WHERE id = :id;
COMMIT;
`FOR UPDATE SKIP LOCKED` is the actual race-safety primitive — the column
pair is observability (who has it, since when, for stale-claim recovery).
A partial index would help once we have a real queue depth; deliberately
not added now per Travis's "no gold-plating".
"""
from __future__ import annotations
from collections.abc import Sequence
from alembic import op
revision: str = "c8f1d9e34b7a"
down_revision: str | Sequence[str] | None = "7b3e8a5c1f29"
branch_labels: str | Sequence[str] | None = None
depends_on: str | Sequence[str] | None = None
def upgrade() -> None:
op.execute("SET search_path TO second_brain")
op.execute("ALTER TABLE sources ADD COLUMN claimed_by VARCHAR(100)")
op.execute("ALTER TABLE sources ADD COLUMN claimed_at TIMESTAMP")
def downgrade() -> None:
op.execute("SET search_path TO second_brain")
op.execute("ALTER TABLE sources DROP COLUMN IF EXISTS claimed_at")
op.execute("ALTER TABLE sources DROP COLUMN IF EXISTS claimed_by")

33
config/settings.toml
View File

@ -2,9 +2,10 @@
# Copy this file and edit paths to match your setup.
# Override the config file path with SECOND_BRAIN_CONFIG env var.
# Path to your Obsidian vault (git-managed directory). The wiki compiler
# writes markdown into this directory and commits inside it. We do NOT
# dual-write into petalbrain.wiki — the file-based vault stays canonical.
# 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
@ -20,21 +21,6 @@ whisper_model = "small"
# Active knowledge domains
domains = ["development", "content", "business", "homelab"]
[database]
# Postgres + pgvector lives on the petalbrain instance.
#
# Preferred: leave `url` empty and export SECOND_BRAIN_DATABASE_URL (or the
# shared HERBYLAB_DATABASE_URL) in the environment so the password never
# lands in this checked-in file.
#
# Containerised second-brain runs should target homelab-postgres on the
# homelab docker network:
# postgresql+psycopg://lovebug:<password>@homelab-postgres:5432/petalbrain
#
# Host-side dev runs can point at the published port instead:
# postgresql+psycopg://lovebug:<password>@127.0.0.1:5433/petalbrain
url = ""
[extractor]
# "cli" → runs `claude -p` under your Max OAuth (no per-token billing).
# "api" → uses Anthropic SDK, requires ANTHROPIC_API_KEY.
@ -43,17 +29,6 @@ model = "claude-sonnet-4-6"
cli_binary = "claude"
timeout_sec = 600
[embeddings]
# Best-effort: extraction summaries are chunked (512 tokens / 64 overlap)
# and embedded via Ollama's nomic-embed-text (768-d) into the shared
# public.embeddings table. Failures (no DB / no Ollama) log a warning and
# the pipeline keeps moving — the relational + file vault stay canonical.
enabled = true
# Containerized default; for host-side runs set OLLAMA_URL=http://127.0.0.1:11434
ollama_url = "http://ollama:11434"
model = "nomic-embed-text"
embed_field = "summary"
[scheduler]
# Time window for overnight processing (24h format, may cross midnight)
window_start = "22:00"

69
config/settings.toml.example
View File

@ -1,69 +0,0 @@
# 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 your Obsidian vault (git-managed directory). The wiki compiler
# writes markdown into this directory and commits inside it. We do NOT
# dual-write into petalbrain.wiki — the file-based vault stays canonical.
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"]
[database]
# Postgres + pgvector lives on the petalbrain instance.
#
# Preferred: leave `url` empty and export SECOND_BRAIN_DATABASE_URL (or the
# shared HERBYLAB_DATABASE_URL) in the environment so the password never
# lands in this checked-in file.
#
# Containerised second-brain runs should target homelab-postgres on the
# homelab docker network:
# postgresql+psycopg://lovebug:<password>@homelab-postgres:5432/petalbrain
#
# Host-side dev runs can point at the published port instead:
# postgresql+psycopg://lovebug:<password>@127.0.0.1:5433/petalbrain
url = ""
[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
[embeddings]
# Best-effort: extraction summaries are chunked (512 tokens / 64 overlap)
# and embedded via Ollama's nomic-embed-text (768-d) into the shared
# public.embeddings table. Failures (no DB / no Ollama) log a warning and
# the pipeline keeps moving — the relational + file vault stay canonical.
enabled = true
# Containerized default; for host-side runs set OLLAMA_URL=http://127.0.0.1:11434
ollama_url = "http://ollama:11434"
model = "nomic-embed-text"
embed_field = "summary"
[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_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

167
deploy/tower/README.md
View File

@ -1,167 +0,0 @@
# Tower-side transcribe worker — deploy notes
Target host: **EndeavourOS tower** (Arch base), RTX 3080. Runs the
`second-brain transcribe-worker` daemon under systemd. Reaches the
petalbrain Postgres over WireGuard at `db.wg.herbylab.dev:5432`.
This worker only does **pull + transcribe** for VIDEO sources. Articles
and all extraction/embedding stay on the dev side.
---
## 1. System packages
```bash
# Build / runtime tooling.
sudo pacman -S --needed base-devel git ffmpeg python uv
# NVIDIA — driver, CUDA, cuDNN. faster-whisper / CTranslate2 needs cuDNN
# at runtime; missing-symbol errors at first transcribe are usually a
# cuDNN install gap. Use whichever driver/cuda channel matches your
# Arch derivative; on stock EndeavourOS the AUR/community packages work:
sudo pacman -S --needed nvidia nvidia-utils cuda cudnn
# Confirm the GPU is visible.
nvidia-smi
```
yt-dlp comes in via the Python project (`uv sync` step below), so the
distro yt-dlp is *optional*. Leave it off the host unless you want it on
your CLI PATH.
---
## 2. Project checkout + venv
```bash
sudo mkdir -p /opt/projects/homelab
sudo chown herbyadmin:herbyadmin /opt/projects/homelab
cd /opt/projects/homelab
git clone gitea-lovebug:petal-power/second-brain.git
cd second-brain
# Install the runtime deps + the `tower` extra (faster-whisper / CTranslate2).
# The dev side does NOT install --extra tower; only the tower needs it.
uv sync --extra tower
```
`faster-whisper` will pull a couple hundred MB of CUDA wheels. First-run
model download (~3 GB for large-v3) lands under `~/.cache/huggingface/`
the first time the worker transcribes; pre-warm it if you want:
```bash
uv run python -c "from faster_whisper import WhisperModel; WhisperModel('large-v3', device='cuda', compute_type='float16')"
```
---
## 3. WireGuard
The worker assumes the tunnel is already configured and that
`wg-quick@wg-lan.service` brings it up at boot. Topology:
| Peer | WG IP |
|------|-------|
| hub (herbydev) | `10.99.0.1` |
| tower | `10.99.0.2` |
`db.wg.herbylab.dev` resolves to `10.99.0.1`. Confirm connectivity from
the tower:
```bash
ping -c2 db.wg.herbylab.dev
nc -vz db.wg.herbylab.dev 5432
```
Tunnel setup itself is **out of scope for this README** — that's
configured separately as part of Travis's infra.
---
## 4. EnvironmentFile
```bash
sudo install -m 0600 -o root -g root \
deploy/tower/second-brain-transcribe.env.example \
/etc/default/second-brain-transcribe
sudo $EDITOR /etc/default/second-brain-transcribe # paste the lovebug password
```
The lovebug password lives in `/opt/backups/postgres-consolidation/credentials.env`
on the dev host. Copy it across through whatever channel you trust
(personal Bitwarden, scp over WG, etc.) — do not commit it.
---
## 5. systemd unit
```bash
sudo install -m 0644 deploy/tower/second-brain-transcribe.service \
/etc/systemd/system/second-brain-transcribe.service
sudo systemctl daemon-reload
sudo systemctl enable --now second-brain-transcribe.service
```
The unit declares `After=` + `Wants=` on `wg-quick@wg-lan.service`, so
the worker starts after the tunnel comes up at boot. If the tunnel
disappears mid-run the worker keeps polling and logs DB-unreachable
warnings; it doesn't crash-loop.
### Day-to-day commands
```bash
# What's it doing right now?
systemctl status second-brain-transcribe.service
journalctl -u second-brain-transcribe.service -f
# Pause without uninstalling — toggle from the dashboard at
# https://brain.herbylab.dev/settings (or whichever host).
# Or stop the unit:
sudo systemctl stop second-brain-transcribe.service
# Pick up code updates:
cd /opt/projects/homelab/second-brain
git pull
uv sync --extra tower
sudo systemctl restart second-brain-transcribe.service
```
---
## 6. Validation — what to confirm once WG is live
These can only be verified on the tower itself:
1. **WG reachability.** `nc -vz db.wg.herbylab.dev 5432` succeeds.
2. **DB auth.** `uv run alembic current` returns the latest revision id.
3. **GPU large-v3 path.** With a short test video queued via
`second-brain add <url>` on the dev side, run `uv run second-brain transcribe-worker --once`
on the tower. Watch `journalctl` — first run pulls the model from
HuggingFace (slow), subsequent runs are fast. The source row should
end up in `TRANSCRIBED` with `transcript_text` populated and
`claimed_by` cleared.
4. **Race safety smoke test.** Run two `--once` invocations in parallel
against an empty queue, then with one PENDING video. Only one should
claim it; the other should see no work.
5. **Settings round-trip.** Flip `transcription_enabled` off on the
dashboard; within one poll cycle (~15s) the worker logs
`transcription_enabled=false — sleeping`. Flip back on, the next
PENDING video gets picked up.
---
## Follow-ups (out of scope for this round)
- **Media + subtitles directory currently lives on the tower's local
disk** (under `~/.local/share/second-brain/`). Move to a NAS mount
later so the dev side can compile/inspect the SRTs without a
cross-machine fetch. Open question on the right mount point + cred
flow — see the project's `postgres-migration-planning.md` for the
related vault-location decision Travis still owes.
- **Tower-side observability.** No metrics endpoint yet; journald is
the only signal. A `/api/worker-status` heartbeat the dashboard could
read would be nice once two workers exist.
- **GPU concurrency > 1.** `transcription_max_concurrent_gpu_jobs`
exists in `pipeline_settings` but the worker only runs one job at a
time. Scale-out would need a per-job semaphore (or just running N
worker instances with distinct `SECOND_BRAIN_WORKER_ID`s).

24
deploy/tower/second-brain-transcribe.env.example
View File

@ -1,24 +0,0 @@
# /etc/default/second-brain-transcribe
#
# Copy to /etc/default/second-brain-transcribe, chmod 0600, chown root:root
# (systemd reads it before dropping to User=herbyadmin so a non-readable
# permissions mode is fine — the herbyadmin user never touches this file).
# REQUIRED — petalbrain Postgres reached over the WireGuard tunnel.
# Hub-side hostname is db.wg.herbylab.dev (10.99.0.1); the tower is 10.99.0.2.
# No TLS — WireGuard already encrypts, and lovebug doesn't require SSL.
SECOND_BRAIN_DATABASE_URL=postgresql+psycopg://lovebug:REPLACE_ME@db.wg.herbylab.dev:5432/petalbrain
# OPTIONAL — only set if the tower has a non-standard whisper layout.
# Defaults from settings.toml: model=large-v3, device=cuda, compute_type=float16.
# WHISPER_MODEL=large-v3
# WHISPER_DEVICE=cuda
# WHISPER_COMPUTE_TYPE=float16
# OPTIONAL — identifier stamped into sources.claimed_by (default tower:<hostname>).
# Useful if you ever run multiple tower workers and want to tell them apart.
# SECOND_BRAIN_WORKER_ID=tower-main
# OPTIONAL — bump worker log verbosity. INFO is the default.
# PYTHONUNBUFFERED=1
# SECOND_BRAIN_LOG_LEVEL=DEBUG

46
deploy/tower/second-brain-transcribe.service
View File

@ -1,46 +0,0 @@
[Unit]
Description=second-brain transcribe worker (tower-side, faster-whisper on GPU)
Documentation=https://gitea.plantbasedsoutherner.com/petal-power/second-brain
# The worker reaches petalbrain over WireGuard, so the tunnel must be up first.
# `Wants=` is the soft form — if the tunnel disappears later the worker logs
# DB-unreachable warnings and retries, it doesn't crash-loop.
After=network-online.target wg-quick@wg-lan.service
Wants=network-online.target wg-quick@wg-lan.service
[Service]
Type=simple
User=herbyadmin
Group=herbyadmin
# Adjust to wherever you `git clone`d the repo on the tower.
WorkingDirectory=/opt/projects/homelab/second-brain
# Env (SECOND_BRAIN_DATABASE_URL, optional knobs). 0600, owned by herbyadmin.
EnvironmentFile=/etc/default/second-brain-transcribe
# `uv run` resolves the project venv. Keep the worker in the foreground so
# systemd owns the lifecycle; the worker handles SIGTERM/SIGINT cleanly and
# exits after the current iteration finishes.
ExecStart=/usr/bin/uv run second-brain transcribe-worker
Restart=always
RestartSec=10
# Crash loop guard — systemd will give up after this if the unit keeps
# dying. The worker itself logs+backs-off on DB outages so this only
# trips on genuine failure.
StartLimitIntervalSec=300
StartLimitBurst=5
# A few sandboxing nice-to-haves. Loose on purpose — the worker needs
# /dev/nvidia* for CUDA and writes to user-owned media/subtitles dirs.
ProtectSystem=full
ProtectHome=no
NoNewPrivileges=true
# Stdout/stderr → journald.
StandardOutput=journal
StandardError=journal
[Install]
WantedBy=multi-user.target

20
deploy/web/.env.example
View File

@ -1,20 +0,0 @@
# deploy/web/.env — runtime secrets for the second-brain-web container.
#
# Copy this to deploy/web/.env, fill in REPLACE_ME, chmod 0600. The real
# .env is gitignored (see deploy/web/.gitignore). Source of truth for
# the lovebug password: /opt/backups/postgres-consolidation/credentials.env
# on herbys-dev (mode 0600, host-only).
# REQUIRED. Containerised connection — the web container reaches Postgres
# by docker-network DNS name, NOT via the host's 127.0.0.1:5433 publish.
SECOND_BRAIN_DATABASE_URL=postgresql+psycopg://lovebug:REPLACE_ME@homelab-postgres:5432/petalbrain
# OPTIONAL. The dashboard does not currently embed search queries, but
# the embedding module reads this lazily. Default targets the homelab
# `ollama` container.
OLLAMA_URL=http://ollama:11434
EMBEDDING_MODEL=nomic-embed-text
# OPTIONAL. Pool sizing — single-instance web doesn't need much.
# SECOND_BRAIN_DB_POOL_MIN=1
# SECOND_BRAIN_DB_POOL_MAX=4

2
deploy/web/.gitignore vendored
View File

@ -1,2 +0,0 @@
# Real .env carries the lovebug Postgres password. Never commit.
.env

63
deploy/web/Dockerfile
View File

@ -1,63 +0,0 @@
# second-brain web container — review/dashboard UI only.
#
# Slim by design: this image runs `uvicorn` against second_brain.web.app.
# It does NOT need ffmpeg, the claude CLI, faster-whisper, or any GPU
# tooling. Extraction runs on the dev host's CLI; transcription runs on
# the tower. The web role is just "FastAPI + psycopg + Jinja templates".
FROM python:3.12-slim AS base
ENV PYTHONUNBUFFERED=1 \
PIP_DISABLE_PIP_VERSION_CHECK=1 \
UV_LINK_MODE=copy \
UV_PROJECT_ENVIRONMENT=/app/.venv
# uv is fetched as a static binary from its official image so we don't
# pull a full python stack to install it.
COPY --from=ghcr.io/astral-sh/uv:latest /uv /uvx /usr/local/bin/
# `git` is needed at *build* time because pyproject.toml pins
# embedding-chunking to a Gitea VCS source. `ca-certificates` so the
# https handshake validates. Slim image to clean up apt afterwards.
RUN apt-get update \
&& apt-get install -y --no-install-recommends git ca-certificates \
&& rm -rf /var/lib/apt/lists/*
WORKDIR /app
# Project metadata first so the layer cache survives source edits.
COPY pyproject.toml uv.lock ./
# `--frozen` enforces lockfile parity. `--no-dev` excludes pytest/ruff.
# `--no-install-project` so we don't try to install the (not-yet-copied)
# source tree in this layer.
RUN uv sync --frozen --no-dev --no-install-project
# Source + runtime config. config/settings.toml.example is committed; a
# real settings.toml gets bind-mounted in by compose at runtime.
COPY src/ ./src/
COPY config/ ./config/
COPY alembic/ ./alembic/
COPY alembic.ini ./alembic.ini
# Now install the project itself into the venv (puts the `second-brain`
# console script + the second_brain package on PATH).
RUN uv sync --frozen --no-dev
# Drop privileges. The image doesn't write to the host bind-mounts; the
# settings.toml mount is read-only.
RUN useradd --uid 1000 --create-home --shell /sbin/nologin app \
&& chown -R app:app /app
USER app
EXPOSE 8000
# Bind to all interfaces inside the container — the host port publish
# decides who can reach it from outside. `--proxy-headers` lets Traefik
# (off-box on 10.0.11.20) forward X-Forwarded-* correctly.
CMD ["/app/.venv/bin/uvicorn", \
"second_brain.web.app:app", \
"--host", "0.0.0.0", \
"--port", "8000", \
"--proxy-headers", \
"--forwarded-allow-ips=*"]

159
deploy/web/README.md
View File

@ -1,159 +0,0 @@
# second-brain web app — deploy notes
Target host: **herbys-dev** (10.0.21.207). Runs the FastAPI + Jinja UI
under `docker compose`, on the existing `homelab` bridge network so it
can reach `homelab-postgres:5432` and `ollama:11434` by service DNS.
Traefik (file-provider VM at **10.0.11.20**) handles TLS + the public
hostname `brain.herbylab.dev`. The compose file does NOT carry
`traefik.*` labels — file-provider Traefik can't see them.
---
## ⚠ Security note (no SSO)
Travis opted **not** to put this behind Authentik for the initial roll-out.
The web app has **no authentication, no CSRF, no rate limiting**, and it
exposes mutating endpoints (accept/reject sources, edit pipeline settings).
→ The Traefik route should be reachable **from LAN / trusted networks
only**. If you ever publish this past Cloudflare for off-LAN access, gate
it behind the Authentik forward-auth middleware first, or add an
`auth_required` decorator to the FastAPI app.
---
## 1. One-time setup on herbys-dev
```bash
cd /opt/projects/homelab/second-brain/deploy/web
# Real env file — paste the lovebug password.
cp .env.example .env
chmod 0600 .env
$EDITOR .env
# SECOND_BRAIN_DATABASE_URL must use host=homelab-postgres (NOT 127.0.0.1).
# Password lives at /opt/backups/postgres-consolidation/credentials.env.
# Build + start. Compose will join the existing external `homelab` net.
docker compose up -d --build
# Confirm it's healthy.
docker compose ps
docker compose logs --tail 50
curl -fsS http://10.0.21.207:8080/dashboard | head
```
Day-to-day:
```bash
docker compose logs -f # tail
docker compose restart # bounce (config reload)
docker compose pull && docker compose up -d # if image were registry-hosted
git pull && docker compose up -d --build # build-on-host update path
```
---
## 2. INFRA STEPS — Travis applies these manually
The compose stack is *up* once `docker compose up -d --build` returns.
The **public URL** at `https://brain.herbylab.dev` only resolves after
the two steps below land on their respective infra. They live on
machines outside this repo's authority (Knot on the DNS box; Traefik on
10.0.11.20), so this README documents them rather than executing them.
### 2a. Knot DNS — A record for `brain.herbylab.dev`
Add this to the `herbylab.dev` zone file on the Knot host (file path /
include layout matches the existing per-service A records — pick the
spot used for the other `*.herbylab.dev` services):
```zone
brain.herbylab.dev. 300 IN A 10.0.11.20
```
300s TTL matches the existing convention. Reload Knot:
```bash
sudo knotc reload
dig +short brain.herbylab.dev @<knot-host> # should return 10.0.11.20
```
### 2b. Traefik file-provider router on 10.0.11.20
Add the snippet below to the Traefik dynamic config (same file the rest
of the `*.herbylab.dev` services live in — file-provider config picks
up changes without a restart):
```yaml
http:
routers:
second-brain:
rule: "Host(`brain.herbylab.dev`)"
entryPoints:
- websecure
tls:
certResolver: cloudflare
service: second-brain
# No middleware on day one. If/when SSO lands, prepend the
# authentik forward-auth chain here (see other routers for the
# shape) and remove the LAN restriction.
services:
second-brain:
loadBalancer:
servers:
- url: "http://10.0.21.207:8080"
passHostHeader: true
```
Verify on the Traefik host:
```bash
# Watch for config-reload events:
docker logs traefik --tail 20 -f
# Or hit the API if it's exposed:
curl -fsS http://localhost:8080/api/http/routers/second-brain@file
```
Once both 2a and 2b are in place, `https://brain.herbylab.dev/dashboard`
should resolve through Cloudflare → Traefik → 10.0.21.207:8080 → the
`second-brain-web` container.
---
## 3. Verification checklist
After `docker compose up -d --build`:
| Check | Command | Expect |
|---|---|---|
| Container up | `docker compose ps` | `second-brain-web` is `Up`, not restarting |
| Startup log clean | `docker compose logs --tail 50` | uvicorn `Started server process` + `Application startup complete`, no tracebacks |
| Local-host reachable | `curl -fsS http://10.0.21.207:8080/dashboard \| head -1` | `<!DOCTYPE html>` or similar |
| DB reachable | `curl -fsS http://10.0.21.207:8080/dashboard` | dashboard renders status counts (proves the homelab-net path to Postgres) |
| Settings round-trip | open `/settings` in a browser, save | green "Saved." flash + new `updated_at` |
After Travis's 2a + 2b infra steps:
| Check | Command | Expect |
|---|---|---|
| DNS | `dig +short brain.herbylab.dev` | `10.0.11.20` |
| TLS | `curl -fsS https://brain.herbylab.dev/dashboard \| head -1` | dashboard HTML |
| Cert | `openssl s_client -connect brain.herbylab.dev:443 -servername brain.herbylab.dev <<<''` | Cloudflare-issued cert |
---
## 4. Follow-ups (not built this round)
- **SSO via Authentik.** Add the existing forward-auth middleware to the
Traefik router once we want this off-LAN. Authentik lives at
10.0.21.207:9000 (same host as this container — different port).
- **CSRF.** Settings save is an HTMX POST with no token; trivial to
forge from a same-origin browser. Acceptable while LAN-restricted.
- **Image registry.** Currently build-on-host. Push to Gitea's container
registry (`gitea.plantbasedsoutherner.com/petal-power/second-brain-web`)
when there's a second box that needs to pull the image.
- **Dev-side extraction systemd timer.** Travis is running extraction
manually for now; the timer/cron path is deferred.

41
deploy/web/compose.yml
View File

@ -1,41 +0,0 @@
# second-brain web — compose file for herbys-dev (10.0.21.207).
#
# House style: file is named `compose.yml`, no bare `docker run`.
# Bring up with: docker compose up -d --build
# Tear down with: docker compose down
#
# The Traefik VM lives on a different host (10.0.11.20) and uses the
# file provider, NOT docker label discovery. We therefore publish the
# uvicorn port back to a fixed host:port that Traefik's static service
# definition can point at. There are deliberately NO traefik.* labels
# in this file — they'd be silently ignored.
#
# The DB connection is the **containerised** one: `homelab-postgres:5432`
# on the homelab bridge network. Do NOT switch this to 127.0.0.1:5433
# from the container — that loops back inside the container, not the host.
services:
second-brain-web:
build:
context: ../..
dockerfile: deploy/web/Dockerfile
image: second-brain-web:latest
container_name: second-brain-web
restart: unless-stopped
env_file:
- .env
environment:
SECOND_BRAIN_CONFIG: /app/config/settings.toml
volumes:
- ../../config/settings.toml:/app/config/settings.toml:ro
networks:
- homelab
# Bind to the herbys-dev LAN IP only so this isn't accidentally
# exposed on other interfaces. Traefik (10.0.11.20) reaches the
# service at http://10.0.21.207:8080 via the LAN.
ports:
- "10.0.21.207:8080:8000"
networks:
homelab:
external: true

178
postgres-migration-planning.md
View File

@ -1,118 +1,78 @@
# Postgres + pgvector migration — implementation notes
# Postgres + pgvector migration — planning questions
Status: **shipped** (2026-05-24). Originally a planning checklist; rewritten
to capture the implemented design and the decisions that landed. Kept in the
repo so the next iteration has the "why" without grepping commits.
Status: awaiting input from Travis before scoping the migration.
Trigger: user pivoted away from SQLite — wants to use the existing Postgres + pgvector instance, plus the shared `/projects/shared/python/embedding_chunking` module.
---
## What shipped
## 1. Connection / secrets
- Backing store switched from SQLite to the existing **petalbrain** Postgres
instance (pgvector/pg17), connecting over the homelab docker network at
`homelab-postgres:5432`.
- Relational tables (`sources`, `extractions`, `wiki_pages`) live in a new
`second_brain` schema owned by the **`lovebug`** role — no dedicated service
role.
- Embeddings reuse the **shared `public.embeddings` table** (HNSW vector index
already in place). Rows are keyed by
`(source_schema='second_brain', source_table='extractions', source_id, embedding_model='nomic-embed-text')`.
For this round we embed **extraction summaries only** (not transcript
chunks, not key_points / claims).
- Embedding pipeline mirrors `vault_mcp.core.embeddings`: shared
`embedding_chunking` for 512/64-token chunking → Ollama
`nomic-embed-text` (768-d, `num_ctx=8192`) → delete-before-insert upsert
with vector text literals (`%s::vector`).
- **Graceful degradation** is the rule: missing DB URL, Ollama down, or any
DB error logs a warning and returns `None`. The relational row and the
file-based vault stay canonical.
- **Alembic** stands up the `second_brain` schema; `public.embeddings` is
managed out-of-band and intentionally outside the migration tree.
- File-based wiki compiler is unchanged. No dual-write to `petalbrain.wiki`.
- **No SQLite import** — we started clean on Postgres.
- [ ] Host, port, database name, user
- [ ] Where does the connection string live? Options:
- env var (e.g. `SECOND_BRAIN_DB_URL`)
- `config/settings.toml` (would need a `.local.toml` override so the password doesn't get committed)
- `.pgpass`
- other
- [ ] Is the Postgres instance on `herbydev`, the NAS, or elsewhere? Tailscale-only or LAN-reachable?
- [ ] Driver preference:
- `psycopg` v3 sync (default suggestion — pipeline + scheduler are sync today)
- `asyncpg` (only worth it if we also rewrite the web layer to async)
## 2. Schema layout
- [ ] Same Postgres **instance** as `vault-mcp`'s `petalbrain.wiki`? Yes / No
- [ ] If yes — same **database** with a dedicated schema (e.g. `second_brain.sources`, `second_brain.extractions`)? Or its own database entirely?
- [ ] Naming conventions to match:
- snake_case tables — assumed yes
- `created_at` / `updated_at` as `timestamptz` (vs current naive UTC) — preference?
- PK type: `bigserial` / `bigint identity` / UUID?
- [ ] Should second-brain integrate with `vault-mcp` at all?
- Option A: standalone — second-brain owns its tables, the wiki compiler still writes Obsidian markdown files
- Option B: dual-write — published extractions land in `petalbrain.wiki` via `vault-mcp` so they show up in the unified wiki
- Option C: replace the file-based compiler entirely with `vault-mcp` writes
## 3. Embeddings
- [ ] Public API of `/projects/shared/python/embedding_chunking`:
- function signatures
- what it returns (raw text chunks? chunks + vectors? chunker only, embeddings done separately?)
- embedding model + vector dimension (drives the `vector(N)` column type)
- any batching / rate-limit behavior we need to respect
- [ ] Where should embeddings attach?
- (a) on `Extraction` rows — semantic search over summaries / key_points / claims
- (b) on chunked transcripts — chunk-level retrieval for "find the part of the video that says X"
- (c) both (separate tables: `extraction_embeddings` + `transcript_chunks`)
- [ ] pgvector index strategy:
- HNSW — better recall, slower build, recommended default for personal-scale corpus
- IVFFlat — faster build, needs periodic `ANALYZE` + a meaningful `lists` value
- [ ] Distance metric: cosine (typical), L2, or inner product?
## 4. Migration tooling
- [ ] Schema is small and not yet in production. Cleanest path:
- SQLAlchemy `Base.metadata.create_all()` against Postgres
- one-off import script to move existing SQLite rows over (if any are worth keeping)
- [ ] Alternative: stand up Alembic now while the cost is low. Pays off the first time you `ALTER` a table in anger; costs ~1 session of setup.
- [ ] Travis's call — defaulting to `create_all` + import script unless you say otherwise.
## 5. Pooling
- [ ] Three consumers share the DB: CLI (`process`, `compile`), the FastAPI web server, the scheduler.
- [ ] Default plan: SQLAlchemy `QueuePool` with `pool_size=5, max_overflow=5`.
- [ ] Is PgBouncer already in front of this instance? If yes — switch to `NullPool` on our side and let PgBouncer pool.
## 6. Other open items (carryover from prior session)
- [ ] Vault location decision: I recommended `/opt/projects/second-brain-vault/` as a peer dir. Not yet confirmed.
- [ ] NAS mount: is it mounted on `herbydev`, and at what path? Relevant if media (downloaded videos, transcripts) should live there rather than on the herbydev SSD.
---
## Locked decisions (from the planning round)
## What I will produce once these are answered
| Question | Decision |
|---|---|
| DB instance | Existing petalbrain (homelab-postgres) |
| Schema | New `second_brain` schema (owned by `lovebug`) |
| Connection | Containerised: `homelab-postgres:5432` (NOT host `127.0.0.1:5433`) |
| Pooling | `psycopg_pool.ConnectionPool` min=1/max=10 — no PgBouncer in front |
| Driver | psycopg v3, SQLAlchemy uses `postgresql+psycopg://` |
| PK style | Int autoincrement (`SERIAL`) — kept |
| Timestamps | Naive UTC `timestamp without time zone` — kept |
| Embedding scope | Extraction summaries only (Phase 1) |
| Embedding model | `nomic-embed-text` (768-d), Ollama via `OLLAMA_URL` |
| Migration tool | Alembic, second-brain owns its own tree |
| `public.embeddings` | NOT managed by our Alembic — shared with vault-mcp / ob1 |
| File-based wiki | Kept intact; no dual-write into petalbrain.wiki |
| Old SQLite data | Not imported — clean start |
---
## Setup (operator runbook)
```bash
# 1. One-time, as postgres superuser, bootstrap the schema. lovebug doesn't
# have CREATE on petalbrain, so this can't be done from the app side.
docker exec homelab-postgres psql -U postgres -d petalbrain -c \
"CREATE SCHEMA second_brain AUTHORIZATION lovebug;
GRANT ALL PRIVILEGES ON SCHEMA second_brain TO lovebug;"
# 2. Point the project at the DB (env wins over settings.toml).
export SECOND_BRAIN_DATABASE_URL=\
"postgresql+psycopg://lovebug:<pw>@homelab-postgres:5432/petalbrain"
# 3. Install deps and run the migration.
uv sync
uv run alembic upgrade head
```
The `lovebug` password lives in `/opt/backups/postgres-consolidation/credentials.env`
(0600, host-only).
---
## Key files (post-migration)
- `src/second_brain/models.py` — schema-qualified ORM, `__table_args__` →
`second_brain`; native Postgres enum types are scoped to the schema too.
- `src/second_brain/database.py` — psycopg-v3 engine + URL resolution
(constructor arg → `SECOND_BRAIN_DATABASE_URL` → `HERBYLAB_DATABASE_URL`
→ `[database].url`).
- `src/second_brain/embeddings/__init__.py` — near-verbatim port of
`vault_mcp/core/embeddings.py` adapted to our key tuple.
- `alembic.ini` + `alembic/env.py` — schema-pinned Alembic; the env.py guards
`CREATE SCHEMA` behind `pg_namespace` because `lovebug` lacks `CREATE` on
the database.
- `alembic/versions/4a1e2f6c9d10_v1_second_brain_pipeline_tables.py` — v1
migration: enum types + `sources` + `extractions` + `wiki_pages`.
- `tests/test_smoke_embedding.py` — end-to-end smoke test exercising the full
Postgres + Ollama path. Auto-skips when either is unreachable.
---
## Known gotchas
1. **`CREATE SCHEMA IF NOT EXISTS` errors as `lovebug`** even when the schema
exists. `pg_namespace` lookup in `alembic/env.py` is the workaround.
2. **`+psycopg` URL prefix.** SQLAlchemy needs it; raw psycopg doesn't.
`database.py` adds it if missing; the embeddings module strips it.
3. **Pool conflict.** `database.py` opens an SQLAlchemy pool; `embeddings/__init__.py`
opens an independent psycopg pool. Both share the same DSN; both are small
(min=1) so they don't strain the cluster.
4. **`anthropic-skills:zero-check` needs `pytest-cov`.** Added to the new
`[dependency-groups].dev` block so `uv sync` resolves it.
---
## Next iterations (open, not blocked)
- Embed transcript chunks (or at least key_points / claims), not just summaries.
- Promote vector retrieval into the web UI (semantic search across extractions).
- Add a periodic backfill script that re-embeds extractions whose
`embedding_model` no longer matches the current default.
- Decide on the production vault path (`/opt/projects/second-brain-vault/`?).
1. Updated `src/second_brain/database.py` with a Postgres engine + URL resolution
2. Schema migration: new tables for embeddings (and chunks if we go with option (c))
3. Wiring of `embedding_chunking` into the extractor (or into a post-extraction step in the scheduler)
4. `config/settings.toml.example` updates with the new `[database]` and `[embeddings]` blocks
5. README / CLAUDE.md updates noting the Postgres dependency
6. One-shot SQLite → Postgres import script (only if there are rows worth keeping)

24
pyproject.toml
View File

@ -4,17 +4,12 @@ version = "0.1.0"
description = "Personal knowledge system: video/article extraction pipeline + LLM-maintained wiki"
requires-python = ">=3.12"
dependencies = [
"alembic>=1.13.0",
"click>=8.1.0",
"embedding-chunking",
"fastapi>=0.115.0",
"httpx>=0.28.0",
"jinja2>=3.1.0",
"psycopg[binary]>=3.2",
"psycopg-pool>=3.2",
"openai-whisper",
"pydantic>=2.0.0",
"python-dotenv>=1.0.0",
"python-multipart>=0.0.20",
"sqlalchemy>=2.0.0",
"tomli>=2.0.0",
"trafilatura>=1.9.0",
@ -28,23 +23,6 @@ dependencies = [
# Max OAuth subscription and has no Python-side Anthropic dependency.
api = ["anthropic>=0.40.0"]
# Tower-only: faster-whisper (CTranslate2) for the transcribe worker.
# Pulls in CUDA wheels — install with `uv sync --extra tower` on the
# tower box, leave it off on dev. CPU/int8 path works on this same wheel.
tower = ["faster-whisper>=1.0.0"]
[dependency-groups]
dev = [
"pytest>=8.0",
"pytest-cov>=5.0",
"ruff>=0.8",
]
[tool.uv.sources]
# Shared chunker that vault-mcp / ob1-enricher / ob1-reflector also pin.
# Same source so the 512/64 chunking stays in lockstep across services.
embedding-chunking = { git = "https://gitea.plantbasedsoutherner.com/petal-power/embedding-chunking.git", tag = "v0.1.0" }
[project.scripts]
second-brain = "second_brain.main:cli"

30
reviews/a-review-2026-05-24-2.md
View File

@ -1,30 +0,0 @@
# a-review — 2026-05-24
**Model:** gemini
**Diff base:** `HEAD~1` (12 files changed, 208 insertions(+), 21 deletions(-))
**Session reference:** none
**Context:** CLAUDE.md + session notes + expanded diff
---
Based on the provided context, here is my code review.
### [INFO] Project Dependencies
* **File:** `pyproject.toml`
* **Finding:** The change introduces a `[dependency-groups].dev` section to manage development-specific dependencies like `pytest`, `pytest-cov`, and `ruff`.
* **Reasoning:** This is a positive change that follows standard Python project structure. It cleanly separates the project's runtime dependencies from the tools required for testing and linting, which improves dependency management and clarifies the project's setup for new developers.
### [INFO] Code Cleanup
* **Files:** `src/second_brain/adapters/article.py`, `src/second_brain/adapters/youtube.py`, `src/second_brain/compiler/wiki.py`, `src/second_brain/context/assembler.py`, `src/second_brain/extractor/engine.py`, `src/second_brain/extractor/schema.py`, `src/second_brain/main.py`, `src/second_brain/scheduler/runner.py`, `src/second_brain/web/app.py`
* **Finding:** Multiple unused imports (`Optional`, `datetime`, `json`, `Path`, etc.) were removed from several files.
* **Reasoning:** This is a beneficial code hygiene practice. Removing unused imports reduces clutter, makes the code easier to read, and helps linters and static analysis tools run more effectively. It has no impact on runtime behavior but improves maintainability.
### Summary
The submitted changes are of high quality. They consist of two main improvements:
1. Properly defining development dependencies in `pyproject.toml`.
2. Cleaning up unused imports across the codebase, likely as a result of running a linter like the newly added `ruff`.
I found no security vulnerabilities, bugs, or convention violations. The changes are safe and improve the project's overall maintainability and structure. The implementation aligns with the goal of setting up a better development and testing environment.

65
reviews/a-review-2026-05-24-3.md
View File

@ -1,65 +0,0 @@
# a-review — 2026-05-24
**Model:** gemini
**Diff base:** `4696d7f` (24 files changed, 1496 insertions(+), 144 deletions(-))
**Session reference:** none
**Context:** CLAUDE.md + session notes + expanded diff
---
Excellent, the changes look solid overall. This is a substantial and well-executed migration from SQLite to Postgres, with the bonus of a new embedding feature. The code quality is high, particularly the database and transaction management logic.
I have a few findings, mostly related to documentation and test robustness.
### [WARNING] Documentation is now out of sync with the codebase
* **Location:** `CLAUDE.md`
* **Issue:** The project's main documentation file, `CLAUDE.md`, has not been updated to reflect these significant changes.
1. It still states the "Postgres + pgvector migration" is "In-flight work" and refers to `postgres-migration-planning.md`. This change appears to complete that migration.
2. The "Project layout" and "Key files" sections are missing the new `alembic/` directory for migrations and the new `src/second_brain/embeddings/` module.
3. The configuration sections do not mention the new `[database]` or `[embeddings]` tables in `config/settings.toml`.
* **Impact:** New developers (or future you!) will be confused by the discrepancy between the documentation and the actual state of the code. This makes maintenance harder and increases the risk of mistakes.
* **Recommendation:** After a large feature lands, it's crucial to update the project's core documentation. `CLAUDE.md` should be updated to reflect that the project now uses Postgres, describe the new embedding pipeline, and document the new configuration options. The `postgres-migration-planning.md` file should likely be archived or deleted.
### [INFO] Test fragility due to hardcoded string literal
* **Location:** `tests/test_smoke_embedding.py`, line 111
* **Issue:** The test query hardcodes the embedding model name: `AND embedding_model = 'nomic-embed-text'`. The application code correctly reads the model from the configuration. If the default model in `config/settings.toml` were to change in the future, this test would fail even if the application logic is correct.
* **Impact:** This creates a tight coupling between the test and a specific configuration value, making the test suite more brittle.
* **Recommendation:** The test should load the application `Config` and use the configured embedding model name in its verification query. This ensures the test is checking that the system works with *whatever* model is currently configured.
```python
# tests/test_smoke_embedding.py
# at the top
from second_brain.config import load_config
# in the test function
config = load_config()
embedding_model = config.embeddings.get("model", "nomic-embed-text")
# ... later in the raw SQL query
cur.execute(
"""
SELECT COUNT(*) ...
WHERE ...
AND embedding_model = %s
""",
(extraction_id, embedding_model),
)
```
### [INFO] Unused imports removed
* **Location:** Multiple files, including `src/second_brain/adapters/article.py`, `src/second_brain/compiler/wiki.py`, `src/second_brain/extractor/schema.py`, etc.
* **Issue:** The diff removes several unused imports. While cleaning this up is good, the fact they were present suggests a linter (like `ruff`, which is in the dev dependencies) may not be running automatically as part of the development or pre-commit workflow.
* **Impact:** This is a minor code hygiene issue, but consistent linting helps catch small issues early and maintains code quality across the project.
* **Recommendation:** Consider setting up a pre-commit hook to run `ruff check --fix` to automate the removal of unused imports and fix other simple linting issues.
## Summary
Overall, this is a high-quality contribution. The migration to Postgres is thoughtful and robust, correctly using Alembic for schema management and a connection pool for performance. The implementation of the embedding pipeline is well-contained, idempotent, and gracefully handles potential failures. The addition of a comprehensive smoke test is a fantastic practice that significantly increases confidence in the new functionality.
The primary area for improvement is in process: the project's documentation needs to be treated as part of the code and updated in the same change that implements new features or makes significant architectural modifications.
Aside from the documentation gap, the code is clean, correct, and demonstrates a strong understanding of the technologies involved. Great work.

1
src/second_brain/adapters/article.py
View File

@ -8,6 +8,7 @@ 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

110
src/second_brain/adapters/youtube.py
View File

@ -1,20 +1,18 @@
"""
YouTube pull adapter (yt-dlp only).
YouTube pull + Whisper transcribe adapter.
Transcription used to live here on a torch-whisper backend. It moved to
`second_brain.transcribe` (faster-whisper) once the pipeline split so the
tower owns transcription and the dev side never imports the heavy model.
This module is now safe to import from any process — yt-dlp is small, and
no GPU / whisper dependency is loaded at import time.
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
@ -87,18 +85,84 @@ class VideoDownloader:
}
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:
"""yt-dlp pull side of a video Source. Transcription is handled by
`second_brain.transcribe.Transcriber` on the tower."""
"""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."""
@ -123,12 +187,38 @@ class YouTubeAdapter:
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:
@ -139,4 +229,4 @@ def _parse_upload_date(upload_date: Optional[str]) -> Optional[datetime]:
return None
__all__ = ["VideoDownloader", "YouTubeAdapter"]
__all__ = ["VideoDownloader", "SubtitleExtractor", "YouTubeAdapter"]

154
src/second_brain/claim.py
View File

@ -1,154 +0,0 @@
"""Queue-claim helpers shared by the tower transcribe worker, the dev-side
process command, and the dev-side extract scheduler.
The race-safety primitive is `SELECT ... FOR UPDATE SKIP LOCKED`. The
`claimed_by` / `claimed_at` columns are observability + stale-claim
recovery only — Postgres row-locks are what actually keep two machines
from grabbing the same row.
Usage pattern (worker side, simplified)::
with db.session() as sess:
source_id = claim_next_source(
sess,
worker_id="tower",
statuses=[SourceStatus.PENDING, SourceStatus.PULLED],
source_type=SourceType.VIDEO,
)
# commit drops the row lock — the claimed_by/at columns persist.
if source_id is None:
# nothing to do; sleep and retry
return
with db.session() as sess:
source = sess.get(Source, source_id)
try:
...do work, mutate status...
finally:
release_claim(sess, source_id)
The claim is *released* once the row reaches a terminal status for this
stage (TRANSCRIBED, FAILED, etc.) so a follower stage can pick it up. If
a worker dies mid-claim, `reap_stale_claims` (called at worker startup)
clears claims older than the configured grace window.
"""
from __future__ import annotations
from datetime import timedelta
from typing import Iterable, Optional
from sqlalchemy import text
from sqlalchemy.orm import Session
from second_brain.models import Source, SourceStatus, SourceType, utcnow
def claim_next_source(
sess: Session,
*,
worker_id: str,
statuses: Iterable[SourceStatus],
source_type: Optional[SourceType] = None,
) -> Optional[int]:
"""Atomically claim the oldest unclaimed source matching the filters.
Returns the source's id (so the caller can re-fetch it in a fresh
session if it likes), or None if no row matched. The caller's
session-level COMMIT drops the row lock — the claim survives via the
`claimed_by` / `claimed_at` columns.
"""
if not statuses:
return None
status_values = [s.value for s in statuses]
params: dict = {
"statuses": status_values,
"worker": worker_id,
"now": utcnow(),
}
type_clause = ""
if source_type is not None:
type_clause = "AND source_type = :source_type"
params["source_type"] = source_type.value
# Raw SQL because SQLAlchemy ORM doesn't expose SKIP LOCKED cleanly
# across all minor versions and we want the exact lock semantics
# documented in the module header. `status::text` lets us bind a
# parameter list against the native enum cleanly.
pick = sess.execute(
text(
f"""
SELECT id
FROM second_brain.sources
WHERE status::text = ANY(:statuses)
{type_clause}
AND claimed_by IS NULL
ORDER BY ingested_at ASC
LIMIT 1
FOR UPDATE SKIP LOCKED
"""
),
params,
).first()
if pick is None:
return None
source_id = int(pick[0])
sess.execute(
text(
"""
UPDATE second_brain.sources
SET claimed_by = :worker,
claimed_at = :now
WHERE id = :id
"""
),
{"worker": worker_id, "now": params["now"], "id": source_id},
)
return source_id
def release_claim(sess: Session, source_id: int) -> None:
"""Drop the claim on a source so a follower stage can pick it up.
The work-state columns (`status`, `transcript_text`, …) are the
caller's responsibility — this just nulls the claim pair.
"""
src = sess.get(Source, source_id)
if src is None:
return
src.claimed_by = None
src.claimed_at = None
def reap_stale_claims(
sess: Session,
*,
older_than: timedelta = timedelta(minutes=30),
) -> int:
"""Clear claims older than the grace window — recovery for crashed workers.
Returns the number of rows reaped. Safe to call at worker startup;
pairs with the worker setting a short poll interval so a stuck
transcribe doesn't dam the queue forever.
"""
cutoff = utcnow() - older_than
result = sess.execute(
text(
"""
UPDATE second_brain.sources
SET claimed_by = NULL,
claimed_at = NULL
WHERE claimed_by IS NOT NULL
AND claimed_at < :cutoff
"""
),
{"cutoff": cutoff},
)
return result.rowcount or 0
__all__ = ["claim_next_source", "release_claim", "reap_stale_claims"]

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

@ -15,6 +15,7 @@ from __future__ import annotations
import re
import subprocess
from datetime import datetime
from pathlib import Path
from typing import Optional

85
src/second_brain/config.py
View File

@ -3,11 +3,6 @@ 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.
The relational store is Postgres (petalbrain DB, `second_brain` schema). Connection
details are resolved by `second_brain.database` — see that module for the URL
resolution order. The vault is still a file-based Obsidian directory; this
project does NOT dual-write to petalbrain.wiki.
"""
from __future__ import annotations
@ -26,23 +21,12 @@ except ModuleNotFoundError:
# ---------------------------------------------------------------------------
_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"],
"whisper": {
# Tower default: faster-whisper large-v3 on cuda/float16. The
# transcribe helper picks int8 automatically if the device is cpu.
"whisper_model": "large-v3",
"whisper_device": "cuda",
"whisper_compute_type": "",
},
"database": {
# Empty by default — the runtime expects SECOND_BRAIN_DATABASE_URL
# (or HERBYLAB_DATABASE_URL) in the environment. Fill this only if
# you prefer keeping the URL inside settings.toml.
"url": "",
},
"extractor": {
# "cli" runs `claude -p` under the Max OAuth subscription (default).
# "api" uses the Anthropic SDK and requires ANTHROPIC_API_KEY.
@ -51,19 +35,6 @@ _DEFAULTS: dict[str, Any] = {
"cli_binary": "claude",
"timeout_sec": 600,
},
"embeddings": {
# Best-effort embedding of extraction summaries into public.embeddings.
# If disabled, or if Ollama / the DB is unreachable, the pipeline
# logs a warning and keeps going — file/relational data stays the
# source of truth.
"enabled": True,
# Default targets the `ollama` service on the homelab docker network
# (same convention vault-mcp uses). For host-side runs, set
# OLLAMA_URL=http://127.0.0.1:11434 in the env instead.
"ollama_url": "http://ollama:11434",
"model": "nomic-embed-text",
"embed_field": "summary",
},
"scheduler": {
"window_start": "22:00",
"window_end": "06:00",
@ -78,13 +49,6 @@ _DEFAULTS: dict[str, Any] = {
DOMAINS = ["development", "content", "business", "homelab"]
def _merge(default: dict, override: dict) -> dict:
"""Shallow merge — override wins per top-level key."""
out = dict(default)
out.update(override or {})
return out
class Config:
"""Holds resolved configuration values."""
@ -93,6 +57,10 @@ class Config:
# --- 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()
@ -107,24 +75,11 @@ class Config:
self._raw.get("subtitles_dir", _DEFAULTS["subtitles_dir"])
).expanduser()
# --- transcription ---
@property
def whisper(self) -> dict[str, Any]:
"""Merged [whisper] block. Env vars in `transcribe.resolve_settings`
still win over these for the tower worker."""
return _merge(_DEFAULTS["whisper"], self._raw.get("whisper", {}))
# --- extraction ---
@property
def whisper_model(self) -> str:
"""Backwards-compat shortcut — older callers (and settings.toml
in the wild) reference whisper_model at the top level."""
return (
self._raw.get("whisper_model")
or self.whisper.get("whisper_model", "large-v3")
)
# --- extraction ---
return self._raw.get("whisper_model", _DEFAULTS["whisper_model"])
@property
def domains(self) -> list[str]:
@ -134,23 +89,17 @@ class Config:
def anthropic_api_key(self) -> str | None:
return os.getenv("ANTHROPIC_API_KEY")
# --- structured sections ---
@property
def database(self) -> dict[str, Any]:
return _merge(_DEFAULTS["database"], self._raw.get("database", {}))
# --- extractor ---
@property
def extractor(self) -> dict[str, Any]:
return _merge(_DEFAULTS["extractor"], self._raw.get("extractor", {}))
return {**_DEFAULTS["extractor"], **self._raw.get("extractor", {})}
@property
def embeddings(self) -> dict[str, Any]:
return _merge(_DEFAULTS["embeddings"], self._raw.get("embeddings", {}))
# --- scheduler ---
@property
def scheduler(self) -> dict[str, Any]:
return _merge(_DEFAULTS["scheduler"], self._raw.get("scheduler", {}))
return {**_DEFAULTS["scheduler"], **self._raw.get("scheduler", {})}
# --- prompts dir ---
@ -163,7 +112,7 @@ class Config:
def ensure_dirs(self) -> None:
"""Create runtime directories if they don't exist."""
for d in (self.media_dir, self.subtitles_dir):
for d in (self.db_path.parent, self.media_dir, self.subtitles_dir):
d.mkdir(parents=True, exist_ok=True)
@ -201,10 +150,4 @@ def load_config(path: Path | None = None) -> Config:
return _config_instance
def reset_config_singleton() -> None:
"""Test hook — clear the cached Config so a re-read picks up new settings."""
global _config_instance
_config_instance = None
__all__ = ["Config", "DOMAINS", "load_config", "reset_config_singleton"]
__all__ = ["Config", "DOMAINS", "load_config"]

2
src/second_brain/context/assembler.py
View File

@ -9,6 +9,8 @@ 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

117
src/second_brain/database.py
View File

@ -1,105 +1,43 @@
"""
Database engine and session management for second-brain.
Postgres + pgvector backend on the petalbrain instance.
Resolution order for the connection URL:
1. constructor arg passed to `get_database()`
2. env var `SECOND_BRAIN_DATABASE_URL`
3. env var `HERBYLAB_DATABASE_URL` (shared homelab convention)
4. `[database].url` in settings.toml
The URL is expected to use SQLAlchemy's `postgresql+psycopg://` driver
(psycopg v3). Plain `postgresql://` URLs are accepted and rewritten to use
psycopg; this matches the vault-mcp convention where one URL feeds both
Alembic (`+psycopg`) and raw psycopg (no prefix).
Connection mode: containerized (homelab-postgres:5432 on the homelab docker
network). For local-host dev runs, point at 127.0.0.1:5433 instead.
Ported and adapted from xtract/database.py.
Uses a singleton pattern: call get_database() everywhere.
"""
from __future__ import annotations
import os
from contextlib import contextmanager
from pathlib import Path
from typing import Generator
from sqlalchemy import create_engine, text
from sqlalchemy.engine import Engine
from sqlalchemy import create_engine
from sqlalchemy.orm import Session, sessionmaker
from second_brain.models import SCHEMA, Base
def _normalize_url(url: str) -> str:
"""Ensure the URL uses SQLAlchemy's psycopg-v3 driver."""
if url.startswith("postgresql+psycopg://"):
return url
if url.startswith("postgresql://"):
return url.replace("postgresql://", "postgresql+psycopg://", 1)
return url
def _resolve_url(explicit: str | None) -> str:
if explicit:
return _normalize_url(explicit)
for var in ("SECOND_BRAIN_DATABASE_URL", "HERBYLAB_DATABASE_URL"):
v = os.environ.get(var)
if v:
return _normalize_url(v)
# Fall back to settings.toml [database].url
try:
from second_brain.config import load_config
cfg = load_config()
url = cfg.database.get("url")
if url:
return _normalize_url(url)
except Exception:
pass
raise RuntimeError(
"No Postgres connection URL found. Set SECOND_BRAIN_DATABASE_URL "
"(or HERBYLAB_DATABASE_URL), or fill `[database].url` in "
"config/settings.toml."
)
from second_brain.models import Base
class Database:
"""Manages the SQLAlchemy engine and session factory for second-brain."""
"""Manages the SQLAlchemy engine and session factory."""
def __init__(self, url: str | None = None) -> None:
self.url = _resolve_url(url)
def __init__(self, db_path: Path) -> None:
self.db_path = db_path
self.db_path.parent.mkdir(parents=True, exist_ok=True)
# Pool sizing matches vault-mcp (min=1/max=10). No PgBouncer; a normal
# QueuePool is fine for the CLI + scheduler + web workers we run.
self.engine: Engine = create_engine(
self.url,
self.engine = create_engine(
f"sqlite:///{self.db_path}",
echo=False,
pool_size=int(os.environ.get("SECOND_BRAIN_DB_POOL_MIN", "1")),
max_overflow=int(os.environ.get("SECOND_BRAIN_DB_POOL_MAX", "9")),
pool_pre_ping=True,
future=True,
connect_args={"check_same_thread": False},
)
self.SessionLocal = sessionmaker(
autocommit=False,
autoflush=False,
bind=self.engine,
future=True,
)
def init_db(self) -> None:
"""Create the schema and tables if missing.
Schema creation is idempotent and only relevant for dev/test setups
where Alembic hasn't been applied yet. In production, prefer
`alembic upgrade head` — this method just defensively ensures the
schema exists before `create_all` issues unqualified DDL.
"""
with self.engine.begin() as conn:
conn.execute(text(f"CREATE SCHEMA IF NOT EXISTS {SCHEMA}"))
"""Create all tables (idempotent)."""
Base.metadata.create_all(bind=self.engine)
def get_session(self) -> Session:
@ -119,9 +57,6 @@ class Database:
finally:
sess.close()
def close(self) -> None:
self.engine.dispose()
# ---------------------------------------------------------------------------
# Singleton
@ -130,25 +65,17 @@ class Database:
_db_instance: Database | None = None
def get_database(url: str | None = None) -> Database:
"""Return the singleton Database, creating it lazily on first call.
Schema/tables are *not* auto-created here — run `alembic upgrade head`
once during setup. Tests may explicitly call `Database(url).init_db()`
against a scratch database.
"""
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:
_db_instance = Database(url)
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
def reset_database_singleton() -> None:
"""Test hook — drop the cached singleton so a new URL can take effect."""
global _db_instance
if _db_instance is not None:
_db_instance.close()
_db_instance = None
__all__ = ["Database", "get_database", "reset_database_singleton"]
__all__ = ["Database", "get_database"]

257
src/second_brain/embeddings/__init__.py
View File

@ -1,257 +0,0 @@
"""Embed second-brain extractions into the shared public.embeddings table.
This is a near-verbatim copy of the vault-mcp recipe (vault_mcp.core.embeddings),
adapted for our `source_schema='second_brain' / source_table='extractions'`
key tuple and our embed payload (the LLM-generated summary).
Behaviour:
- Reuses the shared `embedding_chunking` library for 512-token chunking with
64-token overlap.
- Calls Ollama's /api/embeddings with `num_ctx=8192` to embed nomic-embed-text
vectors (768-d).
- Writes vector literals as `%s::vector` to avoid taking a hard dependency on
pgvector's Python adapter.
- Delete-before-insert on the (schema, table, source_id, embedding_model)
tuple so a re-embed of an extraction that previously had N chunks but now
has M<N doesn't leave orphans.
Graceful degradation: any failure (no DB URL, no Ollama, DB error, empty
summary) logs at WARNING and returns None. The relational `extractions` row
and the file-based vault remain the source of truth — embeddings are an
index, not authoritative state.
"""
from __future__ import annotations
import atexit
import logging
import os
from typing import Any
import httpx
from embedding_chunking import Chunk, chunk_text
from psycopg_pool import ConnectionPool
logger = logging.getLogger(__name__)
SOURCE_SCHEMA = "second_brain"
SOURCE_TABLE = "extractions"
_pool: ConnectionPool | None = None
# ---------------------------------------------------------------------------
# Pool / URL plumbing
# ---------------------------------------------------------------------------
def _database_url() -> str | None:
"""Resolve a raw psycopg DSN (no SQLAlchemy `+psycopg` prefix)."""
for var in ("SECOND_BRAIN_DATABASE_URL", "HERBYLAB_DATABASE_URL"):
v = os.environ.get(var)
if v:
return v.replace("postgresql+psycopg://", "postgresql://", 1)
# Fallback: pull from settings.toml so a host-side `process` works even
# when the operator forgot to export the env var.
try:
from second_brain.config import load_config
url = load_config().database.get("url")
if url:
return url.replace("postgresql+psycopg://", "postgresql://", 1)
except Exception:
pass
return None
def _get_pool() -> ConnectionPool | None:
"""Lazy-init a small connection pool. Returns None if no DB URL is configured."""
global _pool
if _pool is not None:
return _pool
dsn = _database_url()
if not dsn:
return None
_pool = ConnectionPool(dsn, min_size=1, max_size=4, open=True, timeout=10)
# Make sure the pool's worker threads get a clean shutdown when the
# interpreter exits; without this psycopg_pool prints
# "couldn't stop thread pool-1-worker-N within 5.0 seconds" on every
# `second-brain process` run.
atexit.register(close_pool)
return _pool
def close_pool() -> None:
"""Close the embedding-side pool. Safe to call multiple times."""
global _pool
if _pool is not None:
_pool.close()
_pool = None
# ---------------------------------------------------------------------------
# Embedding model client
# ---------------------------------------------------------------------------
def _ollama_url() -> str:
"""Env var takes precedence over settings.toml so containerized runs can
point at a different Ollama instance without rebuilding the image."""
env = os.environ.get("OLLAMA_URL")
if env:
return env
try:
from second_brain.config import load_config
return load_config().embeddings.get("ollama_url", "http://ollama:11434")
except Exception:
return "http://ollama:11434"
def _embedding_model() -> str:
env = os.environ.get("EMBEDDING_MODEL")
if env:
return env
try:
from second_brain.config import load_config
return load_config().embeddings.get("model", "nomic-embed-text")
except Exception:
return "nomic-embed-text"
def embed_text(text: str) -> list[float]:
"""Call Ollama's embeddings endpoint. Returns the model-native vector."""
resp = httpx.post(
f"{_ollama_url()}/api/embeddings",
json={
"model": _embedding_model(),
"prompt": text,
# nomic-embed-text supports 8192-token context. Ollama defaults
# the runner to 2048, which 500s on long summaries.
"options": {"num_ctx": 8192},
},
timeout=120.0,
)
resp.raise_for_status()
data = resp.json()
vec = data.get("embedding")
if not vec or not isinstance(vec, list):
raise RuntimeError(
f"Ollama returned no embedding for model {_embedding_model()}"
)
return vec
def _vector_literal(vec: list[float]) -> str:
"""pgvector accepts text-literal vectors of form '[v1,v2,...]'."""
return "[" + ",".join(repr(float(v)) for v in vec) + "]"
# ---------------------------------------------------------------------------
# Public API
# ---------------------------------------------------------------------------
def embed_extraction(extraction_id: int, summary: str | None) -> int | None:
"""Embed an extraction summary into public.embeddings.
Args:
extraction_id: PK of the row in second_brain.extractions.
summary: the LLM-generated summary string; we embed this and
nothing else for the current round (key_points / claims
are not embedded yet).
Returns the number of chunk rows written, or None on skip/failure.
Best-effort: failures are logged, not raised.
"""
if not summary or not summary.strip():
logger.debug("embed_extraction: skip — extraction %s has no summary", extraction_id)
return None
pool = _get_pool()
if pool is None:
logger.info(
"embed_extraction: skip — no SECOND_BRAIN_DATABASE_URL configured"
)
return None
model = _embedding_model()
try:
chunks: list[Chunk] = chunk_text(summary)
chunk_embeddings: list[tuple[Chunk, list[float]]] = [
(c, embed_text(c.text)) for c in chunks
]
except Exception as exc:
logger.warning(
"embed_extraction: chunk/embed failed for extraction %s: %s",
extraction_id,
exc,
)
return None
try:
with pool.connection() as conn:
n = _upsert_embeddings(
conn, extraction_id, chunk_embeddings, model
)
conn.commit()
logger.info(
"embed_extraction: extraction=%s chunks=%d model=%s",
extraction_id,
n,
model,
)
return n
except Exception as exc:
logger.warning(
"embed_extraction: DB write failed for extraction %s: %s",
extraction_id,
exc,
)
return None
def _upsert_embeddings(
conn: Any,
extraction_id: int,
chunk_embeddings: list[tuple[Chunk, list[float]]],
model: str,
) -> int:
"""Replace all public.embeddings rows for this (extraction, model) pair."""
with conn.cursor() as cur:
cur.execute(
"""
DELETE FROM public.embeddings
WHERE source_schema = %s
AND source_table = %s
AND source_id = %s
AND embedding_model = %s
""",
(SOURCE_SCHEMA, SOURCE_TABLE, extraction_id, model),
)
for chunk, vec in chunk_embeddings:
cur.execute(
"""
INSERT INTO public.embeddings
(source_schema, source_table, source_id, chunk_index,
chunk_text, chunk_token_count, embedding, embedding_model)
VALUES (%s, %s, %s, %s, %s, %s, %s::vector, %s)
""",
(
SOURCE_SCHEMA,
SOURCE_TABLE,
extraction_id,
chunk.index,
chunk.text,
chunk.token_count,
_vector_literal(vec),
model,
),
)
return len(chunk_embeddings)
__all__ = ["embed_extraction", "embed_text", "close_pool"]

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

@ -18,7 +18,7 @@ import os
from typing import Optional
from second_brain.extractor.schema import ExtractionResult, LLMExtraction, SourceMeta
from second_brain.llm.claude_cli import ClaudeCLI, DEFAULT_MODEL
from second_brain.llm.claude_cli import ClaudeCLI, ClaudeCLIError, DEFAULT_MODEL
from second_brain.models import Source

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

@ -7,9 +7,10 @@ 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
from pydantic import BaseModel, Field, HttpUrl
class LLMExtraction(BaseModel):

140
src/second_brain/main.py
View File

@ -11,6 +11,8 @@ Commands:
from __future__ import annotations
import sys
from pathlib import Path
from typing import Optional
from urllib.parse import urlparse
@ -52,7 +54,7 @@ def add(url: str, domain: str, focus: Optional[str], title: Optional[str]) -> No
"""Queue a source URL for processing."""
config = load_config()
config.ensure_dirs()
db = get_database()
db = get_database(config.db_path)
source_type = SourceType.ARTICLE if _is_article(url) else SourceType.VIDEO
@ -104,13 +106,8 @@ def process(
skip_transcribe: bool,
skip_extract: bool,
) -> None:
"""Run the dev-side pipeline: article pulls + extract/embed.
Video pull + transcribe have moved to the tower-side
`transcribe-worker` daemon (faster-whisper on the 3080); this command
is article-only on the pull side, but still picks up TRANSCRIBED rows
of any source_type for the extract/embed step.
"""
"""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
@ -118,8 +115,9 @@ def process(
config = load_config()
config.ensure_dirs()
db = get_database()
db = get_database(config.db_path)
yt_adapter = YouTubeAdapter(config)
art_adapter = ArticleAdapter(config)
assembler = ContextAssembler(config)
@ -139,61 +137,52 @@ def process(
click.echo("[process] Extraction step will be skipped.")
skip_extract = True
# Dev side: pull only ARTICLE sources (videos belong to the tower
# worker) — but pick up TRANSCRIBED of any type for extract/embed.
with db.session() as sess:
pull_ids = []
if not skip_pull:
pull_ids = [
row[0]
for row in sess.query(Source.id)
.filter(
Source.status == SourceStatus.PENDING,
Source.source_type == SourceType.ARTICLE,
query = sess.query(Source.id).filter(
Source.status.in_([
SourceStatus.PENDING,
SourceStatus.PULLED,
SourceStatus.TRANSCRIBED,
])
)
.all()
]
extract_ids = [
row[0]
for row in sess.query(Source.id)
.filter(Source.status == SourceStatus.TRANSCRIBED)
.all()
]
# Keep order stable (pulls first so freshly-transcribed articles
# roll into extract in the same invocation) and de-duplicate.
source_ids: list[int] = []
for sid in pull_ids + extract_ids:
if sid not in source_ids:
source_ids.append(sid)
if limit:
source_ids = source_ids[:limit]
query = query.limit(limit)
source_ids = [row[0] for row in query.all()]
if not source_ids:
click.echo("[process] No sources to process.")
return
click.echo(f"[process] Processing {len(source_ids)} source(s)…")
if skip_transcribe:
click.echo("[process] (--skip-transcribe is now a no-op — transcription "
"runs on the tower worker.)")
for source_id in source_ids:
with db.session() as sess:
source = sess.get(Source, source_id)
click.echo(f"\n [{source.id}] {source.url[:80]}")
# Pull step — articles only on the dev side.
if (
not skip_pull
and source.status == SourceStatus.PENDING
and source.source_type == SourceType.ARTICLE
):
click.echo(" → pull (article)")
# 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")
@ -210,32 +199,19 @@ def process(
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)
ext_row = existing_ext
else:
ext_row = Extraction(
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(ext_row)
sess.add(extraction)
source.status = SourceStatus.ANALYZED
source.updated_at = utcnow()
# Flush so the new extraction gets its PK before we kick
# off embedding (we key public.embeddings rows by it).
sess.flush()
extraction_id = ext_row.id
extraction_summary = ext_row.summary
click.echo(" ✓ analyzed")
if config.embeddings.get("enabled", True):
from second_brain.embeddings import embed_extraction
n = embed_extraction(extraction_id, extraction_summary)
if n is not None:
click.echo(f" ✓ embedded ({n} chunk(s))")
except Exception as exc:
source.error_message = str(exc)
source.status = SourceStatus.FAILED
@ -302,52 +278,6 @@ def schedule(dry_run: bool) -> None:
scheduler.run(dry_run=dry_run)
# ---------------------------------------------------------------------------
# transcribe-worker (tower-side daemon)
# ---------------------------------------------------------------------------
@cli.command("transcribe-worker")
@click.option("--once", is_flag=True, default=False,
help="Process at most one source then exit (useful for ad-hoc runs / tests).")
@click.option("--poll-interval", default=None, type=int,
help="Seconds between empty-queue polls (default 15).")
@click.option("--worker-id", default=None,
help="Identifier stamped into sources.claimed_by (default tower:<hostname>).")
def transcribe_worker(
once: bool,
poll_interval: Optional[int],
worker_id: Optional[str],
) -> None:
"""Long-running tower-side worker: claim → yt-dlp pull → faster-whisper.
Reads pipeline_settings each iteration so the dashboard's enable
toggle / active-hours window / max-items-per-run / max-video-length
all take effect within one poll cycle.
"""
import logging
from second_brain.scheduler.transcribe_worker import (
DEFAULT_POLL_INTERVAL_SECONDS,
TranscribeWorker,
)
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s %(levelname)s %(name)s: %(message)s",
)
config = load_config()
config.ensure_dirs()
worker = TranscribeWorker(
config,
worker_id=worker_id,
poll_interval=poll_interval or DEFAULT_POLL_INTERVAL_SECONDS,
)
processed = worker.run(once=once)
click.echo(f"[transcribe-worker] exiting; processed={processed}")
# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------

124
src/second_brain/models.py
View File

@ -1,10 +1,6 @@
"""
SQLAlchemy ORM models for second-brain.
All tables live in the Postgres `second_brain` schema (in the petalbrain DB).
PK style is int autoincrement (SERIAL in Postgres); timestamps are naive UTC
(`timestamp without time zone`) per the locked design decisions in CLAUDE.md.
Tables:
- Source — a queued/processed URL (video or article)
- Extraction — structured LLM output for a source
@ -14,36 +10,28 @@ Tables:
from __future__ import annotations
import enum
from datetime import datetime, time, timezone
from datetime import datetime, timezone
from typing import Optional
from sqlalchemy import (
JSON,
Boolean,
DateTime,
Enum,
ForeignKey,
Integer,
MetaData,
String,
Text,
Time,
)
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, relationship
def utcnow() -> datetime:
"""Naive UTC timestamp. Replaces `datetime.utcnow()` (deprecated in 3.12)."""
return datetime.now(timezone.utc).replace(tzinfo=None)
SCHEMA = "second_brain"
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. All tables live in the `second_brain` schema."""
metadata = MetaData(schema=SCHEMA)
"""Shared declarative base."""
# ---------------------------------------------------------------------------
@ -66,22 +54,6 @@ class SourceStatus(enum.Enum):
FAILED = "failed"
# Native Postgres enum types, scoped to the second_brain schema so they
# don't collide with anything else in the cluster.
_SourceTypeEnum = Enum(
SourceType,
name="source_type",
schema=SCHEMA,
values_callable=lambda et: [e.value for e in et],
)
_SourceStatusEnum = Enum(
SourceStatus,
name="source_status",
schema=SCHEMA,
values_callable=lambda et: [e.value for e in et],
)
# ---------------------------------------------------------------------------
# Models
# ---------------------------------------------------------------------------
@ -97,7 +69,7 @@ class Source(Base):
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(
_SourceTypeEnum, nullable=False, default=SourceType.VIDEO
Enum(SourceType), nullable=False, default=SourceType.VIDEO
)
# Domain and focus
@ -106,7 +78,7 @@ class Source(Base):
# Pipeline status
status: Mapped[SourceStatus] = mapped_column(
_SourceStatusEnum, nullable=False, default=SourceStatus.PENDING
Enum(SourceStatus), nullable=False, default=SourceStatus.PENDING
)
# File paths (set after pull/transcribe steps)
@ -124,10 +96,6 @@ class Source(Base):
updated_at: Mapped[Optional[datetime]] = mapped_column(DateTime, nullable=True)
error_message: Mapped[Optional[str]] = mapped_column(Text, nullable=True)
# Claim mechanism — see `second_brain.claim` for the queue dance.
claimed_by: Mapped[Optional[str]] = mapped_column(String(100), nullable=True)
claimed_at: Mapped[Optional[datetime]] = mapped_column(DateTime, nullable=True)
# Relationships
extraction: Mapped[Optional["Extraction"]] = relationship(
"Extraction", back_populates="source", uselist=False, cascade="all, delete-orphan"
@ -147,9 +115,7 @@ class Extraction(Base):
id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
source_id: Mapped[int] = mapped_column(
ForeignKey(f"{SCHEMA}.sources.id", ondelete="CASCADE"),
nullable=False,
unique=True,
ForeignKey("sources.id", ondelete="CASCADE"), nullable=False, unique=True
)
# Core fields (mirrors the extraction schema contract)
@ -199,72 +165,12 @@ class WikiPage(Base):
return f"WikiPage(id={self.id}, vault_path={self.vault_path!r})"
class PipelineSettings(Base):
"""Single-row config the web dashboard edits and workers read.
The transcription_* fields persist now and will be honored by the
tower-side transcribe worker once it exists; the extraction_* fields
are already wired into the dev-side scheduler. The row is pinned to
`id=1` by a DB-level CHECK so upserts-by-PK keep the table singleton.
"""
__tablename__ = "pipeline_settings"
id: Mapped[int] = mapped_column(Integer, primary_key=True, default=1)
# Transcription (consumed by the tower-side worker — not yet built).
transcription_enabled: Mapped[bool] = mapped_column(
Boolean, nullable=False, default=True
)
transcription_active_hours_start: Mapped[Optional[time]] = mapped_column(
Time, nullable=True
)
transcription_active_hours_end: Mapped[Optional[time]] = mapped_column(
Time, nullable=True
)
transcription_max_concurrent_gpu_jobs: Mapped[int] = mapped_column(
Integer, nullable=False, default=1
)
transcription_max_video_length_seconds: Mapped[Optional[int]] = mapped_column(
Integer, nullable=True
)
transcription_max_items_per_run: Mapped[int] = mapped_column(
Integer, nullable=False, default=10
)
# Extraction (consumed by the dev-side scheduler today).
extraction_enabled: Mapped[bool] = mapped_column(
Boolean, nullable=False, default=True
)
extraction_active_hours_start: Mapped[Optional[time]] = mapped_column(
Time, nullable=True
)
extraction_active_hours_end: Mapped[Optional[time]] = mapped_column(
Time, nullable=True
)
extraction_max_items_per_run: Mapped[int] = mapped_column(
Integer, nullable=False, default=20
)
updated_at: Mapped[datetime] = mapped_column(
DateTime, default=utcnow, onupdate=utcnow
)
def __repr__(self) -> str:
return (
f"PipelineSettings(transcription_enabled={self.transcription_enabled}, "
f"extraction_enabled={self.extraction_enabled})"
)
__all__ = [
"Base",
"SCHEMA",
"Source",
"SourceStatus",
"SourceType",
"Extraction",
"PipelineSettings",
"WikiPage",
"utcnow",
]

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

@ -21,11 +21,11 @@ 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, utcnow
from second_brain.settings_store import get_settings, in_active_window
class Scheduler:
@ -52,38 +52,6 @@ class Scheduler:
from second_brain.models import Extraction
db = get_database()
# Pull the DB-side runtime knobs once at the top of the run. The
# dashboard edits these; settings.toml is the static fallback. The
# values are snapshotted for the duration of this run so a mid-run
# toggle doesn't half-apply (consistent with the way max_calls is
# tracked locally).
with db.session() as sess:
settings = get_settings(sess)
extraction_enabled = settings.extraction_enabled
db_window = (
settings.extraction_active_hours_start,
settings.extraction_active_hours_end,
)
max_items = int(settings.extraction_max_items_per_run or 0)
if not extraction_enabled:
print("[Scheduler] extraction_enabled=false in pipeline_settings. Skipping run.")
return
# DB window (if set) overrides the static settings.toml window. If
# only one side is set, treat the other as unbounded (no constraint).
if db_window[0] is not None or db_window[1] is not None:
self.window_start = db_window[0] or dtime(0, 0)
self.window_end = db_window[1] or dtime(23, 59)
if not in_active_window(datetime.now().time(), db_window[0], db_window[1]):
print(
f"[Scheduler] outside extraction active window "
f"{_fmt_opt(db_window[0])}–{_fmt_opt(db_window[1])}. Skipping run."
)
return
assembler = ContextAssembler(self.config)
ex_cfg = self.config.extractor
engine = ExtractionEngine(
@ -107,14 +75,10 @@ class Scheduler:
print(
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 max_items={max_items or '∞'}"
f"gap {self.min_gap_seconds}s"
)
while self._in_window():
# DB-side hard cap on items per run (0 = unlimited).
if max_items and processed >= max_items:
print(f"[Scheduler] reached extraction_max_items_per_run={max_items}. Done.")
break
# Reset hourly buckets
elapsed = time.time() - hour_start
if elapsed >= 3600:
@ -163,27 +127,18 @@ class Scheduler:
existing.raw_response = getattr(result, "_raw_response", None)
existing.input_tokens = getattr(result, "_input_tokens", None)
existing.output_tokens = getattr(result, "_output_tokens", None)
ext_row = existing
else:
ext_row = Extraction(
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,
)
sess.add(ext_row)
))
source.status = SourceStatus.ANALYZED
source.updated_at = utcnow()
# Flush so the embedding step can key on the new PK.
sess.flush()
if self.config.embeddings.get("enabled", True):
from second_brain.embeddings import embed_extraction
embed_extraction(ext_row.id, ext_row.summary)
calls_this_hour += 1
used = (getattr(result, "_input_tokens", 0) or 0) + (
getattr(result, "_output_tokens", 0) or 0
@ -228,8 +183,4 @@ def _fmt(t: dtime) -> str:
return t.strftime("%H:%M")
def _fmt_opt(t: dtime | None) -> str:
return t.strftime("%H:%M") if t is not None else "—"
__all__ = ["Scheduler"]

312
src/second_brain/scheduler/transcribe_worker.py
View File

@ -1,312 +0,0 @@
"""Tower-side transcribe worker — long-running poll loop.
Runs on the EndeavourOS tower (RTX 3080) as a systemd service. Reaches
petalbrain over WireGuard at db.wg.herbylab.dev:5432.
Loop body:
1. Read pipeline_settings.
- transcription_enabled = false → sleep, retry.
- outside transcription_active_hours → sleep, retry.
- transcription_max_concurrent_gpu_jobs is consulted (default 1 =
this worker is the single in-flight job; future scale-out can fan
a per-job lock here).
2. Claim the oldest unclaimed VIDEO source in {PENDING, PULLED} via
`SELECT ... FOR UPDATE SKIP LOCKED`.
3. If PENDING → yt-dlp pull → PULLED (claim retained).
4. If PULLED → faster-whisper transcribe → TRANSCRIBED (claim released).
5. Honor transcription_max_video_length_seconds: skip & mark FAILED
with a clear error_message if the metadata's duration exceeds it.
Graceful degradation:
- DB unreachable → log at WARNING, sleep, retry. Never crash-loop.
- Whisper failure on one file → mark that source FAILED, release claim,
continue with the next one.
- SIGTERM (systemd stop) → finish the current iteration, exit cleanly.
"""
from __future__ import annotations
import logging
import os
import signal
import socket
import time
from datetime import datetime
from pathlib import Path
from second_brain.claim import (
claim_next_source,
reap_stale_claims,
release_claim,
)
from second_brain.config import Config
from second_brain.database import get_database
from second_brain.models import Source, SourceStatus, SourceType, utcnow
from second_brain.settings_store import get_settings, in_active_window
logger = logging.getLogger(__name__)
# Default loop pace — short so a dropped-in video starts transcribing
# right away. The worker also re-reads pipeline_settings every iteration
# so a dashboard toggle takes effect within ~POLL_INTERVAL_SECONDS.
DEFAULT_POLL_INTERVAL_SECONDS = 15
DEFAULT_DB_BACKOFF_SECONDS = 30
class TranscribeWorker:
"""The tower's transcribe poll loop."""
def __init__(
self,
config: Config,
*,
worker_id: str | None = None,
poll_interval: int = DEFAULT_POLL_INTERVAL_SECONDS,
db_backoff: int = DEFAULT_DB_BACKOFF_SECONDS,
) -> None:
self.config = config
self.worker_id = (
worker_id
or os.environ.get("SECOND_BRAIN_WORKER_ID")
or f"tower:{socket.gethostname()}"
)
self.poll_interval = poll_interval
self.db_backoff = db_backoff
self._stop = False
# SIGTERM/SIGINT → clean shutdown after the current iteration.
signal.signal(signal.SIGTERM, self._on_signal)
signal.signal(signal.SIGINT, self._on_signal)
def _on_signal(self, signum, frame): # noqa: ARG002 — frame required by signal API
logger.info("worker received signal %s — stopping after current iteration", signum)
self._stop = True
# ------------------------------------------------------------------
# Top-level run loop
# ------------------------------------------------------------------
def run(self, *, once: bool = False, max_iterations: int | None = None) -> int:
"""Drive the loop. Returns the count of sources processed in this run.
`once=True` runs a single iteration regardless of queue depth —
useful for ad-hoc CLI invocation. `max_iterations` caps loops so
tests don't run forever.
"""
processed = 0
iterations = 0
startup_reaped = False
while not self._stop:
iterations += 1
if max_iterations is not None and iterations > max_iterations:
break
try:
db = get_database()
except Exception as exc:
logger.warning("DB not configured: %s — sleeping %ds", exc, self.db_backoff)
if once:
return processed
self._sleep(self.db_backoff)
continue
try:
with db.session() as sess:
if not startup_reaped:
reaped = reap_stale_claims(sess)
if reaped:
logger.info("reaped %d stale claim(s) on startup", reaped)
startup_reaped = True
settings = get_settings(sess)
enabled = settings.transcription_enabled
window = (
settings.transcription_active_hours_start,
settings.transcription_active_hours_end,
)
max_items = int(settings.transcription_max_items_per_run or 0)
max_video_length = settings.transcription_max_video_length_seconds
except Exception as exc:
logger.warning("DB unreachable: %s — sleeping %ds", exc, self.db_backoff)
if once:
return processed
self._sleep(self.db_backoff)
continue
if not enabled:
logger.info("transcription_enabled=false — sleeping")
if once:
return processed
self._sleep(self.poll_interval)
continue
if not in_active_window(datetime.now().time(), window[0], window[1]):
logger.info("outside transcription active window — sleeping")
if once:
return processed
self._sleep(self.poll_interval)
continue
if max_items and processed >= max_items:
logger.info("reached transcription_max_items_per_run=%d — exiting", max_items)
break
handled = self._process_one(max_video_length)
if handled is None:
# Nothing to claim — sleep and re-poll.
if once:
return processed
self._sleep(self.poll_interval)
continue
processed += 1
if once:
return processed
logger.info("worker stopping (processed=%d, iterations=%d)", processed, iterations)
return processed
# ------------------------------------------------------------------
# Per-source step
# ------------------------------------------------------------------
def _process_one(self, max_video_length: int | None) -> int | None:
"""Try to claim and advance one video source. Returns the source id
on success, or None if nothing was claimable.
"""
db = get_database()
# Step 1: claim. Short transaction, FOR UPDATE SKIP LOCKED inside.
with db.session() as sess:
source_id = claim_next_source(
sess,
worker_id=self.worker_id,
statuses=[SourceStatus.PENDING, SourceStatus.PULLED],
source_type=SourceType.VIDEO,
)
if source_id is None:
return None
# Step 2: do the work. Fresh session because we crossed a commit.
try:
with db.session() as sess:
source = sess.get(Source, source_id)
if source is None:
return None
logger.info("[%s] starting %s id=%d url=%s",
self.worker_id, source.status.value, source.id, source.url)
self._advance(source, max_video_length, sess)
except Exception as exc:
logger.exception("unexpected error on source %s: %s", source_id, exc)
with db.session() as sess:
src = sess.get(Source, source_id)
if src is not None:
src.status = SourceStatus.FAILED
src.error_message = f"transcribe worker crashed: {exc}"
src.updated_at = utcnow()
release_claim(sess, source_id)
return source_id
def _advance(
self,
source: Source,
max_video_length: int | None,
sess,
) -> None:
"""Advance a single claimed source through pull → transcribe."""
from second_brain.adapters.youtube import YouTubeAdapter
from second_brain.transcribe import Transcriber, resolve_settings, write_srt
# Length guard — relies on yt-dlp's metadata for PENDING sources;
# for already-PULLED sources we trust whatever the prior pull
# recorded on the row.
if source.status == SourceStatus.PENDING:
yt = YouTubeAdapter(self.config)
try:
meta = yt.downloader.extract_metadata(source.url)
except Exception as exc:
logger.warning("metadata fetch failed for source %s: %s", source.id, exc)
meta = {}
duration = meta.get("duration_seconds")
if (
max_video_length is not None
and duration is not None
and duration > max_video_length
):
source.error_message = (
f"video length {duration}s exceeds "
f"transcription_max_video_length_seconds={max_video_length}"
)
source.status = SourceStatus.FAILED
source.updated_at = utcnow()
release_claim(sess, source.id)
logger.info("skipping source %s — too long (%ss)", source.id, duration)
return
# Pre-fill the metadata so we don't re-hit yt-dlp during pull.
if not source.title:
source.title = meta.get("title")
if not source.author:
source.author = meta.get("author")
if not source.duration_seconds:
source.duration_seconds = duration
if not source.published_at:
source.published_at = meta.get("published_at")
ok = yt.pull(source)
if not ok:
source.updated_at = utcnow()
release_claim(sess, source.id)
return
# Transcribe.
if source.status == SourceStatus.PULLED:
if not source.media_path:
source.error_message = "no media_path after pull"
source.status = SourceStatus.FAILED
source.updated_at = utcnow()
release_claim(sess, source.id)
return
wh = resolve_settings(self.config.whisper)
transcriber = Transcriber(
model=wh["model"],
device=wh["device"],
compute_type=wh["compute_type"],
)
media_file = Path(source.media_path)
srt_path = self.config.subtitles_dir / f"{media_file.stem}.srt"
try:
text, segments = transcriber.transcribe(media_file)
except Exception as exc:
source.error_message = f"transcribe failed: {exc}"
source.status = SourceStatus.FAILED
source.updated_at = utcnow()
release_claim(sess, source.id)
logger.exception("transcribe failed for source %s", source.id)
return
write_srt(segments, srt_path)
source.transcript_path = str(srt_path)
source.transcript_text = text
source.status = SourceStatus.TRANSCRIBED
source.updated_at = utcnow()
release_claim(sess, source.id)
logger.info("transcribed source %s — %d chars", source.id, len(text))
# ------------------------------------------------------------------
# Helpers
# ------------------------------------------------------------------
def _sleep(self, seconds: int) -> None:
"""Sleep that wakes early on SIGTERM."""
end = time.time() + seconds
while time.time() < end and not self._stop:
time.sleep(1)
__all__ = ["TranscribeWorker", "DEFAULT_POLL_INTERVAL_SECONDS"]

121
src/second_brain/settings_store.py
View File

@ -1,121 +0,0 @@
"""Helpers for the singleton `pipeline_settings` row.
The web dashboard edits this row; workers read it at the start of each
run to decide whether to do anything and how much. DB values override
the static defaults in `settings.toml`.
All functions take a SQLAlchemy `Session` so the caller owns the
transaction boundary — matches the rest of the codebase, which keeps
session lifecycle inside `with db.session()` blocks.
"""
from __future__ import annotations
from datetime import time
from typing import Any
from sqlalchemy.orm import Session
from second_brain.models import PipelineSettings, utcnow
# Field names allowed in `update_settings`. Anything else is rejected so a
# malicious form post can't toggle id/updated_at/etc.
_EDITABLE_FIELDS: tuple[str, ...] = (
"transcription_enabled",
"transcription_active_hours_start",
"transcription_active_hours_end",
"transcription_max_concurrent_gpu_jobs",
"transcription_max_video_length_seconds",
"transcription_max_items_per_run",
"extraction_enabled",
"extraction_active_hours_start",
"extraction_active_hours_end",
"extraction_max_items_per_run",
)
def get_settings(sess: Session) -> PipelineSettings:
"""Return the singleton settings row, creating it on first call.
The v2 migration seeds the row with all DB-side defaults, so this
"create on miss" path only kicks in when somebody bootstrapped the
schema without applying the migration (e.g. an unconfigured test
using `Base.metadata.create_all`).
"""
row = sess.get(PipelineSettings, 1)
if row is None:
row = PipelineSettings(id=1)
sess.add(row)
sess.flush()
return row
def update_settings(sess: Session, **fields: Any) -> PipelineSettings:
"""Apply a partial update to the singleton row.
Unknown keys raise `KeyError`. Empty-string values for nullable fields
are normalised to None so the HTML form's blank inputs (active hours,
video-length cap) round-trip as NULL rather than as the strings "".
"""
for k in fields:
if k not in _EDITABLE_FIELDS:
raise KeyError(f"Unknown pipeline_settings field: {k!r}")
row = get_settings(sess)
for k, v in fields.items():
if v == "":
v = None
setattr(row, k, v)
row.updated_at = utcnow()
return row
def parse_time_or_none(raw: str | None) -> time | None:
"""Parse an HTML <input type=time> value (HH:MM) into a `time`.
Empty / missing values map to None — `update_settings` will store
NULL, which workers interpret as "no window, always active".
"""
if raw is None or raw == "":
return None
parts = raw.split(":")
if len(parts) < 2:
raise ValueError(f"Could not parse time: {raw!r}")
h, m = int(parts[0]), int(parts[1])
return time(h, m)
def parse_int_or_none(raw: str | None) -> int | None:
if raw is None or raw == "":
return None
return int(raw)
def in_active_window(now_t: time, start: time | None, end: time | None) -> bool:
"""Return True if `now_t` lies in the [start, end] window.
Either bound being None means "no constraint on that side", which
matches the dashboard's "leave blank = always active" semantics.
Handles overnight windows (start > end) by treating them as the
complement that crosses midnight.
"""
if start is None and end is None:
return True
if start is None:
return now_t <= end # type: ignore[operator]
if end is None:
return now_t >= start
if start <= end:
return start <= now_t <= end
# crosses midnight
return now_t >= start or now_t <= end
__all__ = [
"get_settings",
"update_settings",
"parse_time_or_none",
"parse_int_or_none",
"in_active_window",
]

171
src/second_brain/transcribe.py
View File

@ -1,171 +0,0 @@
"""Audio/video transcription via faster-whisper (CTranslate2 backend).
Lazy-imports `faster_whisper` so the dev side — which doesn't transcribe
anything — doesn't need the heavyweight CUDA wheels installed.
Config keys (read from `[whisper]` in settings.toml, env-overrides win):
- `whisper_model` : the model name. Default "large-v3".
- `whisper_device` : "cuda" | "cpu" | "auto". Default "cuda".
- `whisper_compute_type` : "float16" | "int8" | "int8_float16" | …
Default float16 on cuda, int8 on cpu.
Outputs an SRT file next to `config.subtitles_dir/{stem}.srt` plus a
plain-text transcript string. The pipeline writes the text into
`sources.transcript_text` and the path into `sources.transcript_path`,
matching the previous torch-whisper contract so downstream extract/embed
doesn't need to change.
"""
from __future__ import annotations
import logging
import os
from pathlib import Path
from typing import Iterable, Optional
logger = logging.getLogger(__name__)
# ---------------------------------------------------------------------------
# Config resolution
# ---------------------------------------------------------------------------
def _resolve(cfg: dict | None, env_key: str, cfg_key: str, default: str) -> str:
env = os.environ.get(env_key)
if env:
return env
if cfg is not None:
v = cfg.get(cfg_key)
if v:
return v
return default
def resolve_settings(config_block: dict | None = None) -> dict[str, str]:
"""Resolve whisper runtime settings from env > config block > defaults.
Pulls model/device/compute_type and picks a sane compute_type per device
when none was specified. Returns plain strings so the caller can stash
them in logs / pass straight to faster-whisper.
"""
model = _resolve(config_block, "WHISPER_MODEL", "whisper_model", "large-v3")
device = _resolve(config_block, "WHISPER_DEVICE", "whisper_device", "cuda")
explicit_compute = None
if config_block is not None:
explicit_compute = config_block.get("whisper_compute_type")
explicit_compute = os.environ.get("WHISPER_COMPUTE_TYPE", explicit_compute)
if explicit_compute:
compute_type = explicit_compute
elif device == "cpu":
compute_type = "int8"
else:
compute_type = "float16"
return {"model": model, "device": device, "compute_type": compute_type}
# ---------------------------------------------------------------------------
# Transcriber
# ---------------------------------------------------------------------------
class Transcriber:
"""faster-whisper wrapper. Model is loaded lazily on first transcribe."""
def __init__(
self,
model: str = "large-v3",
device: str = "cuda",
compute_type: str = "float16",
) -> None:
self.model = model
self.device = device
self.compute_type = compute_type
self._impl = None
def _load(self) -> None:
if self._impl is not None:
return
try:
from faster_whisper import WhisperModel # type: ignore[import-not-found]
except ImportError as exc:
raise RuntimeError(
"faster-whisper is not installed. On the tower run: "
"uv sync --extra tower"
) from exc
logger.info(
"loading faster-whisper model=%s device=%s compute_type=%s",
self.model,
self.device,
self.compute_type,
)
self._impl = WhisperModel(
self.model, device=self.device, compute_type=self.compute_type
)
def transcribe(
self,
media_file: Path,
*,
language: Optional[str] = None,
beam_size: int = 5,
) -> tuple[str, list[dict]]:
"""Transcribe `media_file`. Returns (full_text, segments).
Each segment dict has start/end (float seconds) and text (str) —
same shape the SRT writer needs.
"""
self._load()
assert self._impl is not None # for the type-checker
segments_iter, info = self._impl.transcribe(
str(media_file),
language=language,
beam_size=beam_size,
)
logger.info(
"transcribe start: %s language=%s duration=%.1fs",
media_file.name,
getattr(info, "language", "?"),
getattr(info, "duration", 0.0) or 0.0,
)
segments: list[dict] = []
text_parts: list[str] = []
for seg in segments_iter:
t = (seg.text or "").strip()
segments.append({"start": float(seg.start), "end": float(seg.end), "text": t})
if t:
text_parts.append(t)
return " ".join(text_parts), segments
# ---------------------------------------------------------------------------
# SRT writer
# ---------------------------------------------------------------------------
def write_srt(segments: Iterable[dict], output: Path) -> None:
"""Write a list of {start,end,text} dicts as SRT into `output`."""
output.parent.mkdir(parents=True, exist_ok=True)
with open(output, "w", encoding="utf-8") as fh:
for i, seg in enumerate(segments, start=1):
start = _fmt_timestamp(seg["start"])
end = _fmt_timestamp(seg["end"])
fh.write(f"{i}\n{start} --> {end}\n{seg['text']}\n\n")
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}"
__all__ = ["Transcriber", "resolve_settings", "write_srt"]

185
src/second_brain/web/app.py
View File

@ -18,17 +18,11 @@ from fastapi import FastAPI, Form, HTTPException, Request
from fastapi.responses import HTMLResponse
from fastapi.templating import Jinja2Templates
from pydantic import BaseModel
from sqlalchemy import func
from sqlalchemy.orm import Session
from second_brain.config import DOMAINS
from second_brain.config import DOMAINS, load_config
from second_brain.database import get_database
from second_brain.models import Extraction, Source, SourceStatus
from second_brain.settings_store import (
get_settings,
parse_int_or_none,
parse_time_or_none,
update_settings,
)
from second_brain.models import Extraction, Source, SourceStatus, SourceType
# ---------------------------------------------------------------------------
# App setup
@ -302,176 +296,3 @@ def _source_to_response(s: Source) -> SourceResponse:
ingested_at=s.ingested_at.isoformat() if s.ingested_at else "",
has_extraction=s.extraction is not None,
)
# ---------------------------------------------------------------------------
# Dashboard — pipeline observability
# ---------------------------------------------------------------------------
@app.get("/dashboard", response_class=HTMLResponse)
async def dashboard(request: Request):
"""Overview: status counts, recent activity, and a settings snapshot."""
db = get_database()
with db.session() as sess:
# Counts by status. Materialise inside the session so the keys stay
# plain str → int and templates don't try to touch a closed session.
rows = (
sess.query(Source.status, func.count(Source.id))
.group_by(Source.status)
.all()
)
counts = {s.value: 0 for s in SourceStatus}
for status, n in rows:
counts[status.value] = int(n)
total = sum(counts.values())
recent_rows = (
sess.query(Source)
.order_by(Source.ingested_at.desc())
.limit(10)
.all()
)
recent = [_source_to_dict(s) for s in recent_rows]
settings_row = get_settings(sess)
settings_summary = _settings_to_dict(settings_row)
return templates.TemplateResponse(
request,
"dashboard.html",
{
"counts": counts,
"total": total,
"recent": recent,
"statuses_in_order": [s.value for s in SourceStatus],
"settings": settings_summary,
},
)
# ---------------------------------------------------------------------------
# Settings — view + HTMX save
# ---------------------------------------------------------------------------
@app.get("/settings", response_class=HTMLResponse)
async def settings_view(request: Request):
db = get_database()
with db.session() as sess:
row = get_settings(sess)
ctx = _settings_to_dict(row)
return templates.TemplateResponse(
request,
"settings.html",
{"settings": ctx, "flash": None, "error": None},
)
@app.post("/settings/save", response_class=HTMLResponse)
async def settings_save(
request: Request,
transcription_enabled: Optional[str] = Form(None),
transcription_active_hours_start: Optional[str] = Form(None),
transcription_active_hours_end: Optional[str] = Form(None),
transcription_max_concurrent_gpu_jobs: Optional[str] = Form(None),
transcription_max_video_length_seconds: Optional[str] = Form(None),
transcription_max_items_per_run: Optional[str] = Form(None),
extraction_enabled: Optional[str] = Form(None),
extraction_active_hours_start: Optional[str] = Form(None),
extraction_active_hours_end: Optional[str] = Form(None),
extraction_max_items_per_run: Optional[str] = Form(None),
):
"""HTMX save endpoint — returns the rendered settings card with a flash."""
db = get_database()
error: Optional[str] = None
flash: Optional[str] = None
try:
fields = {
# Checkboxes only show up in the form payload when checked. The
# template emits a hidden "_present" companion so we can tell
# "unchecked" apart from "field missing because of a partial post".
"transcription_enabled": transcription_enabled == "on",
"transcription_active_hours_start": parse_time_or_none(
transcription_active_hours_start
),
"transcription_active_hours_end": parse_time_or_none(
transcription_active_hours_end
),
"transcription_max_concurrent_gpu_jobs": int(
transcription_max_concurrent_gpu_jobs or "1"
),
"transcription_max_video_length_seconds": parse_int_or_none(
transcription_max_video_length_seconds
),
"transcription_max_items_per_run": int(
transcription_max_items_per_run or "10"
),
"extraction_enabled": extraction_enabled == "on",
"extraction_active_hours_start": parse_time_or_none(
extraction_active_hours_start
),
"extraction_active_hours_end": parse_time_or_none(
extraction_active_hours_end
),
"extraction_max_items_per_run": int(extraction_max_items_per_run or "20"),
}
# Cheap inline validation — keep counts non-negative and gpu jobs >= 1.
if fields["transcription_max_concurrent_gpu_jobs"] < 1:
raise ValueError("transcription_max_concurrent_gpu_jobs must be ≥ 1")
if fields["transcription_max_items_per_run"] < 0:
raise ValueError("transcription_max_items_per_run must be ≥ 0")
if fields["extraction_max_items_per_run"] < 0:
raise ValueError("extraction_max_items_per_run must be ≥ 0")
if (
fields["transcription_max_video_length_seconds"] is not None
and fields["transcription_max_video_length_seconds"] < 0
):
raise ValueError("transcription_max_video_length_seconds must be ≥ 0")
with db.session() as sess:
row = update_settings(sess, **fields)
ctx = _settings_to_dict(row)
flash = "Saved."
except (ValueError, KeyError) as exc:
error = str(exc)
# Re-render the form using whatever the user submitted so they don't
# lose in-progress edits on a validation bounce.
with db.session() as sess:
ctx = _settings_to_dict(get_settings(sess))
return templates.TemplateResponse(
request,
"_settings_card.html",
{"settings": ctx, "flash": flash, "error": error},
)
# ---------------------------------------------------------------------------
# Settings serialiser
# ---------------------------------------------------------------------------
def _settings_to_dict(row) -> dict:
"""Render the settings row as a template-friendly dict (no ORM bleed)."""
def t(v):
return v.strftime("%H:%M") if v is not None else ""
return {
"transcription_enabled": row.transcription_enabled,
"transcription_active_hours_start": t(row.transcription_active_hours_start),
"transcription_active_hours_end": t(row.transcription_active_hours_end),
"transcription_max_concurrent_gpu_jobs": row.transcription_max_concurrent_gpu_jobs,
"transcription_max_video_length_seconds": row.transcription_max_video_length_seconds,
"transcription_max_items_per_run": row.transcription_max_items_per_run,
"extraction_enabled": row.extraction_enabled,
"extraction_active_hours_start": t(row.extraction_active_hours_start),
"extraction_active_hours_end": t(row.extraction_active_hours_end),
"extraction_max_items_per_run": row.extraction_max_items_per_run,
"updated_at": row.updated_at.strftime("%Y-%m-%d %H:%M:%S")
if row.updated_at
else "",
}

130
src/second_brain/web/templates/_settings_card.html
View File

@ -1,130 +0,0 @@
{# HTMX-replaceable inner card. Posted to /settings/save; swaps itself in place. #}
<div id="settings-card" class="bg-white rounded-lg shadow-sm p-6">
{% if flash %}
<div class="mb-4 px-3 py-2 rounded bg-green-50 border border-green-200 text-green-700 text-sm">
✓ {{ flash }}
</div>
{% endif %}
{% if error %}
<div class="mb-4 px-3 py-2 rounded bg-red-50 border border-red-200 text-red-700 text-sm">
✗ {{ error }}
</div>
{% endif %}
<form hx-post="/settings/save"
hx-target="#settings-card"
hx-swap="outerHTML"
class="space-y-8">
<!-- Transcription -->
<section>
<div class="flex items-center justify-between mb-4">
<h3 class="text-base font-semibold text-gray-800">Transcription</h3>
<p class="text-xs text-gray-400">consumed by tower-side worker (not yet built)</p>
</div>
<div class="grid md:grid-cols-2 gap-x-6 gap-y-4">
<!-- enabled -->
<label class="flex items-center gap-3 cursor-pointer select-none">
<span class="relative inline-flex items-center">
<input type="checkbox" name="transcription_enabled"
{% if settings.transcription_enabled %}checked{% endif %}
class="peer sr-only">
<span class="w-10 h-5 bg-gray-300 rounded-full peer-checked:bg-indigo-600 transition-colors"></span>
<span class="absolute left-0.5 top-0.5 w-4 h-4 bg-white rounded-full transition-transform peer-checked:translate-x-5"></span>
</span>
<span class="text-sm font-medium text-gray-700">Enabled</span>
</label>
<div class="md:col-span-1"></div>
<div>
<label class="block text-xs font-medium text-gray-600 mb-1">Active hours — start</label>
<input type="time" name="transcription_active_hours_start"
value="{{ settings.transcription_active_hours_start }}"
class="text-sm border border-gray-300 rounded px-3 py-1.5 focus:outline-none focus:ring-2 focus:ring-indigo-400 w-full">
<p class="text-xs text-gray-400 mt-1">blank = always active</p>
</div>
<div>
<label class="block text-xs font-medium text-gray-600 mb-1">Active hours — end</label>
<input type="time" name="transcription_active_hours_end"
value="{{ settings.transcription_active_hours_end }}"
class="text-sm border border-gray-300 rounded px-3 py-1.5 focus:outline-none focus:ring-2 focus:ring-indigo-400 w-full">
</div>
<div>
<label class="block text-xs font-medium text-gray-600 mb-1">Max concurrent GPU jobs</label>
<input type="number" min="1" name="transcription_max_concurrent_gpu_jobs"
value="{{ settings.transcription_max_concurrent_gpu_jobs }}"
class="text-sm border border-gray-300 rounded px-3 py-1.5 focus:outline-none focus:ring-2 focus:ring-indigo-400 w-full">
</div>
<div>
<label class="block text-xs font-medium text-gray-600 mb-1">Max items per run</label>
<input type="number" min="0" name="transcription_max_items_per_run"
value="{{ settings.transcription_max_items_per_run }}"
class="text-sm border border-gray-300 rounded px-3 py-1.5 focus:outline-none focus:ring-2 focus:ring-indigo-400 w-full">
</div>
<div class="md:col-span-2">
<label class="block text-xs font-medium text-gray-600 mb-1">Max video length (seconds)</label>
<input type="number" min="0" name="transcription_max_video_length_seconds"
value="{{ settings.transcription_max_video_length_seconds if settings.transcription_max_video_length_seconds is not none else '' }}"
placeholder="blank = no cap"
class="text-sm border border-gray-300 rounded px-3 py-1.5 focus:outline-none focus:ring-2 focus:ring-indigo-400 w-full max-w-xs">
</div>
</div>
</section>
<!-- Extraction -->
<section>
<div class="flex items-center justify-between mb-4">
<h3 class="text-base font-semibold text-gray-800">Extraction</h3>
<p class="text-xs text-gray-400">consumed by the dev-side scheduler</p>
</div>
<div class="grid md:grid-cols-2 gap-x-6 gap-y-4">
<label class="flex items-center gap-3 cursor-pointer select-none">
<span class="relative inline-flex items-center">
<input type="checkbox" name="extraction_enabled"
{% if settings.extraction_enabled %}checked{% endif %}
class="peer sr-only">
<span class="w-10 h-5 bg-gray-300 rounded-full peer-checked:bg-indigo-600 transition-colors"></span>
<span class="absolute left-0.5 top-0.5 w-4 h-4 bg-white rounded-full transition-transform peer-checked:translate-x-5"></span>
</span>
<span class="text-sm font-medium text-gray-700">Enabled</span>
</label>
<div class="md:col-span-1"></div>
<div>
<label class="block text-xs font-medium text-gray-600 mb-1">Active hours — start</label>
<input type="time" name="extraction_active_hours_start"
value="{{ settings.extraction_active_hours_start }}"
class="text-sm border border-gray-300 rounded px-3 py-1.5 focus:outline-none focus:ring-2 focus:ring-indigo-400 w-full">
<p class="text-xs text-gray-400 mt-1">blank = always active</p>
</div>
<div>
<label class="block text-xs font-medium text-gray-600 mb-1">Active hours — end</label>
<input type="time" name="extraction_active_hours_end"
value="{{ settings.extraction_active_hours_end }}"
class="text-sm border border-gray-300 rounded px-3 py-1.5 focus:outline-none focus:ring-2 focus:ring-indigo-400 w-full">
</div>
<div>
<label class="block text-xs font-medium text-gray-600 mb-1">Max items per run</label>
<input type="number" min="0" name="extraction_max_items_per_run"
value="{{ settings.extraction_max_items_per_run }}"
class="text-sm border border-gray-300 rounded px-3 py-1.5 focus:outline-none focus:ring-2 focus:ring-indigo-400 w-full">
</div>
</div>
</section>
<div class="flex items-center justify-between pt-2 border-t border-gray-100">
<p class="text-xs text-gray-400">last saved {{ settings.updated_at }}</p>
<button type="submit"
class="px-5 py-2 bg-indigo-600 text-white text-sm font-medium rounded hover:bg-indigo-700 transition-colors">
Save
</button>
</div>
</form>
</div>

2
src/second_brain/web/templates/base.html
View File

@ -16,10 +16,8 @@
<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="/dashboard" class="hover:text-indigo-200 transition-colors">Dashboard</a>
<a href="/" class="hover:text-indigo-200 transition-colors">Sources</a>
<a href="/queue" class="hover:text-indigo-200 transition-colors">Queue</a>
<a href="/settings" class="hover:text-indigo-200 transition-colors">Settings</a>
</nav>
</div>
</header>

158
src/second_brain/web/templates/dashboard.html
View File

@ -1,158 +0,0 @@
{% extends "base.html" %}
{% block title %}Dashboard — second-brain{% endblock %}
{% block content %}
{% 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'
} %}
<div class="mb-6 flex items-center justify-between">
<h2 class="text-2xl font-semibold text-gray-800">Dashboard</h2>
<span class="text-sm text-gray-500">total sources: {{ total }}</span>
</div>
<!-- Pipeline overview -->
<section class="bg-white rounded-lg shadow-sm p-5 mb-6">
<h3 class="text-sm font-semibold uppercase tracking-wide text-gray-500 mb-3">
Pipeline
</h3>
<div class="grid grid-cols-2 md:grid-cols-4 lg:grid-cols-7 gap-3">
{% for status in statuses_in_order %}
<a href="/?status={{ status }}"
class="block rounded-lg border border-gray-200 hover:border-indigo-400 hover:shadow-sm transition-all px-3 py-3">
<div class="text-2xl font-bold text-gray-900 leading-none">{{ counts.get(status, 0) }}</div>
<div class="mt-1.5">
<span class="text-xs px-2 py-0.5 rounded-full font-medium
{{ status_colors.get(status, 'bg-gray-100 text-gray-700') }}">
{{ status }}
</span>
</div>
</a>
{% endfor %}
</div>
</section>
<!-- Settings snapshot + Recent activity, side by side on wide screens -->
<div class="grid lg:grid-cols-3 gap-6">
<!-- Settings snapshot -->
<section class="bg-white rounded-lg shadow-sm p-5 lg:col-span-1">
<div class="flex items-center justify-between mb-3">
<h3 class="text-sm font-semibold uppercase tracking-wide text-gray-500">
Settings
</h3>
<a href="/settings" class="text-xs text-indigo-600 hover:text-indigo-700 font-medium">
edit →
</a>
</div>
<dl class="text-sm space-y-2">
<div class="flex justify-between items-center">
<dt class="text-gray-600">Transcription</dt>
<dd>
{% if settings.transcription_enabled %}
<span class="text-xs px-2 py-0.5 rounded-full bg-green-100 text-green-700 font-medium">enabled</span>
{% else %}
<span class="text-xs px-2 py-0.5 rounded-full bg-gray-200 text-gray-600 font-medium">paused</span>
{% endif %}
</dd>
</div>
<div class="flex justify-between items-center">
<dt class="text-gray-600"> active window</dt>
<dd class="text-gray-800 font-mono text-xs">
{% if settings.transcription_active_hours_start or settings.transcription_active_hours_end %}
{{ settings.transcription_active_hours_start or "—" }} – {{ settings.transcription_active_hours_end or "—" }}
{% else %}
always
{% endif %}
</dd>
</div>
<div class="flex justify-between items-center">
<dt class="text-gray-600"> GPU jobs</dt>
<dd class="text-gray-800 font-mono text-xs">{{ settings.transcription_max_concurrent_gpu_jobs }}</dd>
</div>
<div class="flex justify-between items-center">
<dt class="text-gray-600"> max items/run</dt>
<dd class="text-gray-800 font-mono text-xs">{{ settings.transcription_max_items_per_run }}</dd>
</div>
<div class="border-t border-gray-100 pt-2 mt-2"></div>
<div class="flex justify-between items-center">
<dt class="text-gray-600">Extraction</dt>
<dd>
{% if settings.extraction_enabled %}
<span class="text-xs px-2 py-0.5 rounded-full bg-green-100 text-green-700 font-medium">enabled</span>
{% else %}
<span class="text-xs px-2 py-0.5 rounded-full bg-gray-200 text-gray-600 font-medium">paused</span>
{% endif %}
</dd>
</div>
<div class="flex justify-between items-center">
<dt class="text-gray-600"> active window</dt>
<dd class="text-gray-800 font-mono text-xs">
{% if settings.extraction_active_hours_start or settings.extraction_active_hours_end %}
{{ settings.extraction_active_hours_start or "—" }} – {{ settings.extraction_active_hours_end or "—" }}
{% else %}
always
{% endif %}
</dd>
</div>
<div class="flex justify-between items-center">
<dt class="text-gray-600"> max items/run</dt>
<dd class="text-gray-800 font-mono text-xs">{{ settings.extraction_max_items_per_run }}</dd>
</div>
</dl>
<p class="text-xs text-gray-400 mt-4">
updated {{ settings.updated_at }}
</p>
</section>
<!-- Recent activity -->
<section class="bg-white rounded-lg shadow-sm p-5 lg:col-span-2">
<h3 class="text-sm font-semibold uppercase tracking-wide text-gray-500 mb-3">
Recent activity
</h3>
{% if recent %}
<div class="divide-y divide-gray-100">
{% for source in recent %}
<a href="/sources/{{ source.id }}"
class="flex items-start justify-between gap-3 py-2.5 hover:bg-gray-50 -mx-2 px-2 rounded transition-colors">
<div class="min-w-0 flex-1">
<p class="text-sm font-medium text-gray-900 truncate">{{ source.title }}</p>
<p class="text-xs text-gray-400 truncate">{{ source.url }}</p>
</div>
<div class="flex flex-col items-end gap-1 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>
</a>
{% endfor %}
</div>
{% else %}
<p class="text-sm text-gray-400 py-8 text-center">No sources yet.</p>
{% endif %}
</section>
</div>
{% endblock %}

14
src/second_brain/web/templates/settings.html
View File

@ -1,14 +0,0 @@
{% extends "base.html" %}
{% block title %}Settings — second-brain{% endblock %}
{% block content %}
<div class="mb-6">
<h2 class="text-2xl font-semibold text-gray-800">Settings</h2>
<p class="text-sm text-gray-500 mt-1">
Workers read these at the start of each run. DB values override <code class="bg-gray-100 px-1 rounded text-xs">settings.toml</code>.
</p>
</div>
{% include "_settings_card.html" %}
{% endblock %}

107
tests/test_claim.py
View File

@ -1,107 +0,0 @@
"""Live-DB test: claim_next_source is race-safe against concurrent callers.
Skips when no DB URL is configured. This is the integration-level guard
for the SELECT ... FOR UPDATE SKIP LOCKED dance — unit-testing the SQL
without Postgres would prove nothing.
"""
from __future__ import annotations
import os
import threading
import uuid
import psycopg
import pytest
def _db_reachable() -> bool:
url = os.environ.get("SECOND_BRAIN_DATABASE_URL") or os.environ.get(
"HERBYLAB_DATABASE_URL"
)
if not url:
return False
raw = url.replace("postgresql+psycopg://", "postgresql://", 1)
try:
with psycopg.connect(raw, connect_timeout=2):
return True
except Exception:
return False
pytestmark = pytest.mark.skipif(
not _db_reachable(),
reason="claim race test needs SECOND_BRAIN_DATABASE_URL pointed at a real petalbrain",
)
def test_two_workers_never_grab_the_same_row():
from second_brain.claim import claim_next_source, release_claim
from second_brain.database import Database, reset_database_singleton
from second_brain.models import Source, SourceStatus, SourceType, utcnow
reset_database_singleton()
db = Database()
# Seed a single TRANSCRIBED-stage candidate.
unique = uuid.uuid4().hex[:8]
url = f"https://example.test/claim-race/{unique}"
with db.session() as sess:
src = Source(
url=url,
title=f"claim-race-{unique}",
domain="development",
source_type=SourceType.VIDEO,
status=SourceStatus.PULLED,
ingested_at=utcnow(),
)
sess.add(src)
sess.flush()
source_id = src.id
results: list[int | None] = []
lock = threading.Lock()
def worker(name: str):
with db.session() as sess:
claimed = claim_next_source(
sess,
worker_id=name,
statuses=[SourceStatus.PENDING, SourceStatus.PULLED],
source_type=SourceType.VIDEO,
)
with lock:
results.append(claimed)
t1 = threading.Thread(target=worker, args=("worker-A",))
t2 = threading.Thread(target=worker, args=("worker-B",))
t1.start()
t2.start()
t1.join()
t2.join()
try:
# Exactly one worker should have grabbed the row; the other gets None.
# (SKIP LOCKED guarantees this even if both transactions overlap.)
non_none = [r for r in results if r is not None]
assert non_none == [source_id], (
f"expected exactly one claimant of {source_id}, got results={results}"
)
# Confirm claim columns landed.
with db.session() as sess:
row = sess.get(Source, source_id)
assert row.claimed_by in {"worker-A", "worker-B"}
assert row.claimed_at is not None
# Releasing the claim should null both columns.
with db.session() as sess:
release_claim(sess, source_id)
with db.session() as sess:
row = sess.get(Source, source_id)
assert row.claimed_by is None
assert row.claimed_at is None
finally:
with db.session() as sess:
sess.query(Source).filter(Source.id == source_id).delete()
reset_database_singleton()

157
tests/test_smoke_embedding.py
View File

@ -1,157 +0,0 @@
"""Smoke test for the Postgres + embedding pipeline.
Runs only when SECOND_BRAIN_DATABASE_URL (or HERBYLAB_DATABASE_URL) is set
*and* an Ollama instance with nomic-embed-text is reachable. Otherwise the
test is skipped — there is no in-memory fixture for pgvector, and we
explicitly want the smoke check to exercise the real wire format.
It covers:
1. SQLAlchemy can write a Source + Extraction into the `second_brain`
schema.
2. The embeddings module chunks the summary, calls Ollama, and writes
vector rows into public.embeddings keyed correctly.
3. Re-running embedding is idempotent (delete-before-insert wins).
"""
from __future__ import annotations
import os
import socket
import uuid
import httpx
import psycopg
import pytest
def _ollama_reachable() -> bool:
url = os.environ.get("OLLAMA_URL", "http://127.0.0.1:11434")
try:
r = httpx.get(f"{url}/api/tags", timeout=2.0)
if r.status_code != 200:
return False
models = [m["name"] for m in r.json().get("models", [])]
return any(m.startswith("nomic-embed-text") for m in models)
except (httpx.HTTPError, socket.error):
return False
def _db_reachable() -> bool:
url = os.environ.get("SECOND_BRAIN_DATABASE_URL") or os.environ.get(
"HERBYLAB_DATABASE_URL"
)
if not url:
return False
raw = url.replace("postgresql+psycopg://", "postgresql://", 1)
try:
with psycopg.connect(raw, connect_timeout=2) as conn: # noqa: F841
return True
except Exception:
return False
pytestmark = pytest.mark.skipif(
not (_db_reachable() and _ollama_reachable()),
reason="Smoke test needs SECOND_BRAIN_DATABASE_URL + reachable Ollama (nomic-embed-text)",
)
def test_extraction_round_trip_with_embedding():
from second_brain.database import Database, reset_database_singleton
from second_brain.embeddings import close_pool, embed_extraction
from second_brain.models import (
Extraction,
Source,
SourceStatus,
SourceType,
utcnow,
)
reset_database_singleton()
db = Database()
# Unique URL so re-runs don't collide on the unique constraint.
unique = uuid.uuid4().hex[:8]
url = f"https://example.test/smoke/{unique}"
summary_text = (
"Smoke test extraction summary. "
"The second-brain pipeline embeds this string into Postgres. "
"Re-running the embed should be idempotent — delete-before-insert "
"keys on (source_schema, source_table, source_id, embedding_model)."
)
with db.session() as sess:
src = Source(
url=url,
title=f"smoke-test-{unique}",
domain="development",
source_type=SourceType.ARTICLE,
status=SourceStatus.ANALYZED,
ingested_at=utcnow(),
)
sess.add(src)
sess.flush()
ext = Extraction(source_id=src.id, summary=summary_text)
sess.add(ext)
sess.flush()
extraction_id = ext.id
source_id = src.id
# Read the configured model so the test doesn't break if the default
# is changed in settings.toml or via EMBEDDING_MODEL.
from second_brain.config import load_config
embedding_model = os.environ.get(
"EMBEDDING_MODEL",
load_config().embeddings.get("model", "nomic-embed-text"),
)
try:
n1 = embed_extraction(extraction_id, summary_text)
assert n1 is not None and n1 >= 1, f"first embed should write rows, got {n1}"
# Idempotency: re-embed should produce the same chunk count and not
# accumulate orphan rows.
n2 = embed_extraction(extraction_id, summary_text)
assert n2 == n1, f"re-embed should match first count: {n1} vs {n2}"
raw_url = (
os.environ.get("SECOND_BRAIN_DATABASE_URL")
or os.environ.get("HERBYLAB_DATABASE_URL")
).replace("postgresql+psycopg://", "postgresql://", 1)
with psycopg.connect(raw_url) as conn, conn.cursor() as cur:
cur.execute(
"""
SELECT COUNT(*)
FROM public.embeddings
WHERE source_schema = 'second_brain'
AND source_table = 'extractions'
AND source_id = %s
AND embedding_model = %s
""",
(extraction_id, embedding_model),
)
(count,) = cur.fetchone()
assert count == n1, f"public.embeddings should hold {n1} rows, has {count}"
finally:
# Tidy up so the smoke test stays leak-free.
with db.session() as sess:
sess.query(Source).filter(Source.id == source_id).delete()
raw_url = (
os.environ.get("SECOND_BRAIN_DATABASE_URL")
or os.environ.get("HERBYLAB_DATABASE_URL")
).replace("postgresql+psycopg://", "postgresql://", 1)
with psycopg.connect(raw_url) as conn, conn.cursor() as cur:
cur.execute(
"""
DELETE FROM public.embeddings
WHERE source_schema = 'second_brain'
AND source_table = 'extractions'
AND source_id = %s
""",
(extraction_id,),
)
conn.commit()
close_pool()
reset_database_singleton()

137
tests/test_transcribe.py
View File

@ -1,137 +0,0 @@
"""Unit + integration coverage for second_brain.transcribe.
Unit-level: resolve_settings + write_srt are pure-Python and run anywhere.
Integration: an end-to-end CPU/int8 transcribe of a synthesized audio
clip — auto-skipped when ffmpeg or faster-whisper is missing (the dev
side doesn't install faster-whisper by default). Only the *tower* with
GPU large-v3 can validate the production hot path; this test proves the
wiring + the SRT contract + the lazy-import path on cheap CPU.
"""
from __future__ import annotations
import pytest
# ---------------------------------------------------------------------------
# Pure helpers
# ---------------------------------------------------------------------------
def test_resolve_settings_picks_int8_for_cpu_by_default():
from second_brain.transcribe import resolve_settings
out = resolve_settings({"whisper_device": "cpu"})
assert out["device"] == "cpu"
assert out["compute_type"] == "int8"
assert out["model"] == "large-v3"
def test_resolve_settings_picks_float16_for_cuda_by_default():
from second_brain.transcribe import resolve_settings
out = resolve_settings({"whisper_device": "cuda"})
assert out["device"] == "cuda"
assert out["compute_type"] == "float16"
def test_resolve_settings_env_overrides_block(monkeypatch):
from second_brain.transcribe import resolve_settings
monkeypatch.setenv("WHISPER_MODEL", "small")
monkeypatch.setenv("WHISPER_DEVICE", "cpu")
monkeypatch.setenv("WHISPER_COMPUTE_TYPE", "int8_float16")
out = resolve_settings({"whisper_device": "cuda"})
assert out["model"] == "small"
assert out["device"] == "cpu"
assert out["compute_type"] == "int8_float16"
def test_write_srt_roundtrip(tmp_path):
from second_brain.transcribe import write_srt
segs = [
{"start": 0.0, "end": 1.5, "text": "hello"},
{"start": 1.5, "end": 3.25, "text": "world"},
]
out = tmp_path / "x.srt"
write_srt(segs, out)
body = out.read_text()
assert "1\n00:00:00,000 --> 00:00:01,500\nhello" in body
assert "2\n00:00:01,500 --> 00:00:03,250\nworld" in body
# ---------------------------------------------------------------------------
# Integration: synthesized audio → CPU/int8 faster-whisper
# ---------------------------------------------------------------------------
def _have_faster_whisper() -> bool:
try:
import faster_whisper # type: ignore[import-not-found] # noqa: F401
except ImportError:
return False
return True
def _have_numpy() -> bool:
try:
import numpy # noqa: F401
except ImportError:
return False
return True
@pytest.mark.skipif(
not (_have_faster_whisper() and _have_numpy()),
reason="needs faster-whisper (uv sync --extra tower) + numpy to exercise CPU path",
)
def test_cpu_transcribe_smoke(tmp_path):
"""Hand faster-whisper a synthesized audio array on CPU/int8 with `tiny`.
Pure-tone input gives empty / minimal recognition — the point is the
wiring: model loads, the segments iterator drains, an SRT is writable.
Using `tiny` to keep this under ~30s coldcache. Skipped on dev where
faster-whisper isn't installed; only the tower's `uv sync --extra
tower` brings it in.
Bypassing ffmpeg by passing a numpy array — faster-whisper.transcribe
accepts (path | file-like | np.ndarray). This exercises the same code
path the worker uses, only with zero external binaries.
"""
import numpy as np
from second_brain.transcribe import write_srt
# Build 2s of 440Hz mono at 16kHz, the model's native sample rate.
sr = 16000
duration = 2.0
t = np.arange(0, int(sr * duration)) / sr
audio = (0.2 * np.sin(2 * np.pi * 440 * t)).astype(np.float32)
# Load `tiny` to keep this cheap; we're not asserting transcription
# quality, just that the pipeline executes and returns the documented
# shape.
from faster_whisper import WhisperModel # type: ignore[import-not-found]
model = WhisperModel("tiny", device="cpu", compute_type="int8")
segments_iter, _ = model.transcribe(audio, language="en", beam_size=1)
# Materialize to mirror what Transcriber.transcribe does internally.
segments: list[dict] = []
text_parts: list[str] = []
for seg in segments_iter:
s = (seg.text or "").strip()
segments.append({"start": float(seg.start), "end": float(seg.end), "text": s})
if s:
text_parts.append(s)
text = " ".join(text_parts)
assert isinstance(text, str)
assert isinstance(segments, list)
srt = tmp_path / "out.srt"
write_srt(segments, srt)
assert srt.exists()

919
uv.lock generated
View File

File diff suppressed because it is too large Load Diff