src/knowledge_domain/service.py

"""KnowledgeService — the interface agents read domain knowledge through (K1-K4).

**Three calls, and they are the product.** The end goal is an MCP service
exposing modular domain knowledge to agents we do not own
(`knowledge_pipeline_context.md` §1), so these three are what an external client
would be given. Our planner becomes the FIRST client rather than the only one,
which makes the eventual MCP server a transport rather than a redesign.

    get_context(user_id, catalog=None)   Tier 1 - the card, preloaded
    lookup(user_id, terms=[...])         Tier 2 - detail, on demand
    terms_used(user_id, surfaces=[...])  what a completed run actually touched

**Approved entries only.** Candidates are the model's claims; letting them steer
an analysis is precisely what the review step exists to prevent. The dev flag on
`DomainComposer` exists for development and is not reachable from here.

**Nothing in this module calls a model.** It reads rows and renders text.
"""

from __future__ import annotations

import json
import re

from src.knowledge_domain.compose import DomainComposer
from src.knowledge_domain.models import DomainContext, Measure
from src.knowledge_domain.render import render_card, render_lookup
from src.knowledge_ingest.knowledge_store import KnowledgeStore
from src.knowledge_ingest.selector import resolve_scope_id
from src.middlewares.logging import get_logger

logger = get_logger("knowledge_service")

# Approved knowledge changes at EXPERT speed - a few times a week - not per
# request, so the card is worth caching. One hour matches the chat intent's TTL
# (the 2026-06-11 decision that Redis is short-lived caching only), and a review
# write invalidates the scope immediately, so the TTL is a backstop rather than
# the correctness mechanism.
CACHE_TTL_SECONDS = 3600


def _cache_key(scope_id: str) -> str:
    from src.config.settings import settings

    # `redis_prefix` is in the key for the same reason it is in the chat key
    # (F-16): two environments sharing one Redis must not cross-serve.
    return f"{settings.redis_prefix}knowledge:card:{scope_id}"


class KnowledgeService:
    """Reads domain knowledge for one caller. Never writes."""

    def __init__(
        self,
        store: KnowledgeStore | None = None,
        composer: DomainComposer | None = None,
    ) -> None:
        self._store = store or KnowledgeStore()
        self._composer = composer or DomainComposer(self._store)

    # ── K4: scope resolution, in one place ──────────────────────────────
    async def scope_for(self, user_id: str) -> str:
        """The caller's company.

        Resolved HERE and nowhere else on the read path. Entries are
        company-scoped while the chat pipeline is analysis-scoped, so something
        has to bridge the two; doing it once per turn at the composition root
        and threading it is the same shape the catalog already uses for the
        traceability scratchpad (Q10) - not a fresh lookup inside each prompt
        builder.
        """
        return await resolve_scope_id(user_id)

    # ── K1 + K2 + K3: the Tier-1 card ───────────────────────────────────
    async def get_context(
        self, user_id: str, *, catalog=None, use_cache: bool = True
    ) -> DomainContext:
        """The composed domain context for this caller's company.

        Cached per scope. `catalog` is deliberately NOT part of the cache key:
        bindings are resolved fresh against whatever catalog the caller has, so
        a cached context never carries another analysis's column ids.
        """
        scope_id = await self.scope_for(user_id)
        cached = await self._cached(scope_id) if use_cache else None
        if cached is not None:
            context = cached
        else:
            context = await self._composer.compose(scope_id)
            if use_cache:
                await self._cache(scope_id, context)

        if catalog is not None:
            # Read-time and per-caller, so it stays outside the cache.
            self._composer._bind_to_catalog(context.measures, catalog)
        return context

    async def render_context(self, user_id: str, *, catalog=None) -> str:
        """The Tier-1 card as prompt text. Empty string when there is nothing.

        A caller can append this unconditionally: an empty domain renders to
        nothing rather than to a header implying knowledge that does not exist.
        """
        return render_card(await self.get_context(user_id, catalog=catalog))

    # ── Tier 2 ──────────────────────────────────────────────────────────
    async def lookup(self, user_id: str, terms: list[str]) -> list[Measure]:
        """Everything known about specific terms, matched on any surface form.

        The tier that scales: a corpus of thousands of terms costs nothing until
        a question actually touches one.
        """
        if not terms:
            return []
        context = await self.get_context(user_id)
        wanted = {t.strip().casefold() for t in terms if t and t.strip()}
        hits = [
            m
            for m in context.measures
            if m.name.casefold() in wanted
            or any(f.casefold() in wanted for f in m.surface_forms)
            or m.key.casefold() in wanted
        ]
        logger.info(
            "knowledge_lookup",
            extra={"asked": len(wanted), "found": len(hits)},
        )
        return hits

    async def render_lookup(self, user_id: str, terms: list[str]) -> str:
        """Tier 2 as prompt text, with the governing rules those terms carry."""
        measures = await self.lookup(user_id, terms)
        if not measures:
            return render_lookup([], [])

        scope_id = await self.scope_for(user_id)
        wanted_rules = {r for m in measures for r in m.governed_by}
        rules: list[dict] = []
        if wanted_rules:
            rows = await self._store.list_entries(
                scope_id=scope_id, kind="rule", limit=1000
            )
            rules = [r.payload for r in rows if r.entity_id in wanted_rules]
        return render_lookup(measures, rules)

    # ── what a completed run touched ────────────────────────────────────
    async def terms_used(self, user_id: str, surfaces: list[str]) -> list[Measure]:
        """Measures matching text a finished run actually produced.

        For the assembler and the report: it needs the company's wording for the
        terms in THIS answer, not the whole vocabulary. Whole-word matching
        because the caller passes prose, not a term list.
        """
        if not surfaces:
            return []
        context = await self.get_context(user_id)
        haystack = " \n ".join(s for s in surfaces if s).casefold()

        def _mentions(form: str) -> bool:
            """Whole-word match, so short abbreviations work without false hits.

            A plain substring test had to exclude forms of two characters or
            fewer to avoid matching inside other words — which silently excluded
            PA, UA and Qty, i.e. most of a mining vocabulary. The first live run
            showed exactly that: the planner card fired and the assembler's
            terminology block never did. A word boundary is the right
            instrument — it catches "PA" in prose and not the "pa" in "company".
            """
            form = form.strip().casefold()
            if not form:
                return False
            return re.search(rf"(?<!\w){re.escape(form)}(?!\w)", haystack) is not None

        return [
            m for m in context.measures if any(_mentions(f) for f in m.surface_forms)
        ]

    # ── V4-9/V4-10: the declared half ───────────────────────────────────
    async def propose_declaration(self, user_id: str) -> dict:
        """Draft this company's domain declaration and store it for review.

        Derived from what the corpus demonstrably contains - **no model is
        called**. The expert then approves it, or corrects it through
        `POST /entries/{entity_id}/reviews` with `decision="edited"`, which is
        the same route every other entry uses.

        Nothing here reaches a planner until it is approved: `compose` reads a
        declaration only when it carries an approving review.
        """
        from src.knowledge_domain.declare import propose_declaration

        scope_id = await self.scope_for(user_id)
        # Read the corpus WITH unapproved entries: a draft describes what has
        # been extracted, and refusing to describe an unreviewed corpus would
        # mean the declaration could never be proposed in the first place.
        context = await self._composer.compose(scope_id, include_unapproved=True)
        payload = propose_declaration(
            scope_id,
            covered_subdomains=context.identity.subdomains,
            document_titles=[s.title for s in context.authority.sources if s.title],
            units=[u.symbol for u in context.conventions.units],
        )
        await self._store.save_declaration(
            scope_id=scope_id, user_id=user_id, payload=payload
        )
        await self.invalidate(scope_id)
        return payload

    async def get_declaration(self, user_id: str) -> dict | None:
        """The standing declaration, whether or not an expert has ruled on it."""
        row = await self._store.get_declaration(await self.scope_for(user_id))
        return row.payload if row else None

    # ── K3: cache ───────────────────────────────────────────────────────
    async def invalidate(self, scope_id: str) -> None:
        """Drop a scope's cached card. Called after any review write.

        The review IS the invalidation event: an approval changes what the
        planner should see, and waiting an hour to honour an expert's decision
        would make the loop feel broken.
        """
        try:
            from src.db.redis.connection import get_redis

            redis = await get_redis()
            await redis.delete(_cache_key(scope_id))
            logger.info("knowledge_card_invalidated", extra={"scope_id": scope_id})
        except Exception as e:
            # Never-throw HERE is right, unlike the ingest store: a failed
            # invalidation serves a stale card for at most the TTL, whereas
            # raising would fail the expert's review write itself.
            logger.warning(
                "knowledge_card_invalidate_failed",
                extra={"scope_id": scope_id, "error": repr(e),
                       "degraded_seam": "knowledge_card_invalidate"},
            )

    async def _cached(self, scope_id: str) -> DomainContext | None:
        try:
            from src.db.redis.connection import get_redis

            redis = await get_redis()
            raw = await redis.get(_cache_key(scope_id))
            return DomainContext.model_validate_json(raw) if raw else None
        except Exception as e:
            # A cache miss and a broken cache must behave identically: compose.
            logger.warning(
                "knowledge_card_cache_read_failed",
                extra={"scope_id": scope_id, "error": repr(e),
                       "degraded_seam": "knowledge_card_cache"},
            )
            return None

    async def _cache(self, scope_id: str, context: DomainContext) -> None:
        try:
            from src.db.redis.connection import get_redis

            redis = await get_redis()
            # Stored WITHOUT catalog bindings - those are per-caller and are
            # re-resolved on read, so a cached card can never leak one
            # analysis's column ids into another's prompt.
            plain = context.model_copy(deep=True)
            for measure in plain.measures:
                measure.catalog_binding = None
            await redis.setex(
                _cache_key(scope_id),
                CACHE_TTL_SECONDS,
                json.dumps(plain.model_dump(mode="json")),
            )
        except Exception as e:
            logger.warning(
                "knowledge_card_cache_write_failed",
                extra={"scope_id": scope_id, "error": repr(e),
                       "degraded_seam": "knowledge_card_cache"},
            )