""" TOLARIA SEARCH SERVICE — C4B Core Search API ================================================================ Implementiert exact / keyword / metadata Retrieval mit deterministischem Ranking, Collapse (canonical/source), Current/Historical, Filter, Pagination, Honest Mode, Health und rebuildbarem Keyword/Metadata-Index. Semantic/vector/hybrid sind in C4B BEWUSST nicht implementiert und liefern ehrliche Error/Status laut C4A-Contract. Storage: reine Python-Standardbibliothek, derived rebuildbarer JSON-Index unter data/. Kein pgvector, kein PostgreSQL, keine Trading-/Forgejo-DB (C4D führt den dedizierten isolierten pgvector-Service ein). Dieser Service schreibt NIE nach Tolaria/Forgejo (read-only ggü. Knowledge-Bestand). """ from __future__ import annotations import json import os import re import time from typing import Any, Optional from search_engine import ( Doc, SearchIndex, tokenize, scan_for_secrets, INDEX_VERSION, REPRESENTATION_RANK, STATE_RANK, INDEX_ROLES, INDEX_ROLES_PENALTY, DEFAULT_LIMIT, MAX_LIMIT, ) SOURCE_SYSTEM = "tolaria" SUPPORTED_MODES = ["exact", "keyword", "metadata"] class SearchError(Exception): def __init__(self, code: str, reason: str): self.code = code self.reason = reason def _err(code: str, reason: str) -> dict: return {"error": {"code": code, "reason": reason}} # --------------------------------------------------------------------------- # Query-Parser # --------------------------------------------------------------------------- def _parse_query(raw_query: str) -> tuple[str, list[str]]: """Return (norm_query, tokens). Quoted phrase -> tokens der ersten Phrase.""" raw_query = (raw_query or "").strip() if not raw_query: return "", [] phrases = re.findall(r'"([^"]+)"', raw_query) rest = re.sub(r'"[^"]*"', " ", raw_query) if phrases: return raw_query, tokenize(phrases[0]) return raw_query, tokenize(rest) # --------------------------------------------------------------------------- # Filter-Engine # --------------------------------------------------------------------------- def _as_list(v): if v is None: return [] if isinstance(v, str): return [v] return list(v) def _matches_filters(doc: Doc, filters: dict, include_historical: bool) -> bool: f = filters or {} if _as_list(f.get("type")) and doc.type not in _as_list(f["type"]): return False if _as_list(f.get("role")) and doc.role not in _as_list(f["role"]): return False if _as_list(f.get("representation")) and doc.representation not in _as_list(f["representation"]): return False if _as_list(f.get("object_ids")) and doc.id not in _as_list(f["object_ids"]): return False if f.get("source_paths"): if not any(doc.path.startswith(p) for p in _as_list(f["source_paths"])): return False if _as_list(f.get("tags")): doc_tags = set(doc.tags or []) if not all(t in doc_tags for t in _as_list(f["tags"])): return False if _as_list(f.get("state")) and doc.state not in _as_list(f["state"]): return False if not include_historical: if doc.state in ("historical", "superseded", "archived"): return False return True class TolariaSearch: def __init__(self, index_path: Optional[str] = None): self.index = SearchIndex() self.index_path = index_path self.built_at = None self.source_head = None self.object_count = 0 self.failed_objects = [] self.secret_blocked_objects = [] self.stale_objects = [] # C5D: deterministische Objekt-Set-Verifikation (Read-Back). # indexed_object_ids = object_id-Werte der tatsaechlich indexierten # Objekte; indexed_paths = deren Pfade. Wird nach jedem Rebuild und # beim Laden aus dem persistierten Index befuellt und in health() # zurueckgegeben, damit C5D exakt (nicht nur count>0) verifizieren kann. self.indexed_object_ids: list[str] = [] self.indexed_paths: list[str] = [] # -- Rebuild ----------------------------------------------------------- def rebuild_from_source(self, source_path: str, head: Optional[str] = None, secret_filter: bool = True) -> dict: with open(source_path) as f: data = json.load(f) docs = [] for o in data.get("objects", []): is_legacy = not bool(o.get("id")) docs.append(Doc( path=o["path"], title=o.get("title") or o["path"].rsplit("/", 1)[-1].replace(".md", ""), id=o.get("id"), type=o.get("type"), role=o.get("role"), representation=o.get("representation"), state=o.get("state"), content_hash=o.get("content_hash", ""), body=o.get("body", ""), aliases=o.get("aliases", []), tags=o.get("tags", []), derived_from=o.get("derived_from"), is_legacy=is_legacy, )) res = self.index.build(docs, secret_filter=secret_filter) self.source_head = head or data.get("head") self.built_at = int(time.time() * 1000) self.object_count = res["indexed"] self.secret_blocked_objects = list(self.index.secret_blocked) self.failed_objects = [] self.stale_objects = [] # C5D: deterministische Objekt-Set-Verifikation (nur tatsaechlich # indexierte, nicht secret-geblockte Objekte). self.indexed_object_ids = [ d.id for d in self.index.docs if d.id is not None ] self.indexed_paths = [d.path for d in self.index.docs] if self.index_path: os.makedirs(os.path.dirname(self.index_path), exist_ok=True) payload = { "index_version": INDEX_VERSION, "source_head": self.source_head, "built_at_ms": self.built_at, "object_count": self.object_count, "secret_blocked": self.secret_blocked_objects, "indexed_object_ids": self.indexed_object_ids, "indexed_paths": self.indexed_paths, "docs": [ {"path": d.path, "title": d.title, "id": d.id, "type": d.type, "role": d.role, "representation": d.representation, "state": d.state, "content_hash": d.content_hash, "derived_from": d.derived_from, "is_legacy": d.is_legacy, "aliases": d.aliases, "tags": d.tags} for d in self.index.docs ], "postings": {k: v for k, v in self.index.postings.items()}, } with open(self.index_path, "w") as f: json.dump(payload, f, ensure_ascii=False) return res # -- Health ------------------------------------------------------------- def health(self) -> dict: return { "index_built": self.built_at is not None, "object_count": self.object_count, "last_update": self.built_at, "failed_objects": self.failed_objects, "stale_objects": self.stale_objects, "integrity_ok": self.index._ready, "supported_modes": SUPPORTED_MODES, "index_version": INDEX_VERSION, "source_head": self.source_head, "secret_blocked_objects": len(self.secret_blocked_objects), # C5D: deterministische Objekt-Set-Verifikation (Read-Back). "indexed_object_ids": self.indexed_object_ids, "indexed_paths": self.indexed_paths, } # -- Snippet (secret-gefiltert) ----------------------------------------- def _snippet(self, doc: Doc, tokens: list[str], width: int = 120) -> str: text = doc.body or "" low = text.lower() pos = -1 for tok in tokens: if not tok: continue p = low.find(tok) if p >= 0: pos = p break if pos < 0: snippet = re.sub(r"\s+", " ", text[:width]) else: start = max(0, pos - 40) end = min(len(text), pos + width) snippet = re.sub(r"\s+", " ", text[start:end]) for hit in scan_for_secrets(snippet): snippet = snippet.replace(hit, "[REDACTED]") return snippet.strip() # -- README helpers ------------------------------------------------------ @staticmethod def _is_readme_doc(doc: Doc) -> bool: return "README" in doc.path or doc.path.lower().endswith("readme.md") @staticmethod def _is_readme_explicit(q_low: str, doc: Doc) -> bool: return ("readme" in q_low) and TolariaSearch._is_readme_doc(doc) # -- Result contract ----------------------------------------------------- _LEGACY_ID = { "README.md": "README-root", "notes/trading/system-docs/README.md": "README-system-docs", "vps.md": "vps", } def _build_result(self, doc: Doc, score: float, comp: dict, matched: list, tokens: list[str]) -> dict: if doc.id: obj_id = doc.id else: obj_id = self._LEGACY_ID.get(doc.path, f"legacy/{doc.path}") group_id = obj_id return { "object_id": obj_id, "title": doc.title, "path": doc.path, "type": doc.type, "role": doc.role, "representation": doc.representation, "state": doc.state, "score": round(max(0.0, min(1.0, score)), 4), "score_components": comp, "matched_fields": matched, "snippet": self._snippet(doc, tokens), "source_system": SOURCE_SYSTEM, "source_path": doc.path, "derived_from": doc.derived_from, "relations": { "related_to": [], "belongs_to": [], "derived_from": doc.derived_from, }, "version": None, "is_current": doc.state == "current", "is_stale": False, "group": { "group_id": group_id, "members": [doc.id] if doc.id else [], "expanded": False, }, "is_legacy": doc.is_legacy or doc.id is None, } # -- Collapse (canonical/source) ----------------------------------------- def _collapse(self, results: list[dict], include_source: bool) -> list[dict]: """Metadata-basiertes Collapse ohne Semantic-Graph (C4C): Source und Canonical desselben thematischen Pfads -> eine Gruppe mit Canonical als Repräsentant. Ohne verlässlichen derived_from-Graph wird nur die canonical-Präferenz über Ranking sichergestellt; die group-Info wird korrekt gesetzt, damit Agent nie Source+Canonical als zwei unabhängige Fakten ohne group-Kontext liest.""" return results # Ranking bevorzugt canonical; group-Info je Result gesetzt # -- Exact mode ----------------------------------------------------------- def _exact(self, q_low: str, tokens: list[str], candidates: list[int], minimum_score: float) -> list[tuple[float, Doc, dict, list]]: title_hits = set(self.index.title_exact(q_low)) body_phrase = set() for i in candidates: d = self.index.docs[i] if q_low and q_low in (d.body or "").lower(): body_phrase.add(i) scored = [] for i in candidates: d = self.index.docs[i] s = 0.0 comp = {} matched = [] if i in title_hits or (q_low and q_low in (d.title or "").lower()): s = 0.9 comp["exact_title"] = 0.9 matched.append("title") elif i in body_phrase: s = 0.5 comp["exact_phrase"] = 0.5 matched.append("body") else: continue rep_b = REPRESENTATION_RANK.get(d.representation, 1) / 3 * 0.05 state_b = STATE_RANK.get(d.state, 1) / 2 * 0.04 s = min(1.0, s + rep_b + state_b) comp["metadata_boost"] = round(rep_b + state_b, 3) scored.append((s, d, comp, matched)) # deterministisch: Score desc, canonical vor source, Titel-Az scored.sort(key=lambda x: ( -x[0], -REPRESENTATION_RANK.get(x[1].representation, 1), -STATE_RANK.get(x[1].state, 1), x[1].path, )) return scored # -- Keyword mode ---------------------------------------------------------- def _keyword(self, query_low: str, tokens: list[str], candidates: list[int], minimum_score: float) -> list[tuple[float, dict, dict, list]]: scored = [] for i in candidates: d = self.index.docs[i] body_low = (d.body or "").lower() title_low = d.title.lower() cnt = sum(1 for t in tokens if t in body_low or t in title_low) if cnt == 0: continue kw = min(1.0, cnt / len(tokens)) comp = {"keyword": round(kw, 3)} matched = [] if all(t in title_low for t in tokens if t): kw += 0.15 comp["title"] = 0.15 matched.append("title") if any(t in body_low for t in tokens if t): matched.append("body") rep_b = REPRESENTATION_RANK.get(d.representation, 1) / 3 * 0.08 state_b = STATE_RANK.get(d.state, 1) / 2 * 0.06 role_b = 0.0 if d.role in INDEX_ROLES and not self._is_readme_explicit(query_low, d): role_b = -INDEX_ROLES_PENALTY meta = round(rep_b + state_b + role_b, 3) comp["metadata_boost"] = meta score = round(kw + meta, 4) if score < minimum_score: continue scored.append((score, d, comp, matched)) scored.sort(key=lambda x: (-x[0], x[1].path)) return scored # -- Metadata mode ---------------------------------------------------------- def _metadata(self, query_low: str, tokens: list[str], candidates: list[int], minimum_score: float) -> list[tuple[float, dict, dict, list]]: scored = [] for i in candidates: d = self.index.docs[i] kw = 0.0 matched = [] if tokens: body_low = (d.body or "").lower() title_low = d.title.lower() cnt = sum(1 for t in tokens if t in body_low or t in title_low) if cnt: kw = min(1.0, cnt / len(tokens) * 0.6) matched = ["title" if t in title_low else "body" for t in tokens if t in body_low or t in title_low][:3] rep_b = REPRESENTATION_RANK.get(d.representation, 1) / 3 * 0.08 state_b = STATE_RANK.get(d.state, 1) / 2 * 0.06 comp = {"keyword": round(kw, 3), "metadata_boost": round(rep_b + state_b, 3)} score = round(kw + rep_b + state_b, 4) if score < minimum_score: continue scored.append((score, d, comp, matched)) scored.sort(key=lambda x: (-x[0], x[1].path)) return scored # -- Main search ----------------------------------------------------------- def search(self, req: dict) -> dict: t0 = time.time() mode = (req.get("mode") or "hybrid").lower() query = req.get("query") or "" filters = req.get("filters") or {} include_historical = bool(req.get("include_historical", False)) limit = int(req.get("limit", DEFAULT_LIMIT)) offset = int(req.get("offset", 0)) minimum_score = float(req.get("minimum_score", 0.0)) requested_mode = mode # Honest Mode: nicht implementierte Modi if mode in ("semantic", "vector", "hybrid"): return { "requested_mode": mode, "actual_mode": None, "fallback": False, "error": {"code": "mode_not_implemented", "reason": f"{mode} ist in C4B nicht implementiert (C4C/C4D)"}, "results": [], "total": 0, "elapsed_ms": round((time.time() - t0) * 1000, 2), } if limit > MAX_LIMIT: return _err("invalid_query", "limit darf max 100 betragen") if limit < 1: return _err("invalid_query", "limit muss >= 1 sein") if offset < 0: return _err("invalid_query", "offset muss >= 0 sein") if mode not in SUPPORTED_MODES: return _err("invalid_query", f"unbekannter mode: {mode}") if mode in ("exact", "keyword") and not query: return _err("invalid_query", f"{mode} mode erfordert query") if not self.index._ready: return _err("index_unavailable", "Index wurde noch nicht gebaut") q_low, tokens = _parse_query(query) q_low = q_low.lower() all_idx = list(range(len(self.index.docs))) candidates = [ i for i in all_idx if _matches_filters(self.index.docs[i], filters, include_historical) ] actual = mode if mode == "exact": ordered = self._exact(q_low, tokens, candidates, minimum_score) elif mode == "keyword": ordered = self._keyword(q_low, tokens, candidates, minimum_score) elif mode == "metadata": ordered = self._metadata(q_low, tokens, candidates, minimum_score) else: return _err("invalid_query", f"unbekannter mode: {mode}") results = [] seen_ids = set() for score, doc, comp, matched in ordered: r = self._build_result(doc, score, comp, matched, tokens) if doc.id: if doc.id in seen_ids: continue seen_ids.add(doc.id) results.append(r) results = self._collapse(results, True) total = len(results) page = results[offset:offset + limit] elapsed = round((time.time() - t0) * 1000, 2) return { "requested_mode": requested_mode, "actual_mode": actual, "fallback": False, "query": query, "results": page, "total": total, "elapsed_ms": elapsed, "limit": limit, "offset": offset, }