"""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"(? 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"}, )