pyr0ball 997eb6143e feat: Snipe MVP v0.1 — eBay trust scorer with faceted filter UI

2026-03-25 13:09:49 -07:00

31 KiB

Raw Blame History

circuitforge-core Extraction Implementation Plan

For agentic workers: REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (- [ ]) syntax for tracking.

Goal: Extract the shared scaffold from Peregrine into a standalone circuitforge-core Python package, update Peregrine to depend on it, and leave Peregrine's behaviour unchanged.

Architecture: New private repo circuitforge-core at /Library/Development/CircuitForge/circuitforge-core/. Contains: db base connection, LLM router, tier system, and config loader — the minimum core needed for both Peregrine and Snipe. Vision module, wizard framework, and pipeline module are stubbed (vision and wizard are net-new; pipeline is extracted but task-runner is deferred). Peregrine's requirements.txt gets a -e ../circuitforge-core entry; all internal imports updated.

Tech Stack: Python 3.11+, SQLite (stdlib), pytest, pyproject.toml (core only — Peregrine stays on requirements.txt)

File Map

New files (circuitforge-core repo)

File	Responsibility
`circuitforge_core/__init__.py`	Package version export
`circuitforge_core/db/__init__.py`	Re-exports `get_connection`
`circuitforge_core/db/base.py`	`get_connection()` — SQLite/SQLCipher connection factory (extracted from `peregrine/scripts/db.py`)
`circuitforge_core/db/migrations.py`	`run_migrations(conn, migrations_dir)` — simple sequential migration runner
`circuitforge_core/llm/__init__.py`	Re-exports `LLMRouter`
`circuitforge_core/llm/router.py`	`LLMRouter` class (extracted from `peregrine/scripts/llm_router.py`)
`circuitforge_core/vision/__init__.py`	Re-exports `VisionRouter`
`circuitforge_core/vision/router.py`	`VisionRouter` stub — raises `NotImplementedError` until v0.2
`circuitforge_core/wizard/__init__.py`	Re-exports `BaseWizard`
`circuitforge_core/wizard/base.py`	`BaseWizard` stub — raises `NotImplementedError` until first product wires it
`circuitforge_core/pipeline/__init__.py`	Re-exports `StagingDB`
`circuitforge_core/pipeline/staging.py`	`StagingDB` stub — interface for SQLite staging queue; raises `NotImplementedError`
`circuitforge_core/tiers/__init__.py`	Re-exports `can_use`, `TIERS`, `BYOK_UNLOCKABLE`, `LOCAL_VISION_UNLOCKABLE`
`circuitforge_core/tiers/tiers.py`	Generalised tier system (extracted from `peregrine/app/wizard/tiers.py`, product-specific feature keys removed)
`circuitforge_core/config/__init__.py`	Re-exports `require_env`, `load_env`
`circuitforge_core/config/settings.py`	`require_env(key)`, `load_env(path)` — env validation helpers
`pyproject.toml`	Package metadata, no dependencies beyond stdlib + pyyaml + requests + openai
`tests/test_db.py`	Tests for `get_connection` and migration runner
`tests/test_tiers.py`	Tests for `can_use`, BYOK unlock, LOCAL_VISION unlock
`tests/test_llm_router.py`	Tests for LLMRouter fallback chain (mock backends)
`tests/test_config.py`	Tests for `require_env` missing/present

Modified files (Peregrine repo)

File	Change
`requirements.txt`	Add `-e ../circuitforge-core`
`scripts/db.py`	Replace `get_connection` body with `from circuitforge_core.db import get_connection; ...` re-export
`scripts/llm_router.py`	Replace `LLMRouter` class body with import from `circuitforge_core.llm`
`app/wizard/tiers.py`	Replace tier/BYOK logic with imports from `circuitforge_core.tiers`; keep Peregrine-specific `FEATURES` dict

Task 1: Scaffold circuitforge-core repo

Files:

Create: circuitforge-core/pyproject.toml
Create: circuitforge-core/circuitforge_core/__init__.py
Create: circuitforge-core/.gitignore
Create: circuitforge-core/README.md
Step 1: Create repo directory and init git

mkdir -p /Library/Development/CircuitForge/circuitforge-core
cd /Library/Development/CircuitForge/circuitforge-core
git init

Step 2: Write pyproject.toml

# /Library/Development/CircuitForge/circuitforge-core/pyproject.toml
[build-system]
requires = ["setuptools>=68"]
build-backend = "setuptools.build_meta"

[project]
name = "circuitforge-core"
version = "0.1.0"
description = "Shared scaffold for CircuitForge products"
requires-python = ">=3.11"
dependencies = [
    "pyyaml>=6.0",
    "requests>=2.31",
    "openai>=1.0",
]

[tool.setuptools.packages.find]
where = ["."]
include = ["circuitforge_core*"]

[tool.pytest.ini_options]
testpaths = ["tests"]

Step 3: Write package init.py

# circuitforge-core/circuitforge_core/__init__.py
__version__ = "0.1.0"

Step 4: Write .gitignore

__pycache__/
*.pyc
.env
*.egg-info/
dist/
.pytest_cache/

Step 5: Install editable and verify import

cd /Library/Development/CircuitForge/circuitforge-core
conda run -n job-seeker pip install -e .
conda run -n job-seeker python -c "import circuitforge_core; print(circuitforge_core.__version__)"

Expected: 0.1.0

Step 6: Create tests/init.py

mkdir -p /Library/Development/CircuitForge/circuitforge-core/tests
touch /Library/Development/CircuitForge/circuitforge-core/tests/__init__.py

Step 7: Commit

git add .
git commit -m "feat: scaffold circuitforge-core package"

Task 2: Extract db base connection

Files:

Create: circuitforge-core/circuitforge_core/db/__init__.py
Create: circuitforge-core/circuitforge_core/db/base.py
Create: circuitforge-core/circuitforge_core/db/migrations.py
Create: circuitforge-core/tests/__init__.py
Create: circuitforge-core/tests/test_db.py
Step 1: Write failing tests

# circuitforge-core/tests/test_db.py
import sqlite3
import tempfile
from pathlib import Path
import pytest
from circuitforge_core.db import get_connection, run_migrations


def test_get_connection_returns_sqlite_connection(tmp_path):
    db = tmp_path / "test.db"
    conn = get_connection(db)
    assert isinstance(conn, sqlite3.Connection)
    conn.close()


def test_get_connection_creates_file(tmp_path):
    db = tmp_path / "test.db"
    assert not db.exists()
    conn = get_connection(db)
    conn.close()
    assert db.exists()


def test_run_migrations_applies_sql_files(tmp_path):
    db = tmp_path / "test.db"
    migrations_dir = tmp_path / "migrations"
    migrations_dir.mkdir()
    (migrations_dir / "001_create_foo.sql").write_text(
        "CREATE TABLE foo (id INTEGER PRIMARY KEY, name TEXT);"
    )
    conn = get_connection(db)
    run_migrations(conn, migrations_dir)
    cursor = conn.execute("SELECT name FROM sqlite_master WHERE type='table' AND name='foo'")
    assert cursor.fetchone() is not None
    conn.close()


def test_run_migrations_is_idempotent(tmp_path):
    db = tmp_path / "test.db"
    migrations_dir = tmp_path / "migrations"
    migrations_dir.mkdir()
    (migrations_dir / "001_create_foo.sql").write_text(
        "CREATE TABLE foo (id INTEGER PRIMARY KEY, name TEXT);"
    )
    conn = get_connection(db)
    run_migrations(conn, migrations_dir)
    run_migrations(conn, migrations_dir)  # second run must not raise
    conn.close()


def test_run_migrations_applies_in_order(tmp_path):
    db = tmp_path / "test.db"
    migrations_dir = tmp_path / "migrations"
    migrations_dir.mkdir()
    (migrations_dir / "001_create_foo.sql").write_text(
        "CREATE TABLE foo (id INTEGER PRIMARY KEY);"
    )
    (migrations_dir / "002_add_name.sql").write_text(
        "ALTER TABLE foo ADD COLUMN name TEXT;"
    )
    conn = get_connection(db)
    run_migrations(conn, migrations_dir)
    conn.execute("INSERT INTO foo (name) VALUES ('bar')")
    conn.close()

Step 2: Run tests to verify they fail

cd /Library/Development/CircuitForge/circuitforge-core
conda run -n job-seeker pytest tests/test_db.py -v

Expected: ImportError or AttributeError (module not yet defined)

Step 3: Write db/base.py (extracted from peregrine/scripts/db.py lines 1–40)

# circuitforge-core/circuitforge_core/db/base.py
"""
SQLite connection factory for CircuitForge products.
Supports plain SQLite and SQLCipher (AES-256) when CLOUD_MODE is active.
"""
from __future__ import annotations
import os
import sqlite3
from pathlib import Path


def get_connection(db_path: Path, key: str = "") -> sqlite3.Connection:
    """
    Open a SQLite database connection.

    In cloud mode with a key: uses SQLCipher (API-identical to sqlite3).
    Otherwise: plain sqlite3.

    Args:
        db_path: Path to the database file. Created if absent.
        key:     SQLCipher encryption key. Empty = unencrypted.
    """
    cloud_mode = os.environ.get("CLOUD_MODE", "").lower() in ("1", "true", "yes")
    if cloud_mode and key:
        from pysqlcipher3 import dbapi2 as _sqlcipher  # type: ignore
        conn = _sqlcipher.connect(str(db_path))
        conn.execute(f"PRAGMA key='{key}'")
        return conn
    return sqlite3.connect(str(db_path))

Step 4: Write db/migrations.py

# circuitforge-core/circuitforge_core/db/migrations.py
"""
Sequential SQL migration runner.
Applies *.sql files from migrations_dir in filename order.
Tracks applied migrations in a _migrations table — safe to call multiple times.
"""
from __future__ import annotations
import sqlite3
from pathlib import Path


def run_migrations(conn: sqlite3.Connection, migrations_dir: Path) -> None:
    """Apply any unapplied *.sql migrations from migrations_dir."""
    conn.execute(
        "CREATE TABLE IF NOT EXISTS _migrations "
        "(name TEXT PRIMARY KEY, applied_at TEXT DEFAULT CURRENT_TIMESTAMP)"
    )
    conn.commit()

    applied = {row[0] for row in conn.execute("SELECT name FROM _migrations")}
    sql_files = sorted(migrations_dir.glob("*.sql"))

    for sql_file in sql_files:
        if sql_file.name in applied:
            continue
        conn.executescript(sql_file.read_text())
        conn.execute("INSERT INTO _migrations (name) VALUES (?)", (sql_file.name,))
        conn.commit()

Step 5: Write db/init.py

# circuitforge-core/circuitforge_core/db/__init__.py
from .base import get_connection
from .migrations import run_migrations

__all__ = ["get_connection", "run_migrations"]

Step 6: Run tests to verify they pass

conda run -n job-seeker pytest tests/test_db.py -v

Expected: 5 PASSED

Step 7: Commit

git add circuitforge_core/db/ tests/
git commit -m "feat: add db base connection and migration runner"

Task 3: Extract tier system

Files:

Create: circuitforge-core/circuitforge_core/tiers/__init__.py
Create: circuitforge-core/circuitforge_core/tiers/tiers.py
Create: circuitforge-core/tests/test_tiers.py

Source: peregrine/app/wizard/tiers.py

Step 1: Write failing tests

# circuitforge-core/tests/test_tiers.py
import pytest
from circuitforge_core.tiers import can_use, TIERS, BYOK_UNLOCKABLE, LOCAL_VISION_UNLOCKABLE


def test_tiers_order():
    assert TIERS == ["free", "paid", "premium", "ultra"]


def test_free_feature_always_accessible():
    # Features not in FEATURES dict are free for everyone
    assert can_use("nonexistent_feature", tier="free") is True


def test_paid_feature_blocked_for_free_tier():
    # Caller must register features — test via can_use with explicit min_tier
    assert can_use("test_paid", tier="free", _features={"test_paid": "paid"}) is False


def test_paid_feature_accessible_for_paid_tier():
    assert can_use("test_paid", tier="paid", _features={"test_paid": "paid"}) is True


def test_byok_unlocks_byok_feature():
    byok_feature = next(iter(BYOK_UNLOCKABLE)) if BYOK_UNLOCKABLE else None
    if byok_feature:
        assert can_use(byok_feature, tier="free", has_byok=True) is True


def test_byok_does_not_unlock_non_byok_feature():
    assert can_use("test_paid", tier="free", has_byok=True,
                   _features={"test_paid": "paid"}) is False


def test_local_vision_unlocks_vision_feature():
    vision_feature = next(iter(LOCAL_VISION_UNLOCKABLE)) if LOCAL_VISION_UNLOCKABLE else None
    if vision_feature:
        assert can_use(vision_feature, tier="free", has_local_vision=True) is True


def test_local_vision_does_not_unlock_non_vision_feature():
    assert can_use("test_paid", tier="free", has_local_vision=True,
                   _features={"test_paid": "paid"}) is False

Step 2: Run tests to verify they fail

conda run -n job-seeker pytest tests/test_tiers.py -v

Expected: ImportError

Step 3: Write tiers/tiers.py (generalised from peregrine/app/wizard/tiers.py — remove all Peregrine-specific FEATURES entries; keep the can_use logic and unlock mechanism)

# circuitforge-core/circuitforge_core/tiers/tiers.py
"""
Tier system for CircuitForge products.

Tiers: free < paid < premium < ultra
Products register their own FEATURES dict and pass it to can_use().

BYOK_UNLOCKABLE: features that unlock when the user has any configured
LLM backend (local or API key). These are gated only because CF would
otherwise provide the compute.

LOCAL_VISION_UNLOCKABLE: features that unlock when the user has a local
vision model configured (e.g. moondream2). Distinct from BYOK — a text
LLM key does NOT unlock vision features.
"""
from __future__ import annotations

TIERS: list[str] = ["free", "paid", "premium", "ultra"]

# Features that unlock when the user has any LLM backend configured.
# Each product extends this frozenset with its own BYOK-unlockable features.
BYOK_UNLOCKABLE: frozenset[str] = frozenset()

# Features that unlock when the user has a local vision model configured.
LOCAL_VISION_UNLOCKABLE: frozenset[str] = frozenset()


def can_use(
    feature: str,
    tier: str,
    has_byok: bool = False,
    has_local_vision: bool = False,
    _features: dict[str, str] | None = None,
) -> bool:
    """
    Return True if the given tier (and optional unlocks) can access feature.

    Args:
        feature:          Feature key string.
        tier:             User's current tier ("free", "paid", "premium", "ultra").
        has_byok:         True if user has a configured LLM backend.
        has_local_vision: True if user has a local vision model configured.
        _features:        Feature→min_tier map. Products pass their own dict here.
                          If None, all features are free.
    """
    features = _features or {}
    if feature not in features:
        return True

    if has_byok and feature in BYOK_UNLOCKABLE:
        return True

    if has_local_vision and feature in LOCAL_VISION_UNLOCKABLE:
        return True

    min_tier = features[feature]
    try:
        return TIERS.index(tier) >= TIERS.index(min_tier)
    except ValueError:
        return False


def tier_label(
    feature: str,
    has_byok: bool = False,
    has_local_vision: bool = False,
    _features: dict[str, str] | None = None,
) -> str:
    """Return a human-readable label for the minimum tier needed for feature."""
    features = _features or {}
    if feature not in features:
        return "free"
    if has_byok and feature in BYOK_UNLOCKABLE:
        return "free (BYOK)"
    if has_local_vision and feature in LOCAL_VISION_UNLOCKABLE:
        return "free (local vision)"
    return features[feature]

Step 4: Write tiers/init.py

# circuitforge-core/circuitforge_core/tiers/__init__.py
from .tiers import can_use, tier_label, TIERS, BYOK_UNLOCKABLE, LOCAL_VISION_UNLOCKABLE

__all__ = ["can_use", "tier_label", "TIERS", "BYOK_UNLOCKABLE", "LOCAL_VISION_UNLOCKABLE"]

Step 5: Run tests

conda run -n job-seeker pytest tests/test_tiers.py -v

Expected: 8 PASSED

Step 6: Commit

git add circuitforge_core/tiers/ tests/test_tiers.py
git commit -m "feat: add generalised tier system with BYOK and local vision unlocks"

Task 4: Extract LLM router

Files:

Create: circuitforge-core/circuitforge_core/llm/__init__.py
Create: circuitforge-core/circuitforge_core/llm/router.py
Create: circuitforge-core/tests/test_llm_router.py

Source: peregrine/scripts/llm_router.py

Step 1: Write failing tests

# circuitforge-core/tests/test_llm_router.py
from unittest.mock import MagicMock, patch
import pytest
from circuitforge_core.llm import LLMRouter


def _make_router(config: dict) -> LLMRouter:
    """Build a router from an in-memory config dict (bypass file loading)."""
    router = object.__new__(LLMRouter)
    router.config = config
    return router


def test_complete_uses_first_reachable_backend():
    router = _make_router({
        "fallback_order": ["local"],
        "backends": {
            "local": {
                "type": "openai_compat",
                "base_url": "http://localhost:11434/v1",
                "model": "llama3",
                "supports_images": False,
            }
        }
    })
    mock_client = MagicMock()
    mock_client.chat.completions.create.return_value = MagicMock(
        choices=[MagicMock(message=MagicMock(content="hello"))]
    )
    with patch.object(router, "_is_reachable", return_value=True), \
         patch("circuitforge_core.llm.router.OpenAI", return_value=mock_client):
        result = router.complete("say hello")
    assert result == "hello"


def test_complete_falls_back_on_unreachable_backend():
    router = _make_router({
        "fallback_order": ["unreachable", "working"],
        "backends": {
            "unreachable": {
                "type": "openai_compat",
                "base_url": "http://nowhere:1/v1",
                "model": "x",
                "supports_images": False,
            },
            "working": {
                "type": "openai_compat",
                "base_url": "http://localhost:11434/v1",
                "model": "llama3",
                "supports_images": False,
            }
        }
    })
    mock_client = MagicMock()
    mock_client.chat.completions.create.return_value = MagicMock(
        choices=[MagicMock(message=MagicMock(content="fallback"))]
    )
    def reachable(url):
        return "nowhere" not in url
    with patch.object(router, "_is_reachable", side_effect=reachable), \
         patch("circuitforge_core.llm.router.OpenAI", return_value=mock_client):
        result = router.complete("test")
    assert result == "fallback"


def test_complete_raises_when_all_backends_exhausted():
    router = _make_router({
        "fallback_order": ["dead"],
        "backends": {
            "dead": {
                "type": "openai_compat",
                "base_url": "http://nowhere:1/v1",
                "model": "x",
                "supports_images": False,
            }
        }
    })
    with patch.object(router, "_is_reachable", return_value=False):
        with pytest.raises(RuntimeError, match="exhausted"):
            router.complete("test")

Step 2: Run tests to verify they fail

conda run -n job-seeker pytest tests/test_llm_router.py -v

Expected: ImportError

Step 3: Copy LLM router from Peregrine

Copy the full contents of /Library/Development/CircuitForge/peregrine/scripts/llm_router.py to circuitforge-core/circuitforge_core/llm/router.py, then update the one internal import:

# Change at top of file:
# OLD: CONFIG_PATH = Path(__file__).parent.parent / "config" / "llm.yaml"
# NEW:
CONFIG_PATH = Path.home() / ".config" / "circuitforge" / "llm.yaml"

Step 4: Write llm/init.py

# circuitforge-core/circuitforge_core/llm/__init__.py
from .router import LLMRouter

__all__ = ["LLMRouter"]

Step 5: Run tests

conda run -n job-seeker pytest tests/test_llm_router.py -v

Expected: 3 PASSED

Step 6: Commit

git add circuitforge_core/llm/ tests/test_llm_router.py
git commit -m "feat: add LLM router (extracted from Peregrine)"

Task 5: Add vision stub and config module

Files:

Create: circuitforge-core/circuitforge_core/vision/__init__.py
Create: circuitforge-core/circuitforge_core/vision/router.py
Create: circuitforge-core/circuitforge_core/config/__init__.py
Create: circuitforge-core/circuitforge_core/config/settings.py
Create: circuitforge-core/tests/test_config.py
Step 1: Write failing config tests

# circuitforge-core/tests/test_config.py
import os
import pytest
from circuitforge_core.config import require_env, load_env


def test_require_env_returns_value_when_set(monkeypatch):
    monkeypatch.setenv("TEST_KEY", "hello")
    assert require_env("TEST_KEY") == "hello"


def test_require_env_raises_when_missing(monkeypatch):
    monkeypatch.delenv("TEST_KEY", raising=False)
    with pytest.raises(EnvironmentError, match="TEST_KEY"):
        require_env("TEST_KEY")


def test_load_env_sets_variables(tmp_path, monkeypatch):
    env_file = tmp_path / ".env"
    env_file.write_text("FOO=bar\nBAZ=qux\n")
    monkeypatch.delenv("FOO", raising=False)
    load_env(env_file)
    assert os.environ.get("FOO") == "bar"
    assert os.environ.get("BAZ") == "qux"


def test_load_env_skips_missing_file(tmp_path):
    load_env(tmp_path / "nonexistent.env")  # must not raise

Step 2: Run to verify failure

conda run -n job-seeker pytest tests/test_config.py -v

Expected: ImportError

Step 3: Write config/settings.py

# circuitforge-core/circuitforge_core/config/settings.py
"""Env validation and .env loader for CircuitForge products."""
from __future__ import annotations
import os
from pathlib import Path


def require_env(key: str) -> str:
    """Return env var value or raise EnvironmentError with clear message."""
    value = os.environ.get(key)
    if not value:
        raise EnvironmentError(
            f"Required environment variable {key!r} is not set. "
            f"Check your .env file."
        )
    return value


def load_env(path: Path) -> None:
    """Load key=value pairs from a .env file into os.environ. Skips missing files."""
    if not path.exists():
        return
    for line in path.read_text().splitlines():
        line = line.strip()
        if not line or line.startswith("#") or "=" not in line:
            continue
        key, _, value = line.partition("=")
        os.environ.setdefault(key.strip(), value.strip())

Step 4: Write config/init.py

# circuitforge-core/circuitforge_core/config/__init__.py
from .settings import require_env, load_env

__all__ = ["require_env", "load_env"]

Step 5: Write vision stub

# circuitforge-core/circuitforge_core/vision/router.py
"""
Vision model router — stub until v0.2.
Supports: moondream2 (local) and Claude vision API (cloud).
"""
from __future__ import annotations


class VisionRouter:
    """Routes image analysis requests to local or cloud vision models."""

    def analyze(self, image_bytes: bytes, prompt: str) -> str:
        """
        Analyze image_bytes with the given prompt.
        Raises NotImplementedError until vision backends are wired up.
        """
        raise NotImplementedError(
            "VisionRouter is not yet implemented. "
            "Photo analysis requires a Paid tier or local vision model (v0.2+)."
        )

# circuitforge-core/circuitforge_core/vision/__init__.py
from .router import VisionRouter

__all__ = ["VisionRouter"]

Step 6: Run all tests

conda run -n job-seeker pytest tests/ -v

Expected: All PASSED (config tests pass; vision stub has no tests — it's a placeholder)

Step 7: Commit

git add circuitforge_core/vision/ circuitforge_core/config/ tests/test_config.py
git commit -m "feat: add config module and vision router stub"

Task 5b: Add wizard and pipeline stubs

Files:

Create: circuitforge-core/circuitforge_core/wizard/__init__.py
Create: circuitforge-core/circuitforge_core/wizard/base.py
Create: circuitforge-core/circuitforge_core/pipeline/__init__.py
Create: circuitforge-core/circuitforge_core/pipeline/staging.py
Create: circuitforge-core/tests/test_stubs.py

These modules are required by the spec (section 2.2) but are net-new (wizard) or partially net-new (pipeline). They are stubbed here so downstream products can import them; full implementation happens when each product needs them.

Step 1: Write failing stub import tests

# circuitforge-core/tests/test_stubs.py
import pytest
from circuitforge_core.wizard import BaseWizard
from circuitforge_core.pipeline import StagingDB


def test_wizard_raises_not_implemented():
    wizard = BaseWizard()
    with pytest.raises(NotImplementedError):
        wizard.run()


def test_pipeline_raises_not_implemented():
    staging = StagingDB()
    with pytest.raises(NotImplementedError):
        staging.enqueue("job", {})

Step 2: Run to verify failure

conda run -n job-seeker pytest tests/test_stubs.py -v

Expected: ImportError

Step 3: Write wizard stub

# circuitforge-core/circuitforge_core/wizard/base.py
"""
First-run onboarding wizard base class.
Full implementation is net-new per product (v0.1+ for snipe, etc.)
"""
from __future__ import annotations


class BaseWizard:
    """
    Base class for CircuitForge first-run wizards.
    Subclass and implement run() in each product.
    """

    def run(self) -> None:
        """Execute the onboarding wizard flow. Must be overridden by subclass."""
        raise NotImplementedError(
            "BaseWizard.run() must be implemented by a product-specific subclass."
        )

# circuitforge-core/circuitforge_core/wizard/__init__.py
from .base import BaseWizard

__all__ = ["BaseWizard"]

Step 4: Write pipeline stub

# circuitforge-core/circuitforge_core/pipeline/staging.py
"""
SQLite-backed staging queue for CircuitForge pipeline tasks.
Full implementation deferred — stub raises NotImplementedError.
"""
from __future__ import annotations
from typing import Any


class StagingDB:
    """
    Staging queue for background pipeline tasks (search polling, score updates, etc.)
    Stub: raises NotImplementedError until wired up in a product.
    """

    def enqueue(self, task_type: str, payload: dict[str, Any]) -> None:
        """Add a task to the staging queue."""
        raise NotImplementedError(
            "StagingDB.enqueue() is not yet implemented. "
            "Background task pipeline is a v0.2+ feature."
        )

    def dequeue(self) -> tuple[str, dict[str, Any]] | None:
        """Fetch the next pending task. Returns (task_type, payload) or None."""
        raise NotImplementedError(
            "StagingDB.dequeue() is not yet implemented."
        )

# circuitforge-core/circuitforge_core/pipeline/__init__.py
from .staging import StagingDB

__all__ = ["StagingDB"]

Step 5: Run tests

conda run -n job-seeker pytest tests/test_stubs.py -v

Expected: 2 PASSED

Step 6: Verify all tests still pass

conda run -n job-seeker pytest tests/ -v

Expected: All PASSED

Step 7: Commit

git add circuitforge_core/wizard/ circuitforge_core/pipeline/ tests/test_stubs.py
git commit -m "feat: add wizard and pipeline stubs"

Task 6: Update Peregrine to use circuitforge-core

Files:

Modify: peregrine/requirements.txt
Modify: peregrine/scripts/db.py
Modify: peregrine/scripts/llm_router.py
Modify: peregrine/app/wizard/tiers.py
Step 1: Add circuitforge-core to Peregrine requirements

Add to /Library/Development/CircuitForge/peregrine/requirements.txt (top of file, before other deps):

-e ../circuitforge-core

Step 2: Verify install

cd /Library/Development/CircuitForge/peregrine
conda run -n job-seeker pip install -r requirements.txt
conda run -n job-seeker python -c "from circuitforge_core.db import get_connection; print('ok')"

Expected: ok

Step 3: Update peregrine/scripts/db.py — replace get_connection

Add at top of file (after existing docstring and imports):

from circuitforge_core.db import get_connection as _cf_get_connection

def get_connection(db_path=DEFAULT_DB, key=""):
    """Thin shim — delegates to circuitforge_core.db.get_connection."""
    return _cf_get_connection(db_path, key)

Remove the old get_connection function body (lines ~15–35). Keep all Peregrine-specific schema (CREATE_JOBS, CREATE_COVER_LETTERS, etc.) and functions unchanged.

Step 4: Run Peregrine tests to verify nothing broke

cd /Library/Development/CircuitForge/peregrine
conda run -n job-seeker pytest tests/ -v -x

Expected: All existing tests pass

Step 5: Update peregrine/scripts/llm_router.py — replace LLMRouter

Replace the LLMRouter class definition with:

from circuitforge_core.llm import LLMRouter  # noqa: F401 — re-export for existing callers

Keep the module-level CONFIG_PATH constant so any code that references llm_router.CONFIG_PATH still works.

Step 6: Run Peregrine tests again

conda run -n job-seeker pytest tests/ -v -x

Expected: All passing

Step 7: Update peregrine/app/wizard/tiers.py — import from core

At the top of the file, add:

from circuitforge_core.tiers import can_use as _core_can_use, TIERS, tier_label

Update the can_use function to delegate to core, passing Peregrine's FEATURES dict:

def can_use(feature: str, tier: str, has_byok: bool = False, has_local_vision: bool = False) -> bool:
    return _core_can_use(feature, tier, has_byok=has_byok, has_local_vision=has_local_vision, _features=FEATURES)

Keep the Peregrine-specific FEATURES dict and BYOK_UNLOCKABLE frozenset in place — they are still defined here, just used via the core function.

Step 8: Run full Peregrine tests

conda run -n job-seeker pytest tests/ -v

Expected: All passing

Step 9: Smoke-test Peregrine startup

cd /Library/Development/CircuitForge/peregrine
./manage.sh status  # verify it's running, or start it

Manually open http://localhost:8502 and verify the UI loads without errors.

Step 10: Commit Peregrine changes

cd /Library/Development/CircuitForge/peregrine
git add requirements.txt scripts/db.py scripts/llm_router.py app/wizard/tiers.py
git commit -m "feat: migrate to circuitforge-core for db, llm router, and tiers"

Step 11: Commit circuitforge-core README and push both repos

cd /Library/Development/CircuitForge/circuitforge-core
# Write a minimal README then:
git add README.md
git commit -m "docs: add README"
git remote add origin https://git.opensourcesolarpunk.com/Circuit-Forge/circuitforge-core.git
git push -u origin main

Task 7: Final verification

Step 1: Run circuitforge-core full test suite

cd /Library/Development/CircuitForge/circuitforge-core
conda run -n job-seeker pytest tests/ -v --tb=short

Expected: All PASSED

Step 2: Run Peregrine full test suite

cd /Library/Development/CircuitForge/peregrine
conda run -n job-seeker pytest tests/ -v --tb=short

Expected: All PASSED

Step 3: Verify editable install works from a fresh shell

conda run -n job-seeker python -c "
from circuitforge_core.db import get_connection, run_migrations
from circuitforge_core.llm import LLMRouter
from circuitforge_core.tiers import can_use, TIERS
from circuitforge_core.config import require_env, load_env
from circuitforge_core.vision import VisionRouter
from circuitforge_core.wizard import BaseWizard
from circuitforge_core.pipeline import StagingDB
print('All imports OK')
"

Expected: All imports OK

31 KiB Raw Blame History Unescape Escape

circuitforge-core Extraction Implementation Plan

File Map

New files (circuitforge-core repo)

Modified files (Peregrine repo)

Task 1: Scaffold circuitforge-core repo

Task 2: Extract db base connection

Task 3: Extract tier system

Task 4: Extract LLM router

Task 5: Add vision stub and config module

Task 5b: Add wizard and pipeline stubs

Task 6: Update Peregrine to use circuitforge-core

Task 7: Final verification

31 KiB

Raw Blame History