feat(07-03): classifier wired to load_provider_config + ai_config stub removed — D-04/D-06

- Remove _ProviderConfigStub from services/ai_config.py (replaced by real ProviderConfig)
- Add module-level import: from ai.provider_config import ProviderConfig, PROVIDER_DEFAULTS
- load_provider_config() now returns ProviderConfig (no lazy import inside function body)
- classifier.py: replace inline _settings dict with load_provider_config(session) call (D-06)
- Per-user override path builds ProviderConfig from PROVIDER_DEFAULTS (no api_key — T-07-06)
- Fallback to app_settings defaults when load_provider_config returns None (D-15)
- Truncation delegated to provider._truncate() — no more text slices in classifier (D-12/D-13)
- Promote test_api_key_encrypt_decrypt to passing (round-trip + domain salt isolation)
- Update test_classifier.py: test_per_user_provider + test_default_provider_fallback use ProviderConfig assertions

backend/services/ai_config.py

@@ -22,38 +22,21 @@ from __future__ import annotations
 
 import base64
 import logging
-from typing import Optional, TYPE_CHECKING
+from typing import Optional
 
 import structlog
 from cryptography.fernet import Fernet
 from cryptography.hazmat.primitives import hashes
 from cryptography.hazmat.primitives.kdf.hkdf import HKDF
-from pydantic import BaseModel
 from sqlalchemy import select
 from sqlalchemy.ext.asyncio import AsyncSession
 
+from ai.provider_config import ProviderConfig, PROVIDER_DEFAULTS
 from config import settings
 
-# ProviderConfig redefined in ai/provider_config.py during Plan 02 —
-# load_provider_config will re-import and return that class once Plan 02 lands.
-# The stub below is removed in Plan 03 once the classifier consumes the real class.
-
-
 logger = structlog.get_logger(__name__)
 
 
-# ── Stub ProviderConfig for Plan 01 (replaced by ai/provider_config.py in Plan 02) ──
-
-class _ProviderConfigStub(BaseModel):
-    """Minimal provider config placeholder until Plan 02 creates ai/provider_config.py."""
-
-    provider_id: str
-    api_key: str = ""
-    base_url: Optional[str] = None
-    model: str = ""
-    context_chars: int = 8000
-
-
 # ── HKDF key derivation ───────────────────────────────────────────────────────
 
 def _derive_ai_settings_key(master_key: bytes, provider_id: str) -> Fernet:
@@ -125,31 +108,19 @@ def decrypt_api_key(master_key: bytes, provider_id: str, api_key_enc: str) -> st
 
 # ── Provider config loader ────────────────────────────────────────────────────
 
-async def load_provider_config(session: AsyncSession) -> Optional[_ProviderConfigStub]:
+async def load_provider_config(session: AsyncSession) -> Optional[ProviderConfig]:
     """Load the active AI provider config from the system_settings table.
 
-    Returns a _ProviderConfigStub built from the row where is_active=True,
+    Returns a ProviderConfig built from the row where is_active=True,
     decrypting api_key_enc when present.  Returns None when no active row exists.
 
-    Note: In Plan 02, this function will try to import and return ProviderConfig
-    from ai/provider_config.py instead of the stub. The import is done lazily
-    inside the function body so that Plan 01 does not depend on files that don't
-    exist yet.
-
     Args:
         session: An open AsyncSession.
 
     Returns:
-        A _ProviderConfigStub (or the real ProviderConfig from Plan 02+) if an
-        active row exists; None if the table is empty or no row is marked active.
+        A ProviderConfig if an active row exists; None if the table is empty or
+        no row is marked active.
     """
-    # Lazy import: try to use the real ProviderConfig once Plan 02 lands
-    try:
-        from ai.provider_config import ProviderConfig as _RealProviderConfig  # type: ignore[import]
-        config_cls = _RealProviderConfig
-    except ImportError:
-        config_cls = _ProviderConfigStub  # type: ignore[assignment]
-
     from db.models import SystemSettings  # local import to avoid circular deps
 
     stmt = select(SystemSettings).where(SystemSettings.is_active.is_(True))
@@ -171,7 +142,7 @@ async def load_provider_config(session: AsyncSession) -> Optional[_ProviderConfi
                 provider_id=row.provider_id,
             )
 
-    return config_cls(
+    return ProviderConfig(
         provider_id=row.provider_id,
         api_key=api_key,
         base_url=row.base_url,
@@ -208,11 +179,14 @@ async def seed_system_settings_from_env(session: AsyncSession) -> None:
         # Row already exists — never overwrite (idempotent)
         return
 
+    # Use PROVIDER_DEFAULTS context_chars for the provider, fallback to 8000
+    context_chars = PROVIDER_DEFAULTS.get(provider_id, {}).get("context_chars", 8000)
+
     # Insert default row with no API key (local providers like Ollama don't need one)
     row = SystemSettings(
         provider_id=provider_id,
         model_name=model_name,
-        context_chars=8000,
+        context_chars=context_chars,
         is_active=True,
         api_key_enc=None,
         base_url=None,

backend/services/classifier.py

@@ -13,6 +13,13 @@ the user's namespace via create_topic(user_id=doc.user_id) (D-11).
 Updated in Plan 03-04: classify_document and suggest_topics_for_document now accept
 ai_provider and ai_model kwargs. No longer calls storage.load_settings(). Provider
 resolved via get_provider() using per-user settings from DB (D-14, D-15).
+
+Updated in Plan 07-03: Provider resolved via load_provider_config(session) reading
+from the system_settings DB table (D-04/D-06). Per-user override still honoured:
+when ai_provider is non-None, a ProviderConfig is constructed from PROVIDER_DEFAULTS
+for that provider (note: per-user override does not carry an api_key — admin must
+configure each provider's key in system_settings). Truncation delegated to provider
+_truncate() — no more inline text slicing in the classifier (D-12/D-13).
 """
 from __future__ import annotations
 
@@ -23,7 +30,9 @@ from sqlalchemy.ext.asyncio import AsyncSession
 from config import settings as app_settings
 from db.models import Document
 from services import storage
+from services.ai_config import load_provider_config
 from ai import get_provider
+from ai.provider_config import ProviderConfig, PROVIDER_DEFAULTS
 
 _DEFAULT_SYSTEM_PROMPT = """You are a document classification assistant. When given a document's text content and a list of existing topics, you must:
 1. Assign the document to one or more relevant topics from the list.
@@ -47,19 +56,45 @@ async def classify_document(
 
     ai_provider and ai_model come from the document owner's User record (D-14).
     Falls back to app_settings.default_ai_provider / default_ai_model when None (D-15).
+    Provider config loaded from system_settings DB table via load_provider_config() (D-04).
     """
     meta = await storage.get_metadata(session, doc_id)
     if meta is None:
         raise ValueError(f"Document {doc_id} not found")
 
-    _ai_provider = ai_provider or app_settings.default_ai_provider
-    _ai_model = ai_model or app_settings.default_ai_model
     system_prompt = app_settings.system_prompt or _DEFAULT_SYSTEM_PROMPT
-    _settings = {
-        "active_provider": _ai_provider,
-        "providers": {_ai_provider: {"model": _ai_model}},
-    }
-    provider = get_provider(_settings)
+
+    # ── Provider resolution (D-04/D-06) ─────────────────────────────────────
+    if ai_provider is not None:
+        # Per-user override path: build a ProviderConfig from PROVIDER_DEFAULTS.
+        # per-user override does not carry an api_key — admin must configure each
+        # provider's key in system_settings. When the per-user override selects a
+        # different provider than the active system provider, api_key stays empty
+        # and get_provider() normalises it to "not-needed".
+        config = ProviderConfig(
+            provider_id=ai_provider,
+            model=ai_model or PROVIDER_DEFAULTS.get(ai_provider, {}).get("model", ""),
+            api_key="",
+            base_url=None,
+            context_chars=PROVIDER_DEFAULTS.get(ai_provider, {}).get("context_chars", 8000),
+        )
+    else:
+        # System provider path: load from DB (D-04)
+        config = await load_provider_config(session)
+        if config is None:
+            # No active row in system_settings — fall back to env-var defaults (D-15)
+            fallback_provider = app_settings.default_ai_provider
+            config = ProviderConfig(
+                provider_id=fallback_provider,
+                model=app_settings.default_ai_model,
+                api_key="",
+                base_url=None,
+                context_chars=PROVIDER_DEFAULTS.get(fallback_provider, {}).get(
+                    "context_chars", 8000
+                ),
+            )
+
+    provider = get_provider(config)
 
     # Load the Document ORM object to get the owner's user_id (D-11, D-17)
     try:
@@ -80,6 +115,7 @@ async def classify_document(
         topic_names = [t["name"] for t in all_topics]
 
     text = meta.get("extracted_text", "")
+    # Truncation is performed inside provider.classify() via provider._truncate(). (D-12/D-13)
     result = await provider.classify(text, topic_names, system_prompt)
 
     # Collect all topic names to persist (assigned + suggested)
@@ -108,18 +144,38 @@ async def suggest_topics_for_document(
 
     ai_provider and ai_model come from the document owner's User record (D-14).
     Falls back to app_settings.default_ai_provider / default_ai_model when None (D-15).
+    Provider config loaded from system_settings DB table via load_provider_config() (D-04).
     """
     meta = await storage.get_metadata(session, doc_id)
     if meta is None:
         raise ValueError(f"Document {doc_id} not found")
 
-    _ai_provider = ai_provider or app_settings.default_ai_provider
-    _ai_model = ai_model or app_settings.default_ai_model
     system_prompt = app_settings.system_prompt or _DEFAULT_SYSTEM_PROMPT
-    _settings = {
-        "active_provider": _ai_provider,
-        "providers": {_ai_provider: {"model": _ai_model}},
-    }
-    provider = get_provider(_settings)
+
+    # ── Provider resolution (D-04/D-06) ─────────────────────────────────────
+    if ai_provider is not None:
+        config = ProviderConfig(
+            provider_id=ai_provider,
+            model=ai_model or PROVIDER_DEFAULTS.get(ai_provider, {}).get("model", ""),
+            api_key="",
+            base_url=None,
+            context_chars=PROVIDER_DEFAULTS.get(ai_provider, {}).get("context_chars", 8000),
+        )
+    else:
+        config = await load_provider_config(session)
+        if config is None:
+            fallback_provider = app_settings.default_ai_provider
+            config = ProviderConfig(
+                provider_id=fallback_provider,
+                model=app_settings.default_ai_model,
+                api_key="",
+                base_url=None,
+                context_chars=PROVIDER_DEFAULTS.get(fallback_provider, {}).get(
+                    "context_chars", 8000
+                ),
+            )
+
+    provider = get_provider(config)
     text = meta.get("extracted_text", "")
+    # Truncation is performed inside provider.suggest_topics() via provider._truncate(). (D-12/D-13)
     return await provider.suggest_topics(text, system_prompt)

backend/tests/test_ai_config.py

@@ -1,25 +1,100 @@
 """
-Wave 0 xfail stubs for Phase 7 AI config service tests.
+Tests for Phase 7 AI config service.
 
 Covers:
   - D-04: load_provider_config() reads from system_settings table
   - D-05: API key HKDF encryption round-trip
 
-Each function is a placeholder to be promoted in Plan 07-01 (green gate) once
-the services/ai_config.py module and system_settings table exist.
-
-Stub policy (STATE.md decision: xfail(strict=False) for Wave 0):
-  - Body is a single pytest.xfail() call — no assertion code.
-  - strict=False so unexpected passes (xpass) never break CI.
+Plan 07-03 promotes: test_api_key_encrypt_decrypt, test_load_provider_config.
 """
+import os
 import pytest
 
 
-@pytest.mark.xfail(strict=False, reason="Wave 0 stub — promoted in Plan 07-01")
-async def test_load_provider_config():
-    pytest.xfail("not implemented yet — Plan 07-01")
+# ---------------------------------------------------------------------------
+# test_api_key_encrypt_decrypt — D-05 HKDF encryption round-trip
+# ---------------------------------------------------------------------------
+
+def test_api_key_encrypt_decrypt():
+    """encrypt_api_key + decrypt_api_key round-trip returns original plaintext.
+
+    Also verifies domain salt isolation: different provider_ids produce different
+    ciphertexts from the same master key and plaintext (T-07-02).
+    """
+    from services.ai_config import encrypt_api_key, decrypt_api_key
+
+    # Use a 32-byte master key (CLOUD_CREDS_KEY is Base64-encoded in production,
+    # but the helpers accept raw bytes — use ASCII bytes for simplicity here)
+    master_key = b"a" * 32  # 32 bytes deterministic test key
+
+    # Round-trip: encrypt then decrypt must return original plaintext
+    plaintext = "sk-test-api-key-abc123"
+    ciphertext = encrypt_api_key(master_key, "openai", plaintext)
+    recovered = decrypt_api_key(master_key, "openai", ciphertext)
+    assert recovered == plaintext, "Decrypted value must match original plaintext"
+
+    # Domain salt isolation: same key + same plaintext, different provider_id → different ciphertext
+    ciphertext_anthropic = encrypt_api_key(master_key, "anthropic", plaintext)
+    assert ciphertext != ciphertext_anthropic, (
+        "Different provider_ids must produce different ciphertexts "
+        "(HKDF salt isolation — T-07-02)"
+    )
+
+    # Cross-provider decrypt must fail (Fernet raises InvalidToken)
+    from cryptography.fernet import InvalidToken
+    with pytest.raises(InvalidToken):
+        decrypt_api_key(master_key, "anthropic", ciphertext)  # openai key, anthropic provider_id
 
 
-@pytest.mark.xfail(strict=False, reason="Wave 0 stub — promoted in Plan 07-01")
-async def test_api_key_encrypt_decrypt():
-    pytest.xfail("not implemented yet — Plan 07-01")
+# ---------------------------------------------------------------------------
+# test_load_provider_config — D-04 DB integration (skipped without PostgreSQL)
+# ---------------------------------------------------------------------------
+
+@pytest.mark.skipif(
+    not os.getenv("INTEGRATION"),
+    reason="needs PostgreSQL — set INTEGRATION=1 to run",
+)
+@pytest.mark.asyncio
+async def test_load_provider_config(db_session):
+    """load_provider_config() reads the active row from system_settings and decrypts the api_key.
+
+    Requires a live PostgreSQL session (db_session fixture from conftest.py).
+    Run with: INTEGRATION=1 pytest tests/test_ai_config.py::test_load_provider_config
+    """
+    pytest.importorskip("psycopg")
+
+    from services.ai_config import encrypt_api_key, load_provider_config
+    from db.models import SystemSettings
+    import uuid
+
+    master_key = b"b" * 32  # deterministic test key
+    test_api_key = "sk-integration-test-key"
+    provider_id = f"test-provider-{uuid.uuid4().hex[:8]}"
+
+    # Encrypt the API key
+    api_key_enc = encrypt_api_key(master_key, provider_id, test_api_key)
+
+    # Insert a SystemSettings row with is_active=True
+    row = SystemSettings(
+        provider_id=provider_id,
+        api_key_enc=api_key_enc,
+        base_url="https://api.example.com/v1",
+        model_name="test-model",
+        context_chars=64000,
+        is_active=True,
+    )
+    db_session.add(row)
+    await db_session.flush()
+
+    # Patch settings.cloud_creds_key to match our test master key
+    from unittest.mock import patch
+    with patch("services.ai_config.settings") as mock_settings:
+        mock_settings.cloud_creds_key = master_key.decode("utf-8")
+        result = await load_provider_config(db_session)
+
+    assert result is not None, "load_provider_config must return a ProviderConfig"
+    assert result.provider_id == provider_id
+    assert result.api_key == test_api_key, "api_key must be decrypted correctly"
+    assert result.base_url == "https://api.example.com/v1"
+    assert result.model == "test-model"
+    assert result.context_chars == 64000

backend/tests/test_classifier.py

@@ -119,13 +119,17 @@ async def test_classifier_with_mock_provider(isolated_data_dir):
 @pytest.mark.asyncio
 async def test_per_user_provider(db_session):
     """When ai_provider='openai' and ai_model='gpt-4o' are passed to the classifier,
-    it resolves _settings['active_provider'] == 'openai'.
+    it builds a ProviderConfig with provider_id='openai' and model='gpt-4o'.
 
     DOC-03: AI provider/model comes from the user's DB record (passed through from
     _run) not from global config or the retired load_settings() flat file (D-14).
+
+    Plan 07-03: get_provider now receives a ProviderConfig object (D-06), not a
+    raw dict. Assertions updated accordingly.
     """
     from unittest.mock import AsyncMock, patch, MagicMock
     from ai.base import ClassificationResult
+    from ai.provider_config import ProviderConfig
     from services.classifier import classify_document
     import uuid
 
@@ -136,10 +140,10 @@ async def test_per_user_provider(db_session):
     mock_doc = MagicMock()
     mock_doc.user_id = user_id
 
-    captured_settings = {}
+    captured_configs = []
 
-    def capture_get_provider(settings):
-        captured_settings.update(settings)
+    def capture_get_provider(config):
+        captured_configs.append(config)
         mock_provider = MagicMock()
         mock_provider.classify = AsyncMock(return_value=ClassificationResult(
             topics=[], suggested_new_topics=[], reasoning=""
@@ -156,9 +160,11 @@ async def test_per_user_provider(db_session):
          patch("services.classifier.get_provider", side_effect=capture_get_provider):
         await classify_document(mock_session, doc_id, ai_provider="openai", ai_model="gpt-4o")
 
-    assert captured_settings.get("active_provider") == "openai"
-    assert "openai" in captured_settings.get("providers", {})
-    assert captured_settings["providers"]["openai"]["model"] == "gpt-4o"
+    assert len(captured_configs) == 1
+    config = captured_configs[0]
+    assert isinstance(config, ProviderConfig), "get_provider must receive a ProviderConfig (D-06)"
+    assert config.provider_id == "openai"
+    assert config.model == "gpt-4o"
 
 
 @pytest.mark.asyncio
@@ -224,13 +230,18 @@ async def test_celery_task_uses_user_provider(db_session):
 
 @pytest.mark.asyncio
 async def test_default_provider_fallback(db_session):
-    """When user.ai_provider is None, the classifier receives config.settings.default_ai_provider.
+    """When user.ai_provider is None, the classifier uses the system_settings DB row or env fallback.
 
-    D-15: fallback chain is user.ai_provider → DEFAULT_AI_PROVIDER env var →
+    D-15: fallback chain is user.ai_provider → system_settings DB → DEFAULT_AI_PROVIDER env var →
     code default 'ollama' (CONTEXT.md D-15).
+
+    Plan 07-03: get_provider now receives a ProviderConfig (D-06). When ai_provider=None
+    and load_provider_config returns None, the classifier falls back to
+    app_settings.default_ai_provider. Assertions updated from dict to ProviderConfig.
     """
     from unittest.mock import AsyncMock, patch, MagicMock
     from ai.base import ClassificationResult
+    from ai.provider_config import ProviderConfig
     from services.classifier import classify_document
     import uuid
 
@@ -241,10 +252,10 @@ async def test_default_provider_fallback(db_session):
     mock_doc = MagicMock()
     mock_doc.user_id = user_id
 
-    captured_settings = {}
+    captured_configs = []
 
-    def capture_get_provider(settings):
-        captured_settings.update(settings)
+    def capture_get_provider(config):
+        captured_configs.append(config)
         mock_provider = MagicMock()
         mock_provider.classify = AsyncMock(return_value=ClassificationResult(
             topics=[], suggested_new_topics=[], reasoning=""
@@ -255,6 +266,7 @@ async def test_default_provider_fallback(db_session):
          patch("services.classifier.storage.load_topics_for_user", AsyncMock(return_value=[])), \
          patch("services.classifier.storage.load_topics", AsyncMock(return_value=[])), \
          patch("services.classifier.storage.update_document_topics", AsyncMock(return_value=None)), \
+         patch("services.classifier.load_provider_config", AsyncMock(return_value=None)), \
          patch("services.classifier.get_provider", side_effect=capture_get_provider):
 
         mock_session = AsyncMock()
@@ -262,5 +274,8 @@ async def test_default_provider_fallback(db_session):
         # Pass ai_provider=None to trigger the default fallback (D-15)
         await classify_document(mock_session, doc_id, ai_provider=None, ai_model=None)
 
+    assert len(captured_configs) == 1
+    config = captured_configs[0]
+    assert isinstance(config, ProviderConfig), "get_provider must receive a ProviderConfig (D-06)"
     # Should fall back to app_settings.default_ai_provider = "ollama"
-    assert captured_settings.get("active_provider") == "ollama"
+    assert config.provider_id == "ollama"