feat(kb): multi-topic foundation — kb_topic table, topic-scoped endpoints

Phase 1 of the multi-domain refactor. Adds a kb_topic table seeded with
the existing 'ביטוח לאומי' domain (id=1) and pins all 6 existing
kb_source rows to it, so search/ask can be scoped per domain without
disturbing the current single-topic UX.

- scripts/migrations/001_topics.sql: idempotent migration. Creates
  kb_topic, seeds national-insurance with the system_prompt_addendum
  copied verbatim from kb_public.py's hardcoded text, adds nullable
  kb_source.topic_id, backfills it to 1, then SET NOT NULL. GRANTs to
  shira_kb mirror its existing kb_source privileges. Wrapped in BEGIN
  /COMMIT with a sanity-check that raises if the backfill is incomplete.
- api/services/kb/topics.py: list_topics / get_topic / resolve_topic_id.
  resolve_topic_id raises ValueError on unknown ids so the route layer
  can return 400 instead of silently mixing domains.
- api/services/kb/search.py: search() + _retrieve_rrf accept topic_id;
  the SQL adds AND s.topic_id = $N when present. _expand_query is
  parametrized by topic name (was hardcoded "הביטוח הלאומי").
- api/routes/kb_public.py: new GET /kb/topics. /kb/search /kb/sources
  /kb/ask /kb/ask/stream all accept topic_id and call resolve_topic_id.
  _build_ask_runner_context loads system_prompt_addendum from the DB
  instead of the hardcoded insurance string.
- mcp_server/tools/legal_kb_tools.py: tool renamed
  search_insurance_kb → search_legal_kb, takes topic_id at registration
  so each conversation is pinned to its caller-chosen domain. Tool
  description interpolates the topic name.
- api/services/agent_runner.py: kb_topic_id + kb_topic_name flow
  through to register_legal_kb_tools. Hebrew progress labels updated.

Refs Task Master #12 (espocrm-extensions/KnowledgeBase).

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

api/routes/kb_public.py

@@ -18,6 +18,7 @@ from pydantic import BaseModel, Field
 from api.services.kb import search as kb_search
 from api.services.kb import ingest as kb_ingest
 from api.services.kb import s3 as kb_s3
+from api.services.kb import topics as kb_topics
 
 logger = logging.getLogger("shira.kb.public")
 
@@ -41,17 +42,35 @@ class SearchRequest(BaseModel):
     # for each, merge, then rerank against the original query. Recovers
     # documents that don't share the user's exact wording.
     expand: bool = True
+    # Domain scope. None means "lowest active topic" (back-compat — equivalent
+    # to topic_id=1 'national-insurance' until additional topics are added).
+    topic_id: int | None = None
+
+
+@router.get("/topics")
+async def list_topics(request: Request):
+    """Return the active KB topics for the topic <select> in the UI."""
+    _verify_auth(request)
+    topics = await kb_topics.list_topics(active_only=True)
+    return {"topics": topics, "count": len(topics)}
 
 
 @router.post("/search")
 async def search(body: SearchRequest, request: Request):
     """Hybrid search + rerank. Returns ranked chunks with citations."""
     _verify_auth(request)
+    try:
+        topic_id = await kb_topics.resolve_topic_id(body.topic_id)
+    except ValueError as e:
+        raise HTTPException(status_code=400, detail=str(e))
+    topic = await kb_topics.get_topic(topic_id)
     hits = await kb_search.search(
         query=body.query,
         kind=body.kind,
         top_k=body.top_k,
         expand=body.expand,
+        topic_id=topic_id,
+        topic_name=(topic or {}).get("name"),
     )
     # Dates come back as datetime.date; serialize as ISO strings for the UI.
     serialized = []
@@ -65,21 +84,27 @@ async def search(body: SearchRequest, request: Request):
 
 
 @router.get("/sources")
-async def list_sources(request: Request):
+async def list_sources(request: Request, topic_id: int | None = None):
     """Browse mode: list all active sources with basic metadata."""
     _verify_auth(request)
+    try:
+        topic_id = await kb_topics.resolve_topic_id(topic_id)
+    except ValueError as e:
+        raise HTTPException(status_code=400, detail=str(e))
     from api.services.kb.db import get_pool
     pool = await get_pool()
     async with pool.acquire() as conn:
         rows = await conn.fetch(
             """
             SELECT s.id, s.kind, s.title, s.identifier, s.published_at,
-                   s.effective_at, s.source_url,
+                   s.effective_at, s.source_url, s.topic_id,
                    (SELECT COUNT(*) FROM kb_chunk c WHERE c.source_id = s.id) AS chunk_count
             FROM kb_source s
             WHERE s.superseded_by IS NULL
+              AND s.topic_id = $1
             ORDER BY s.kind, s.published_at DESC NULLS LAST, s.title
-            """
+            """,
+            topic_id,
         )
     sources = []
     for r in rows:
@@ -231,14 +256,25 @@ class AskRequest(BaseModel):
     message: str = Field(..., min_length=1, max_length=2000)
     conversation_id: str | None = None
     conversation_history: list[dict] | None = None
+    # Domain scope. None means "lowest active topic" (back-compat).
+    topic_id: int | None = None
 
 
-def _build_ask_runner_context() -> tuple:
-    """Common construction shared by /kb/ask and /kb/ask/stream."""
+async def _build_ask_runner_context(topic_id: int | None) -> tuple:
+    """Common construction shared by /kb/ask and /kb/ask/stream.
+
+    Resolves the topic and pulls its system_prompt_addendum from the DB,
+    so adding a new domain (Phase 4) automatically updates Shira's
+    instructions without a code change. Returns the resolved topic too
+    so the caller can pin tools (search_legal_kb) to it.
+    """
     from api.services.agent_runner import AgentRunner
     from api.services.prompt_builder import build_office_prompt
     from api.services.skills import list_skills
 
+    resolved_id = await kb_topics.resolve_topic_id(topic_id)
+    topic = await kb_topics.get_topic(resolved_id) or {}
+
     runner = AgentRunner(
         ai_gateway_url=os.environ.get("AI_GATEWAY_URL", "http://localhost:3000"),
         ai_gateway_api_key=os.environ.get("AI_GATEWAY_API_KEY", ""),
@@ -252,16 +288,10 @@ def _build_ask_runner_context() -> tuple:
         user_profile="",
         available_skills=list_skills(),
     )
-    system_prompt += (
-        "\n\nContext: The user is asking from the Knowledge Base UI of the "
-        "Israeli National Insurance firm. Every question should be answered "
-        "from the KB of חוק/תקנות/חוזרי הביטוח הלאומי. Before answering, "
-        "call search_insurance_kb to retrieve the relevant sections, and "
-        "cite them explicitly with section_ref (ס' N, תקנה N, or חוזר N/YYYY). "
-        "Never answer from generic legal training if the KB has a matching "
-        "section — if no KB section matches, say so explicitly."
-    )
-    return runner, context, system_prompt
+    addendum = (topic.get("system_prompt_addendum") or "").strip()
+    if addendum:
+        system_prompt += "\n\n" + addendum
+    return runner, context, system_prompt, topic
 
 
 def _build_ask_messages(body: AskRequest) -> list[dict]:
@@ -286,14 +316,17 @@ def _filter_pdf_sources(sources_used: list[dict]) -> list[dict]:
 @router.post("/ask")
 async def ask(body: AskRequest, request: Request):
     """Full agent loop: the user's question goes to shira, she calls
-    search_insurance_kb as needed and returns a summarized answer.
+    search_legal_kb as needed and returns a summarized answer.
 
     Thin wrapper around the existing smart-assistant chat endpoint so the
     KnowledgeBase UI can expose an ask mode without rebuilding prompt
     plumbing.
     """
     _verify_auth(request)
-    runner, context, system_prompt = _build_ask_runner_context()
+    try:
+        runner, context, system_prompt, topic = await _build_ask_runner_context(body.topic_id)
+    except ValueError as e:
+        raise HTTPException(status_code=400, detail=str(e))
     messages = _build_ask_messages(body)
     sources_used: list[dict] = []
     try:
@@ -305,6 +338,8 @@ async def ask(body: AskRequest, request: Request):
             espocrm_api_key=os.environ.get("ESPOCRM_API_KEY", ""),
             allowed_toolsets=["legal"],
             kb_sources_used=sources_used,
+            kb_topic_id=topic.get("id"),
+            kb_topic_name=topic.get("name"),
         )
     except Exception as e:
         logger.exception("[kb.ask] error")
@@ -329,7 +364,10 @@ async def ask_stream(body: AskRequest, request: Request):
     from fastapi.responses import StreamingResponse
 
     _verify_auth(request)
-    runner, context, system_prompt = _build_ask_runner_context()
+    try:
+        runner, context, system_prompt, topic = await _build_ask_runner_context(body.topic_id)
+    except ValueError as e:
+        raise HTTPException(status_code=400, detail=str(e))
     messages = _build_ask_messages(body)
     sources_used: list[dict] = []
 
@@ -353,6 +391,8 @@ async def ask_stream(body: AskRequest, request: Request):
                 espocrm_api_key=os.environ.get("ESPOCRM_API_KEY", ""),
                 allowed_toolsets=["legal"],
                 kb_sources_used=sources_used,
+                kb_topic_id=topic.get("id"),
+                kb_topic_name=topic.get("name"),
                 on_event=_on_event,
             )
             await queue.put({

api/services/agent_runner.py

@@ -33,7 +33,7 @@ def _thinking_label(iteration: int) -> str:
 
 
 def _tool_label(name: str, args: dict) -> str:
-    if name == "search_insurance_kb":
+    if name == "search_legal_kb":
         q = (args.get("query") or "").strip()
         if q:
             qs = q if len(q) <= 60 else q[:60] + "…"
@@ -51,7 +51,7 @@ def _tool_label(name: str, args: dict) -> str:
 
 
 def _tool_done_label(name: str, result) -> str:
-    if name == "search_insurance_kb":
+    if name == "search_legal_kb":
         # legal_kb_tools formats the result as a Hebrew string starting
         # with a "## נמצאו N קטעים" line. Surface that count.
         s = str(result or "")
@@ -102,6 +102,8 @@ class AgentRunner:
         blocked_tools: set[str] | None = None,
         allowed_toolsets: list[str] | None = None,
         kb_sources_used: list[dict] | None = None,
+        kb_topic_id: int | None = None,
+        kb_topic_name: str | None = None,
         on_event: Callable[[dict], None] | None = None,
     ) -> str:
         """Run a conversation with tool-calling loop. Returns the final text response.
@@ -140,7 +142,12 @@ class AgentRunner:
             register_document_tools(tools_registry, crm, case_id, context)
         if should_register_all or "legal" in (allowed_toolsets or []):
             register_legal_tools(tools_registry, crm, case_id, context)
-            register_legal_kb_tools(tools_registry, sources_used=kb_sources_used)
+            register_legal_kb_tools(
+                tools_registry,
+                sources_used=kb_sources_used,
+                topic_id=kb_topic_id,
+                topic_name=kb_topic_name,
+            )
         if should_register_all or "signature" in (allowed_toolsets or []):
             register_signature_tools(tools_registry, crm, case_id, user_id, context)
 

api/services/kb/search.py

@@ -35,19 +35,28 @@ _EXPANSION_TIMEOUT_S = 12.0
 async def _retrieve_rrf(
     query: str,
     kind: Literal["law", "regulation", "circular", "any"],
+    topic_id: int | None = None,
 ) -> list[dict]:
     """Run vector + lexical retrieval, fuse with RRF. No rerank, no top_k cut.
 
     Each candidate dict carries `chunk_id` (kb_chunk.id) so callers can
     deduplicate when merging results from multiple queries.
+
+    topic_id, when provided, restricts retrieval to a single domain
+    (kb_source.topic_id). When None, all topics are searched — kept for
+    transitional callers; new code should always pass a resolved id.
     """
     [vec] = await voyage.embed([query], input_type="query")
 
-    kind_filter = ""
+    filters: list[str] = []
     params: list = [vec, query, _CANDIDATES_PER_SIDE]
     if kind != "any":
-        kind_filter = "AND s.kind = $4"
         params.append(kind)
+        filters.append(f"AND s.kind = ${len(params)}")
+    if topic_id is not None:
+        params.append(topic_id)
+        filters.append(f"AND s.topic_id = ${len(params)}")
+    extra = "\n        ".join(filters)
 
     sql = f"""
     WITH vec AS (
@@ -56,7 +65,7 @@ async def _retrieve_rrf(
       JOIN kb_source s ON s.id = c.source_id
       WHERE c.embedding IS NOT NULL
         AND s.superseded_by IS NULL
-        {kind_filter}
+        {extra}
       ORDER BY c.embedding <=> $1::vector
       LIMIT $3
     ),
@@ -67,7 +76,7 @@ async def _retrieve_rrf(
            websearch_to_tsquery('simple', $2) AS q
       WHERE c.content_tsv @@ q
         AND s.superseded_by IS NULL
-        {kind_filter}
+        {extra}
       ORDER BY ts_rank_cd(c.content_tsv, q) DESC
       LIMIT $3
     ),
@@ -98,8 +107,7 @@ async def _retrieve_rrf(
 
 
 _EXPANSION_PROMPT = (
-    "אתה עוזר חיפוש בבסיס ידע משפטי של הביטוח הלאומי בישראל "
-    "(החוק, התקנות, ספרי המבחנים והליקויים, וחוזרי המוסד). "
+    "אתה עוזר חיפוש בבסיס ידע משפטי בנושא {topic_name}. "
     "המשתמש שואל שאלה. החזר עד {n} ניסוחים חלופיים שיעזרו לאחזר "
     "את אותו מידע ממסמכים משפטיים בעברית — תרגומים מסגנון שאלתי "
     "לסגנון מסמך, מילים נרדפות (למשל 'תקנה 37' → 'בדיקה מחדש', "
@@ -109,7 +117,7 @@ _EXPANSION_PROMPT = (
 )
 
 
-async def _expand_query(query: str) -> list[str]:
+async def _expand_query(query: str, topic_name: str | None = None) -> list[str]:
     """Ask the LLM for alternative phrasings. Empty list on failure (caller
     just runs the original query)."""
     base = (os.environ.get("AI_GATEWAY_URL") or "http://localhost:3000").rstrip("/")
@@ -117,12 +125,16 @@ async def _expand_query(query: str) -> list[str]:
     if not api_key:
         return []
     client = AsyncOpenAI(base_url=f"{base}/v1", api_key=api_key)
+    prompt = _EXPANSION_PROMPT.format(
+        n=_EXPANSION_VARIANTS,
+        topic_name=topic_name or "כללי",
+    )
     try:
         resp = await asyncio.wait_for(
             client.chat.completions.create(
                 model=os.environ.get("CLAUDE_MODEL", "sonnet"),
                 messages=[
-                    {"role": "system", "content": _EXPANSION_PROMPT.format(n=_EXPANSION_VARIANTS)},
+                    {"role": "system", "content": prompt},
                     {"role": "user", "content": query},
                 ],
                 max_tokens=256,
@@ -172,28 +184,30 @@ async def search(
     kind: Literal["law", "regulation", "circular", "any"] = "any",
     top_k: int = 8,
     expand: bool = False,
+    topic_id: int | None = None,
+    topic_name: str | None = None,
 ) -> list[dict]:
     query = (query or "").strip()
     if not query:
         return []
 
     if not expand:
-        candidates = await _retrieve_rrf(query, kind)
+        candidates = await _retrieve_rrf(query, kind, topic_id=topic_id)
         logger.info(
-            "[kb.search] query=%r kind=%s RRF_candidates=%d",
-            query[:80], kind, len(candidates),
+            "[kb.search] query=%r kind=%s topic_id=%s RRF_candidates=%d",
+            query[:80], kind, topic_id, len(candidates),
         )
         if not candidates:
             return []
         return await _rerank_or_truncate(query, candidates, top_k)
 
     # Multi-query expansion: variants run in parallel with the original.
-    variants = await _expand_query(query)
+    variants = await _expand_query(query, topic_name=topic_name)
     queries = [query] + variants
-    logger.info("[kb.search] expand=true variants=%d", len(variants))
+    logger.info("[kb.search] expand=true topic_id=%s variants=%d", topic_id, len(variants))
 
     results = await asyncio.gather(
-        *[_retrieve_rrf(q, kind) for q in queries],
+        *[_retrieve_rrf(q, kind, topic_id=topic_id) for q in queries],
         return_exceptions=True,
     )
     merged: dict[int, dict] = {}

api/services/kb/topics.py

@@ -0,0 +1,73 @@
+"""kb_topic helpers — fetch topics for the public listing endpoint and
+load a topic's full row (including system_prompt_addendum) for /kb/ask.
+
+The DB schema is set up by scripts/migrations/001_topics.sql. As of Phase
+1 there is exactly one seeded topic (id=1, slug='national-insurance').
+"""
+
+from __future__ import annotations
+
+import logging
+from typing import Optional
+
+from api.services.kb.db import get_pool
+
+logger = logging.getLogger("shira.kb.topics")
+
+
+async def list_topics(active_only: bool = True) -> list[dict]:
+    """Return topics ordered by id. Public payload — never includes the
+    system_prompt_addendum (admin-only data, not surfaced to end users).
+    """
+    pool = await get_pool()
+    where = "WHERE is_active = TRUE" if active_only else ""
+    async with pool.acquire() as conn:
+        rows = await conn.fetch(
+            f"""
+            SELECT id, slug, name, description, is_active
+            FROM kb_topic
+            {where}
+            ORDER BY id
+            """
+        )
+    return [dict(r) for r in rows]
+
+
+async def get_topic(topic_id: int) -> Optional[dict]:
+    """Fetch a single topic by id, including system_prompt_addendum."""
+    pool = await get_pool()
+    async with pool.acquire() as conn:
+        row = await conn.fetchrow(
+            """
+            SELECT id, slug, name, description, system_prompt_addendum,
+                   is_active, created_at, updated_at
+            FROM kb_topic
+            WHERE id = $1
+            """,
+            topic_id,
+        )
+    return dict(row) if row else None
+
+
+async def resolve_topic_id(topic_id: Optional[int]) -> int:
+    """Validate a caller-supplied topic_id, or fall back to the lowest
+    active topic. Raises ValueError if the caller supplied an unknown or
+    inactive id, so the route can return 400 instead of silently mixing
+    domains.
+    """
+    pool = await get_pool()
+    async with pool.acquire() as conn:
+        if topic_id is not None:
+            row = await conn.fetchrow(
+                "SELECT id FROM kb_topic WHERE id = $1 AND is_active = TRUE",
+                topic_id,
+            )
+            if not row:
+                raise ValueError(f"unknown or inactive topic_id: {topic_id}")
+            return topic_id
+        row = await conn.fetchrow(
+            "SELECT id FROM kb_topic WHERE is_active = TRUE ORDER BY id LIMIT 1"
+        )
+        if not row:
+            raise ValueError("no active topics in kb_topic")
+        return row["id"]