diff --git a/hosts/piha/runtime/kb-query/docker-compose.override.yml b/hosts/piha/runtime/kb-query/docker-compose.override.yml new file mode 100644 index 0000000..c8de8a2 --- /dev/null +++ b/hosts/piha/runtime/kb-query/docker-compose.override.yml @@ -0,0 +1,10 @@ +# PIHA-specific overrides for kb-query (KB module 5, phase 4). +# +# FastAPI + uvicorn, stateless (no local DB/queue) -- same weight class as +# llm-gateway (~256m). Kept off oom_score_adj: -900 -- that's reserved for +# control-plane/agent processes that must never be an OOM victim; kb-query is +# a search API, restart-on-OOM (default cgroup behaviour) is an acceptable +# failure mode, unlike for the agents that watch the fleet. +services: + kb-query: + mem_limit: 256m diff --git a/hosts/piha/services.yaml b/hosts/piha/services.yaml index d9d3987..825a589 100644 --- a/hosts/piha/services.yaml +++ b/hosts/piha/services.yaml @@ -104,3 +104,20 @@ services: config_path: /opt/homelab/config/kb-postgres # data is in Docker named volume kb_postgres_data — must land on the NVMe # (Docker data-root on /home); see hosts/piha/runtime/kb-postgres override. + + kb-query: + role: kb-search-api # module 5 phase 4: FastAPI wrapper over kb_retrieval cascade/flat query + deployment_model: docker-compose + exposure: private # LAN bind (LAN_BIND_IP); npm@PIHA vhost + OIDC is a later step + offline_required: false + depends_on: + local: [kb-postgres] + external: [ollama] # SOLARIA may be offline -> /search 502/503, /healthz stays ok + ports: + - name: http + container_port: 8080 + host_port: 8230 + protocol: tcp + runtime: + # .env (KB_DSN, LAN_BIND_IP) lives alongside the compose file; stateless, no data path + config_path: services/kb-query diff --git a/scripts/deploy/deploy.sh b/scripts/deploy/deploy.sh index 68079c2..9bf5bef 100755 --- a/scripts/deploy/deploy.sh +++ b/scripts/deploy/deploy.sh @@ -164,7 +164,14 @@ elif isinstance(svcs, list): fi echo "--- docker build: ${svc} ---" - if ! docker build --quiet "${svc_dir}" >/dev/null; then + # Via `docker compose build` (not a raw `docker build "${svc_dir}"`) so a service + # whose compose file declares a wider `build.context` (e.g. kb-query's repo-root + # context, needed to COPY packages/kb-retrieval/ per CLAUDE.md's packages/ Dockerfile + # convention) resolves the same way here as it does at real deploy time + # (deploy-node.sh's `docker compose ... up -d --build`). For a plain `build: .` + # service this is equivalent to the old `docker build "${svc_dir}"` — context still + # defaults to the compose file's own directory. + if ! docker compose -f "${svc_dir}/docker-compose.yml" build --quiet >/dev/null; then echo "GATE FAIL: docker build failed for ${svc}" >&2 gate_failed=true fi diff --git a/services/kb-query/Dockerfile b/services/kb-query/Dockerfile new file mode 100644 index 0000000..c1c9e5b --- /dev/null +++ b/services/kb-query/Dockerfile @@ -0,0 +1,19 @@ +FROM python:3.12-slim + +ENV PYTHONDONTWRITEBYTECODE=1 \ + PYTHONUNBUFFERED=1 + +WORKDIR /app +COPY services/kb-query/requirements.txt . +RUN pip install --no-cache-dir -r requirements.txt + +# kb_retrieval only -- kb-query never writes/archives mail (kb-mail) and never touches the +# documents-ingest CLI/anthropic dependency (module 5 phase 4 plan §3 decision 1: minimal deps). +# Build context is the repo root (see docker-compose.yml's build.context) so this path resolves. +COPY packages/kb-retrieval/ /packages/kb-retrieval/ +RUN pip install --no-cache-dir /packages/kb-retrieval/ + +COPY services/kb-query/app ./app + +EXPOSE 8080 +CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "8080"] diff --git a/services/kb-query/Dockerfile.dockerignore b/services/kb-query/Dockerfile.dockerignore new file mode 100644 index 0000000..74a0c0c --- /dev/null +++ b/services/kb-query/Dockerfile.dockerignore @@ -0,0 +1,6 @@ +__pycache__/ +*.pyc +.venv/ +.git/ +tests/ +.env diff --git a/services/kb-query/README.md b/services/kb-query/README.md new file mode 100644 index 0000000..44aa6a0 --- /dev/null +++ b/services/kb-query/README.md @@ -0,0 +1,85 @@ +# kb-query + +FastAPI search API in front of the module-5 KB retrieval engine +(`packages/kb-retrieval/`). Runs on **PIHA**, bound to PIHA's LAN IP only +(`exposure: private`, same class as paperless/nextcloud — no public ingress +yet). This is a **search API, not chat**: no answer synthesis over results, +that's phase 5. + +## Endpoints + +| Endpoint | Method | Purpose | +|---|---|---| +| `/healthz` | GET | `{"status": "ok", "sol_status": "up"\|"down"}` — `sol_status` is a live probe of Ollama@SOLARIA, no auth required (monitoring must reach it) | +| `/search?q=&mode=cascade\|flat` | GET | `query_text -> embed -> cascade_query/flat_query -> results`, `mode` defaults to `cascade` | + +`/search` response shape (module 5 phase 4 plan §4): + +```json +{ + "query": "...", "mode": "cascade", "sol_status": "up", + "results": [ + {"envelope_id": "paperless:119", "source": "paperless", "dist": 0.34, + "chunk_index": 2, "text": "...", "link": "https://paper.kapala.org/documents/119/details"}, + {"envelope_id": "", "source": "gmail", "dist": 0.44, + "chunk_index": 0, "text": "...", "subject": "...", "from": "...", "date": "...", + "link": null, "mail_ui_url": null} + ] +} +``` + +`dist` is never filtered server-side — the 0.45/0.55 colour thresholds are a +frontend concern (a later step), not an API contract. + +## Embed path (current step) + +Calls Ollama on SOLARIA directly per request — no cache, no circuit breaker, +no local-PIHA fallback yet (that state machine, plan §2 decision 2/§5, is a +separate later step). If SOLARIA is unreachable, `/search` returns **503**; +`/healthz` still answers (`sol_status: "down"`), same tolerance pattern as +`llm-gateway`. + +## Startup invariant (hard-fail) + +At startup, kb-query queries `document_chunk.model` and +`document_summary.embedding_model` for the set of models behind *active* +embeddings, and refuses to start (crash-loop, visible via container restarts) +if the configured `EMBED_MODEL` (default `bge-m3`) isn't in both sets. This +guards against querying with an embedding space that doesn't match what's +actually indexed — see `app/startup.py` for why the check reads +`document_summary.embedding_model` and not `.model` (the latter is the LLM +that *wrote* the summary, e.g. `claude-haiku-4-5`, not the embedder). + +## Configuration + +`.env` — **gitignored**, copy from `env.example`. Required: `LAN_BIND_IP`, +`KB_DSN`. Optional: `OLLAMA_URL`, `EMBED_MODEL`, `SUMMARY_MODEL`. + +## Deploy (PIHA) + +1. `git pull` on PIHA (`~/homelab-codex-ws`). +2. `cp services/kb-query/env.example services/kb-query/.env` and fill in the + real `KB_DSN` password. +3. ``` + docker compose -f services/kb-query/docker-compose.yml \ + -f hosts/piha/runtime/kb-query/docker-compose.override.yml up -d --build + ``` +4. Verify: `services/kb-query/healthcheck.sh`, then from PIHA: + `curl "http://192.168.31.5:8230/search?q=test"`. + +## Tests + +``` +pip install -e packages/kb-retrieval/ +cd services/kb-query && pytest +``` + +Unit tests mock the DB connection and Ollama HTTP session (no live DB/Ollama +required) — same style as `packages/kb-retrieval/tests/`. + +## Out of scope for this step + +- Local-PIHA embed fallback / circuit breaker (plan §2 decision 2, §5). +- npm@PIHA vhost, OIDC login, DNS (plan §8) — `kb.kapala.org` is not wired up + yet; reach the API directly over LAN/Tailscale for now. +- Frontend (plan §7) — `/search` returns bare JSON. diff --git a/services/kb-query/app/db.py b/services/kb-query/app/db.py new file mode 100644 index 0000000..fb011d3 --- /dev/null +++ b/services/kb-query/app/db.py @@ -0,0 +1,36 @@ +"""kb-postgres access for kb-query beyond what `kb_retrieval.retrieval` covers -- the envelope +join (`source` + `entities` for link/metadata building, app/links.py) is out of scope for the +shared retrieval package (module 5 phase 4 plan §4: "to jest nowy kod w warstwie +HTTP-handlera kb-query, nie zmiana w kb_retrieval").""" +from __future__ import annotations + +import json + +import asyncpg + + +async def create_pool(dsn: str) -> asyncpg.Pool: + return await asyncpg.create_pool(dsn, min_size=1, max_size=5) + + +def _decode_jsonb(value: object) -> object: + """asyncpg may return jsonb as a str or an already-decoded object.""" + if value is None: + return None + if isinstance(value, str): + return json.loads(value) + return value + + +async def fetch_envelopes(conn: asyncpg.Connection, envelope_ids: list[str]) -> dict[str, dict]: + """Batch-fetch `source`/`entities` for a set of chunk hits, keyed by envelope id.""" + if not envelope_ids: + return {} + rows = await conn.fetch( + "SELECT id, source, entities FROM envelope WHERE id = ANY($1::text[])", + envelope_ids, + ) + return { + r["id"]: {"source": r["source"], "entities": _decode_jsonb(r["entities"]) or []} + for r in rows + } diff --git a/services/kb-query/app/links.py b/services/kb-query/app/links.py new file mode 100644 index 0000000..4d27642 --- /dev/null +++ b/services/kb-query/app/links.py @@ -0,0 +1,72 @@ +"""Per-source result shaping for `/search` -- module 5 phase 4 (docs/kb/modules/05-faza4-plan.md, +§4 response shape, §6/decision 3 links). `kb_retrieval.retrieval` only ever touches +`document_chunk`/`document_summary`; the join to `envelope` (for `source` and link/metadata +fields) is new HTTP-layer code that belongs to kb-query, not the shared retrieval package. + +Paperless link format verified live 2026-07-22: `curl -I https://paper.kapala.org/documents/1/details` +-> 302 to `/accounts/login/?next=/documents/1/details` (unauthenticated, but confirms the route +exists and is handled by the Angular app, not a 404) -- the plan's candidate format. + +Gmail has nothing to link to (plan §2 decision 3: `envelope_id` **is** the Message-ID already, +nothing to extract) -- `mail_ui_url` stays a reserved `null` until kb-00's future mail-UI. +""" +from __future__ import annotations + +from typing import Optional + +PAPERLESS_LINK_TEMPLATE = "https://paper.kapala.org/documents/{doc_id}/details" + + +def _extract_headers(entities: list) -> dict: + for entity in entities or []: + if isinstance(entity, dict) and entity.get("type") == "headers": + return entity + return {} + + +def _format_from(headers: dict) -> Optional[str]: + frm = headers.get("from") + if not isinstance(frm, dict): + return None + name = frm.get("name") + address = frm.get("address") + if name and address: + return f"{name} <{address}>" + return address or name + + +def _paperless_link(envelope_id: str) -> str: + # envelope_id is "paperless:" (paperless_adapter.py:126) -- the raw Paperless doc id + # is everything after the first colon. + doc_id = envelope_id.split(":", 1)[1] if ":" in envelope_id else envelope_id + return PAPERLESS_LINK_TEMPLATE.format(doc_id=doc_id) + + +def build_result(chunk: dict, envelope: Optional[dict]) -> dict: + """One `/search` result item: retrieval hit (`chunk`) + its envelope metadata (`envelope`, + `None` if the envelope vanished between chunk-embedding and this query -- shouldn't happen + on a read-only corpus, but the join must not crash on it).""" + source = (envelope or {}).get("source", "unknown") + result = { + "envelope_id": chunk["envelope_id"], + "source": source, + "dist": chunk["dist"], + "chunk_index": chunk["chunk_index"], + "text": chunk["text"], + } + + if source == "paperless": + result["link"] = _paperless_link(chunk["envelope_id"]) + return result + + if source == "gmail": + headers = _extract_headers((envelope or {}).get("entities", [])) + result["subject"] = headers.get("subject") + result["from"] = _format_from(headers) + result["date"] = headers.get("date_raw") + result["link"] = None + result["mail_ui_url"] = None + return result + + result["link"] = None + return result diff --git a/services/kb-query/app/main.py b/services/kb-query/app/main.py new file mode 100644 index 0000000..ee44ffa --- /dev/null +++ b/services/kb-query/app/main.py @@ -0,0 +1,72 @@ +"""kb-query -- module 5 phase 4 (docs/kb/modules/05-faza4-plan.md §4): first user-facing HTTP +entry point to the KB. Wraps `kb_retrieval.cascade_query`/`flat_query` (module 5 phase 3, +already gated PASS -- docs/sessions/2026-07-21.md) in FastAPI. This is a search API, not chat: +no answer synthesis, no LLM call over the results (that is phase 5, out of scope here). + +Embed path is deliberately simple for this step: calls Ollama on SOLARIA directly, no +cache/circuit-breaker/local-PIHA-fallback (plan §2 decision 2, §5) -- that state machine is a +later, separate step. A failed embed call (SOLARIA unreachable) surfaces as 503 to the caller +rather than a bare 500. +""" +from __future__ import annotations + +import os +from contextlib import asynccontextmanager + +import aiohttp +from fastapi import FastAPI, HTTPException, Query + +from kb_retrieval.embed import check_ollama_health + +from app.db import create_pool +from app.search import run_search +from app.startup import validate_embed_model + +KB_DSN = os.environ.get("KB_DSN") +OLLAMA_URL = os.environ.get("OLLAMA_URL", "http://solaria:11434") +EMBED_MODEL = os.environ.get("EMBED_MODEL", "bge-m3") +SUMMARY_MODEL = os.environ.get("SUMMARY_MODEL", "claude-haiku-4-5") +OLLAMA_HEALTH_TIMEOUT_S = 3.0 + + +@asynccontextmanager +async def lifespan(app: FastAPI): + if not KB_DSN: + raise RuntimeError("KB_DSN is required (see env.example)") + + pool = await create_pool(KB_DSN) + async with pool.acquire() as conn: + # Hard invariant (plan §2 decision 2): refuse to start rather than silently serve + # queries against a mismatched embedding space. + await validate_embed_model(conn, EMBED_MODEL) + + app.state.pool = pool + app.state.http = aiohttp.ClientSession() + try: + yield + finally: + await app.state.http.close() + await pool.close() + + +app = FastAPI(lifespan=lifespan) + + +@app.get("/healthz") +async def healthz() -> dict: + sol_up = await check_ollama_health(app.state.http, OLLAMA_URL, OLLAMA_HEALTH_TIMEOUT_S) + return {"status": "ok", "sol_status": "up" if sol_up else "down"} + + +@app.get("/search") +async def search( + q: str = Query(..., min_length=1), + mode: str = Query("cascade", pattern="^(cascade|flat)$"), +) -> dict: + try: + async with app.state.pool.acquire() as conn: + return await run_search( + conn, app.state.http, OLLAMA_URL, q, mode, EMBED_MODEL, SUMMARY_MODEL + ) + except aiohttp.ClientError as exc: + raise HTTPException(status_code=503, detail=f"embed backend unavailable: {exc}") from exc diff --git a/services/kb-query/app/search.py b/services/kb-query/app/search.py new file mode 100644 index 0000000..ff93162 --- /dev/null +++ b/services/kb-query/app/search.py @@ -0,0 +1,48 @@ +"""`/search` core -- module 5 phase 4 (docs/kb/modules/05-faza4-plan.md §4). Kept decoupled +from FastAPI so it can be unit-tested with fake `conn`/`session` objects, the same style as +`kb_retrieval`'s own tests, instead of needing a live DB/Ollama behind a TestClient. + +Response shape (plan §4 exactly): `{"query", "mode", "sol_status", "results": [...]}`, each +result carrying `dist` un-filtered -- the 0.45/0.55 colour thresholds (plan §7) are a frontend +concern (Krok 4, out of this step's scope), never applied server-side. +""" +from __future__ import annotations + +import aiohttp +import asyncpg + +from kb_retrieval.retrieval import cascade_query, flat_query + +from app.db import fetch_envelopes +from app.links import build_result + + +async def run_search( + conn: asyncpg.Connection, + session: aiohttp.ClientSession, + ollama_url: str, + query_text: str, + mode: str, + embed_model: str, + summary_model: str, +) -> dict: + if mode == "flat": + retrieval = await flat_query(conn, session, ollama_url, query_text, embed_model=embed_model) + else: + retrieval = await cascade_query( + conn, session, ollama_url, query_text, + summary_model=summary_model, embed_model=embed_model, + ) + + chunks = retrieval["chunks"] + envelope_ids = sorted({c["envelope_id"] for c in chunks}) + envelopes = await fetch_envelopes(conn, envelope_ids) + + results = [build_result(chunk, envelopes.get(chunk["envelope_id"])) for chunk in chunks] + + return { + "query": query_text, + "mode": mode, + "sol_status": "up", # reaching this point means the embed call above succeeded + "results": results, + } diff --git a/services/kb-query/app/startup.py b/services/kb-query/app/startup.py new file mode 100644 index 0000000..0189ccc --- /dev/null +++ b/services/kb-query/app/startup.py @@ -0,0 +1,46 @@ +"""Startup invariant -- module 5 phase 4 (docs/kb/modules/05-faza4-plan.md, §2 decision 2, +"twardy inwariant"): the configured `EMBED_MODEL` must already be the model behind the active +embeddings in both `document_chunk` and `document_summary`, or kb-query refuses to start +(crash-loop, visible via container restarts in monitoring -- deliberately loud, never a silent +mismatch). There is no per-request model choice today, so this is the only place drift could +sneak in (someone changes `EMBED_MODEL` without a re-index). + +`document_summary.model` is the LLM that WROTE the summary (`claude-haiku-4-5` / `gemma3:12b` +-- see `services/kb-postgres/init/004_summaries.sql`), not the embedder; the column that +records which model embedded the summary text is `embedding_model`. The plan's SQL sketch for +this check named `model` for both tables -- checking `document_summary.model` against +`EMBED_MODEL` would never match (summaries are never written by `bge-m3`) and the service would +refuse to start unconditionally. Checked live against kb-postgres@PIHA on 2026-07-22 before +writing this: `document_summary.model` holds `{claude-haiku-4-5, gemma3:12b}`, +`document_summary.embedding_model` holds `{bge-m3}` -- `embedding_model` is the correct column. +""" +from __future__ import annotations + +import asyncpg + + +class ModelInvariantError(RuntimeError): + """The configured EMBED_MODEL is absent from document_chunk/document_summary embeddings.""" + + +async def validate_embed_model(conn: asyncpg.Connection, embed_model: str) -> None: + chunk_rows = await conn.fetch( + "SELECT DISTINCT model FROM document_chunk " + "WHERE excluded_reason IS NULL AND embedding IS NOT NULL" + ) + chunk_models = {r["model"] for r in chunk_rows} + if embed_model not in chunk_models: + raise ModelInvariantError( + f"EMBED_MODEL={embed_model!r} not found among document_chunk.model " + f"of active embeddings ({sorted(chunk_models) or 'none'})" + ) + + summary_rows = await conn.fetch( + "SELECT DISTINCT embedding_model FROM document_summary WHERE embedding IS NOT NULL" + ) + summary_embed_models = {r["embedding_model"] for r in summary_rows} + if embed_model not in summary_embed_models: + raise ModelInvariantError( + f"EMBED_MODEL={embed_model!r} not found among document_summary.embedding_model " + f"of active embeddings ({sorted(summary_embed_models) or 'none'})" + ) diff --git a/services/kb-query/docker-compose.yml b/services/kb-query/docker-compose.yml new file mode 100644 index 0000000..dc524dd --- /dev/null +++ b/services/kb-query/docker-compose.yml @@ -0,0 +1,29 @@ +services: + kb-query: + build: + # Repo-root context (not the usual `.` = this dir) so the Dockerfile can + # `COPY packages/kb-retrieval/` — the packages/ Dockerfile convention from + # CLAUDE.md. scripts/deploy/deploy.sh's gate and deploy-node.sh's + # `docker compose ... up --build` both resolve this the same way. + context: ../.. + dockerfile: services/kb-query/Dockerfile + container_name: kb-query + restart: unless-stopped + ports: + # LAN-only bind, never 0.0.0.0 — same convention as paperless/nextcloud + # (exposure: private). npm@PIHA vhost + OIDC are a later step (plan §8); + # this bind is what that vhost will eventually proxy to. + # Requires .env (from env.example) next to this file at deploy. + - "${LAN_BIND_IP}:8230:8080" + environment: + - KB_DSN=${KB_DSN} + - OLLAMA_URL=${OLLAMA_URL:-http://solaria:11434} + - EMBED_MODEL=${EMBED_MODEL:-bge-m3} + - SUMMARY_MODEL=${SUMMARY_MODEL:-claude-haiku-4-5} + # python:3.12-slim has no curl/wget; same in-container check pattern as llm-gateway. + healthcheck: + test: ["CMD", "python", "-c", "import urllib.request; urllib.request.urlopen('http://127.0.0.1:8080/healthz', timeout=3).read()"] + interval: 30s + timeout: 10s + retries: 5 + start_period: 10s diff --git a/services/kb-query/env.example b/services/kb-query/env.example new file mode 100644 index 0000000..4b23f5d --- /dev/null +++ b/services/kb-query/env.example @@ -0,0 +1,25 @@ +# kb-query secrets + host-local binds — copy to .env (gitignored) next to +# docker-compose.yml and fill in real values. Never commit .env. + +# LAN IP of PIHA. The published port (8230) binds ONLY to this interface — +# never 0.0.0.0. Verify after host rebuilds: ip -4 addr. +LAN_BIND_IP=192.168.31.5 + +# asyncpg DSN for kb-postgres@PIHA. kb-query runs in its own Docker network +# (separate compose project from kb-postgres), so it reaches kb-postgres's +# published port over the host's LAN interface, not "localhost" — same +# reasoning as paperless-worker@SOLARIA reaching paperless@PIHA over LAN. +KB_DSN=postgresql://kb:CHANGE-ME@192.168.31.5:5433/kb + +# Ollama upstream (SOLARIA, over Tailscale MagicDNS — same trick as +# llm-gateway's OLLAMA_URL). Optional: defaults to this value if unset. +# OLLAMA_URL=http://solaria:11434 + +# Embedding model kb-query enforces as a startup invariant (plan §2 decision +# 2) against document_chunk.model / document_summary.embedding_model. +# Optional: defaults to bge-m3. +# EMBED_MODEL=bge-m3 + +# document_summary.model kb-query's cascade path pre-filters on (the +# compilation track, plan §2 D3). Optional: defaults to claude-haiku-4-5. +# SUMMARY_MODEL=claude-haiku-4-5 diff --git a/services/kb-query/healthcheck.sh b/services/kb-query/healthcheck.sh new file mode 100755 index 0000000..0361628 --- /dev/null +++ b/services/kb-query/healthcheck.sh @@ -0,0 +1,27 @@ +#!/bin/bash +# Healthcheck for kb-query (FastAPI search API -> kb-postgres + Ollama @ SOLARIA) + +SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)" + +# The port is bound to the LAN interface only, so localhost won't answer. +# Read the bind IP from .env (same file compose uses for the port mapping). +if [ -f "$SCRIPT_DIR/.env" ]; then + # shellcheck disable=SC1091 + source "$SCRIPT_DIR/.env" +fi +BIND_IP="${LAN_BIND_IP:-127.0.0.1}" + +# Container must be running +if ! docker ps --filter "name=kb-query" --filter "status=running" | grep -qw "kb-query"; then + echo "[FAIL] kb-query container is not running" + exit 1 +fi + +# Health endpoint must answer with the ok marker +if ! curl -sf "http://${BIND_IP}:8230/healthz" | grep -q '"status":"ok"\|"status": "ok"'; then + echo "[FAIL] kb-query is not responding on ${BIND_IP}:8230" + exit 1 +fi + +echo "[OK] kb-query is healthy" +exit 0 diff --git a/services/kb-query/pytest.ini b/services/kb-query/pytest.ini new file mode 100644 index 0000000..78c5011 --- /dev/null +++ b/services/kb-query/pytest.ini @@ -0,0 +1,3 @@ +[pytest] +asyncio_mode = auto +testpaths = tests diff --git a/services/kb-query/requirements.txt b/services/kb-query/requirements.txt new file mode 100644 index 0000000..6ae617e --- /dev/null +++ b/services/kb-query/requirements.txt @@ -0,0 +1,4 @@ +fastapi==0.115.6 +uvicorn[standard]==0.34.0 +aiohttp==3.11.11 +asyncpg==0.30.0 diff --git a/services/kb-query/service.yaml b/services/kb-query/service.yaml new file mode 100644 index 0000000..62236e9 --- /dev/null +++ b/services/kb-query/service.yaml @@ -0,0 +1,33 @@ +service: + name: kb-query + owner_node: piha + role: kb-search-api # module 5 phase 4: first user-facing HTTP entry point to the KB + exposure: private # LAN/Tailscale only, npm@PIHA vhost (kb.kapala.org) is a later step + dependencies: + - kb-postgres + - forgejo # OIDC identity provider, wired in a later step (plan §8); not yet enforced + # ollama@SOLARIA is an external, optional runtime dependency (embed calls), not a hard + # dependency here -- SOLARIA may be offline; /search then fails per-request with 503, + # /healthz still answers (same tolerance pattern as llm-gateway). + ports: + - container: 8080 + host: 8230 # LAN_BIND_IP only, never 0.0.0.0 + protocol: tcp + healthcheck: + type: http + endpoint: http://192.168.31.5:8230/healthz # LAN bind — localhost does not answer + interval: 30s + timeout: 10s + retries: 5 + restart_policy: unless-stopped + persistence: + paths: [] # stateless — kb-postgres holds all state + runtime: + config_files: + - .env # KB_DSN, LAN_BIND_IP (gitignored, from env.example) + env_vars: + - LAN_BIND_IP # required — compose port-bind interpolation + - KB_DSN # required — asyncpg DSN for kb-postgres@PIHA + - OLLAMA_URL # optional — defaults to http://solaria:11434 + - EMBED_MODEL # optional — defaults to bge-m3; startup invariant vs document_chunk/document_summary + - SUMMARY_MODEL # optional — defaults to claude-haiku-4-5; cascade_query's stage-1 model diff --git a/services/kb-query/tests/test_links.py b/services/kb-query/tests/test_links.py new file mode 100644 index 0000000..718f09c --- /dev/null +++ b/services/kb-query/tests/test_links.py @@ -0,0 +1,76 @@ +"""Unit tests for per-source /search result shaping.""" +from __future__ import annotations + +import pathlib +import sys + +sys.path.insert(0, str(pathlib.Path(__file__).resolve().parents[1])) + +from app.links import build_result # noqa: E402 + + +def _chunk(envelope_id="paperless:119", dist=0.34, chunk_index=2, text="hit text"): + return {"envelope_id": envelope_id, "dist": dist, "chunk_index": chunk_index, "text": text} + + +class TestBuildResultPaperless: + def test_shape_and_link(self): + envelope = {"source": "paperless", "entities": []} + result = build_result(_chunk(envelope_id="paperless:119"), envelope) + assert result["source"] == "paperless" + assert result["link"] == "https://paper.kapala.org/documents/119/details" + assert "subject" not in result + assert "mail_ui_url" not in result + + +class TestBuildResultGmail: + def test_shape_and_headers(self): + envelope = { + "source": "gmail", + "entities": [ + { + "type": "headers", + "from": {"name": "Promocja Lunchroom", "address": "promocja@lunchroom.pl"}, + "subject": "-20% w wybranych restauracjach", + "date_raw": "Wed, 9 May 2018 10:03:32 +0200", + } + ], + } + result = build_result(_chunk(envelope_id=""), envelope) + assert result["source"] == "gmail" + assert result["subject"] == "-20% w wybranych restauracjach" + assert result["from"] == "Promocja Lunchroom " + assert result["date"] == "Wed, 9 May 2018 10:03:32 +0200" + assert result["link"] is None + assert result["mail_ui_url"] is None + + def test_from_without_name_falls_back_to_address(self): + envelope = { + "source": "gmail", + "entities": [ + {"type": "headers", "from": {"name": None, "address": "a@b.com"}, "subject": "s", "date_raw": "d"} + ], + } + result = build_result(_chunk(), envelope) + assert result["from"] == "a@b.com" + + def test_missing_headers_entity_yields_none_fields(self): + envelope = {"source": "gmail", "entities": []} + result = build_result(_chunk(), envelope) + assert result["subject"] is None + assert result["from"] is None + assert result["date"] is None + + +class TestBuildResultEdgeCases: + def test_unknown_source_gets_null_link_no_extra_fields(self): + result = build_result(_chunk(), {"source": "mystery", "entities": []}) + assert result["source"] == "mystery" + assert result["link"] is None + assert "subject" not in result + + def test_missing_envelope_does_not_crash(self): + # e.g. a chunk whose envelope vanished between embedding and query time. + result = build_result(_chunk(), None) + assert result["source"] == "unknown" + assert result["link"] is None diff --git a/services/kb-query/tests/test_search.py b/services/kb-query/tests/test_search.py new file mode 100644 index 0000000..73c1e66 --- /dev/null +++ b/services/kb-query/tests/test_search.py @@ -0,0 +1,163 @@ +"""Unit tests for /search's core logic (app.search.run_search) -- no real DB, no real Ollama. +Same mocking style as packages/kb-retrieval/tests/test_retrieval.py, extended with an +`envelope` table fixture for the join app/db.py adds on top of kb_retrieval.""" +from __future__ import annotations + +import pathlib +import sys + +sys.path.insert(0, str(pathlib.Path(__file__).resolve().parents[1])) + +from app.search import run_search # noqa: E402 + + +class _FakeConn: + """summaries: [(envelope_id, dist), ...]. chunks_by_envelope: envelope_id -> [(chunk_index, + text, dist), ...]. envelopes: envelope_id -> {"source": ..., "entities": [...]}.""" + + def __init__(self, summaries=None, chunks_by_envelope=None, envelopes=None): + self._summaries = list(summaries or []) + self._chunks_by_envelope = chunks_by_envelope or {} + self._envelopes = envelopes or {} + + async def fetch(self, query, *params): + if "FROM document_summary" in query: + _, _model, limit = params + return [{"envelope_id": eid, "dist": dist} for eid, dist in self._summaries[:limit]] + if "FROM document_chunk" in query and "= ANY" in query: + _, envelope_ids, limit = params + rows = [ + {"envelope_id": eid, "chunk_index": idx, "text": text, "dist": dist} + for eid in envelope_ids + for idx, text, dist in self._chunks_by_envelope.get(eid, []) + ] + rows.sort(key=lambda r: r["dist"]) + return rows[:limit] + if "FROM document_chunk" in query: # flat path + _, limit = params + rows = [ + {"envelope_id": eid, "chunk_index": idx, "text": text, "dist": dist} + for eid, chunk_list in self._chunks_by_envelope.items() + for idx, text, dist in chunk_list + ] + rows.sort(key=lambda r: r["dist"]) + return rows[:limit] + if "FROM envelope" in query: + (envelope_ids,) = params + return [ + {"id": eid, "source": self._envelopes[eid]["source"], "entities": self._envelopes[eid]["entities"]} + for eid in envelope_ids + if eid in self._envelopes + ] + raise AssertionError(f"unexpected query: {query}") + + +class _FakeEmbedResponse: + def __init__(self, payload): + self._payload = payload + + async def __aenter__(self): + return self + + async def __aexit__(self, *exc): + return False + + def raise_for_status(self): + pass + + async def json(self): + return self._payload + + +class _FakeSession: + def __init__(self): + self.post_calls: list[dict] = [] + + def post(self, url, json): + self.post_calls.append({"url": url, "json": json}) + return _FakeEmbedResponse({"embedding": [0.01] * 1024}) + + +class TestRunSearchHappyPath: + async def test_cascade_hit_joins_envelope_and_shapes_paperless_link(self): + conn = _FakeConn( + summaries=[("paperless:119", 0.1)], + chunks_by_envelope={"paperless:119": [(2, "hit text", 0.34)]}, + envelopes={"paperless:119": {"source": "paperless", "entities": []}}, + ) + session = _FakeSession() + result = await run_search( + conn, session, "http://fake-ollama", "polisa PZU", "cascade", "bge-m3", "claude-haiku-4-5" + ) + assert result["query"] == "polisa PZU" + assert result["mode"] == "cascade" + assert result["sol_status"] == "up" + assert len(result["results"]) == 1 + hit = result["results"][0] + assert hit["envelope_id"] == "paperless:119" + assert hit["source"] == "paperless" + assert hit["dist"] == 0.34 + assert hit["chunk_index"] == 2 + assert hit["text"] == "hit text" + assert hit["link"] == "https://paper.kapala.org/documents/119/details" + + async def test_flat_mode_skips_cascade_stage1(self): + conn = _FakeConn( + chunks_by_envelope={"paperless:1": [(0, "a", 0.2)]}, + envelopes={"paperless:1": {"source": "paperless", "entities": []}}, + ) + session = _FakeSession() + result = await run_search( + conn, session, "http://fake-ollama", "q", "flat", "bge-m3", "claude-haiku-4-5" + ) + assert result["mode"] == "flat" + assert len(result["results"]) == 1 + + async def test_gmail_hit_carries_header_metadata_not_a_link(self): + conn = _FakeConn( + summaries=[("", 0.1)], + chunks_by_envelope={"": [(0, "body text", 0.4)]}, + envelopes={ + "": { + "source": "gmail", + "entities": [ + {"type": "headers", "from": {"name": "A", "address": "a@b.com"}, "subject": "s", "date_raw": "d"} + ], + } + }, + ) + session = _FakeSession() + result = await run_search( + conn, session, "http://fake-ollama", "q", "cascade", "bge-m3", "claude-haiku-4-5" + ) + hit = result["results"][0] + assert hit["source"] == "gmail" + assert hit["subject"] == "s" + assert hit["link"] is None + assert hit["mail_ui_url"] is None + + +class TestRunSearchNoGoodResults: + async def test_results_above_no_answer_threshold_are_still_returned_unfiltered(self): + # Plan §7: the 0.55 "no answer" colour threshold is a frontend concern -- the API + # must not silently drop/hide a poor match, only report its true dist so the caller + # (UI or eval harness) can apply that policy itself. + conn = _FakeConn( + summaries=[("paperless:1", 0.6)], + chunks_by_envelope={"paperless:1": [(0, "unrelated text", 0.62)]}, + envelopes={"paperless:1": {"source": "paperless", "entities": []}}, + ) + session = _FakeSession() + result = await run_search( + conn, session, "http://fake-ollama", "unrelated query", "cascade", "bge-m3", "claude-haiku-4-5" + ) + assert len(result["results"]) == 1 + assert result["results"][0]["dist"] == 0.62 + + async def test_no_summaries_yields_empty_results_not_an_error(self): + conn = _FakeConn(summaries=[], chunks_by_envelope={}, envelopes={}) + session = _FakeSession() + result = await run_search( + conn, session, "http://fake-ollama", "nothing matches", "cascade", "bge-m3", "claude-haiku-4-5" + ) + assert result["results"] == [] diff --git a/services/kb-query/tests/test_startup.py b/services/kb-query/tests/test_startup.py new file mode 100644 index 0000000..424bbd3 --- /dev/null +++ b/services/kb-query/tests/test_startup.py @@ -0,0 +1,56 @@ +"""Unit tests for the startup model invariant -- no real DB.""" +from __future__ import annotations + +import pathlib +import sys + +import pytest + +sys.path.insert(0, str(pathlib.Path(__file__).resolve().parents[1])) + +from app.startup import ModelInvariantError, validate_embed_model # noqa: E402 + + +class _FakeConn: + def __init__(self, chunk_models: set, summary_embedding_models: set): + self._chunk_models = chunk_models + self._summary_embedding_models = summary_embedding_models + + async def fetch(self, query, *params): + if "FROM document_chunk" in query: + return [{"model": m} for m in self._chunk_models] + if "FROM document_summary" in query: + return [{"embedding_model": m} for m in self._summary_embedding_models] + raise AssertionError(f"unexpected query: {query}") + + +class TestValidateEmbedModel: + async def test_passes_when_model_present_in_both_tables(self): + conn = _FakeConn(chunk_models={"bge-m3"}, summary_embedding_models={"bge-m3"}) + await validate_embed_model(conn, "bge-m3") # must not raise + + async def test_fails_when_chunk_model_missing(self): + conn = _FakeConn(chunk_models={"some-other-model"}, summary_embedding_models={"bge-m3"}) + with pytest.raises(ModelInvariantError, match="document_chunk"): + await validate_embed_model(conn, "bge-m3") + + async def test_fails_when_chunk_table_empty(self): + conn = _FakeConn(chunk_models=set(), summary_embedding_models={"bge-m3"}) + with pytest.raises(ModelInvariantError, match="document_chunk"): + await validate_embed_model(conn, "bge-m3") + + async def test_fails_when_summary_embedding_model_missing(self): + conn = _FakeConn(chunk_models={"bge-m3"}, summary_embedding_models={"some-other-model"}) + with pytest.raises(ModelInvariantError, match="document_summary"): + await validate_embed_model(conn, "bge-m3") + + async def test_does_not_confuse_summary_writer_model_with_embedding_model(self): + # Regression guard: document_summary.model is the LLM that WROTE the summary + # (claude-haiku-4-5/gemma3:12b), never bge-m3 -- checking that column instead of + # embedding_model would make this invariant impossible to satisfy. + conn = _FakeConn( + chunk_models={"bge-m3"}, + summary_embedding_models={"bge-m3"}, + ) + assert conn._summary_embedding_models == {"bge-m3"} + await validate_embed_model(conn, "bge-m3") # must not raise