ai-resume/corpus/config.py

"""Configuration for the corpus pipeline. Environment only.

Nothing in this file holds a credential. The OmniRoute key is read from the
process environment, which systemd populates from
`~/.config/environment.d/10-secrets.conf` (AGENTS.md section 4.4). `describe()`
is the only way the pipeline prints its configuration, and it redacts the key.
"""

from __future__ import annotations

import os
from dataclasses import dataclass

# Bumped whenever chunk boundaries or payload shape change. A stored corpus
# carrying a different version is stale by definition and gets rebuilt.
CHUNKER_VERSION = "blocks-v1"

# Provider defaults. `local-hashing` is a deterministic, offline provider used
# by the tests, by `make corpus-test`, and by any rebuild on a host with no
# model gateway. It is NOT a semantic embedding model: it exists so the whole
# pipeline can be exercised without a provider decision, and so a rebuild never
# depends on the network. The production provider is `omniroute`.
PROVIDERS = {
    "local-hashing": ("local-hashing-v1", 256),
    "omniroute": ("text-embedding-3-small", 1536),
}

DEFAULT_QDRANT_URL = "http://127.0.0.1:6333"
DEFAULT_COLLECTION = "resume_corpus"
DEFAULT_PROVIDER = "local-hashing"
DEFAULT_OMNIROUTE_URL = "https://omniroute.lab.audasmedia.com.au"

# Freshness horizon: a document whose `last_verified` is older than this is
# reported stale by the nightly audit. docs/phase-4-notes.md section 5.
DEFAULT_MAX_AGE_DAYS = 180
DEFAULT_WARN_DAYS = 30


class ConfigError(RuntimeError):
    """A configuration value is missing or unusable."""


@dataclass(frozen=True)
class Config:
    """Everything the pipeline needs, resolved once."""

    root: str
    content_dir: str
    data_dir: str
    db_path: str
    qdrant_url: str
    collection: str
    provider: str
    model: str
    dim: int
    api_url: str
    api_key: str
    batch_size: int
    timeout: float
    max_age_days: int
    warn_days: int
    apprise_url: str
    vectors_on_disk: bool = False

    # -- construction -----------------------------------------------------

    @classmethod
    def from_env(cls, root: str | None = None, **overrides) -> "Config":
        """Build a Config from the environment, then apply explicit overrides.

        Overrides exist so the CLI can accept flags without a second code path;
        they are applied last so a flag always wins over the environment.
        """
        root = os.path.abspath(root or os.environ.get("RESUME_ROOT") or _default_root())
        content_dir = os.environ.get("RESUME_CONTENT") or os.path.join(root, "content")
        data_dir = os.environ.get("RESUME_CORPUS_DATA") or os.path.join(root, "data", "corpus")
        provider = os.environ.get("RESUME_EMBED_PROVIDER") or DEFAULT_PROVIDER
        if provider not in PROVIDERS:
            raise ConfigError(
                f"RESUME_EMBED_PROVIDER={provider!r} is not one of {sorted(PROVIDERS)}"
            )
        default_model, default_dim = PROVIDERS[provider]

        values = dict(
            root=root,
            content_dir=os.path.abspath(content_dir),
            data_dir=os.path.abspath(data_dir),
            db_path=os.path.abspath(
                os.environ.get("RESUME_CORPUS_DB") or os.path.join(data_dir, "corpus.db")
            ),
            qdrant_url=(os.environ.get("RESUME_QDRANT_URL") or DEFAULT_QDRANT_URL).rstrip("/"),
            collection=os.environ.get("RESUME_COLLECTION") or DEFAULT_COLLECTION,
            provider=provider,
            model=os.environ.get("RESUME_EMBED_MODEL") or default_model,
            dim=int(os.environ.get("RESUME_EMBED_DIM") or default_dim),
            api_url=(os.environ.get("OMNIROUTE_URL") or DEFAULT_OMNIROUTE_URL).rstrip("/"),
            api_key=os.environ.get("OMNIROUTE_API_KEY") or "",
            batch_size=int(os.environ.get("RESUME_EMBED_BATCH") or 32),
            timeout=float(os.environ.get("RESUME_QDRANT_TIMEOUT") or 30.0),
            max_age_days=int(
                os.environ.get("RESUME_MAX_AGE_DAYS") or DEFAULT_MAX_AGE_DAYS
            ),
            warn_days=int(os.environ.get("RESUME_WARN_DAYS") or DEFAULT_WARN_DAYS),
            apprise_url=os.environ.get("RESUME_APPRISE_URL") or "",
            vectors_on_disk=(os.environ.get("RESUME_QDRANT_VECTORS_ON_DISK") or "").lower()
            in ("1", "true", "yes"),
        )
        values.update({k: v for k, v in overrides.items() if v not in (None, "")})
        cfg = cls(**values)
        cfg.check()
        return cfg

    # -- validation -------------------------------------------------------

    def check(self) -> None:
        if self.dim <= 0:
            raise ConfigError(f"embedding dimension must be positive, got {self.dim}")
        if self.batch_size <= 0:
            raise ConfigError(f"batch size must be positive, got {self.batch_size}")
        if self.provider == "omniroute" and not self.api_key:
            raise ConfigError(
                "provider 'omniroute' needs OMNIROUTE_API_KEY in the environment "
                "(~/.config/environment.d/10-secrets.conf). It is never read from this repo."
            )
        if not os.path.isdir(self.content_dir):
            raise ConfigError(f"content directory not found: {self.content_dir}")

    # -- presentation -----------------------------------------------------

    def describe(self) -> str:
        """Human-readable configuration. Never includes the API key value."""
        key = "set" if self.api_key else "unset"
        return (
            f"root={self.root}\n"
            f"content={self.content_dir}\n"
            f"data={self.data_dir}\n"
            f"db={self.db_path}\n"
            f"qdrant={self.qdrant_url} collection={self.collection}\n"
            f"provider={self.provider} model={self.model} dim={self.dim} "
            f"batch={self.batch_size}\n"
            f"api_url={self.api_url} api_key={key}\n"
            f"freshness=max_age_days:{self.max_age_days} warn_days:{self.warn_days}\n"
            f"qdrant vectors_on_disk={self.vectors_on_disk}\n"
            f"chunker={CHUNKER_VERSION}"
        )

    # -- derived paths ----------------------------------------------------

    @property
    def chunker_version(self) -> str:
        return CHUNKER_VERSION

    @property
    def meta_path(self) -> str:
        return os.path.join(self.data_dir, "collection.json")


def _default_root() -> str:
    """The repository root: the parent of this package."""
    return os.path.dirname(os.path.dirname(os.path.abspath(__file__)))