::]' will search for all entries of and optionally, section/subsection '\d\d\d-\d\d\d' (two sets of 3 digits) will first try to match CRE ids before it performs a free text search Anything else will be a case insensitive LIKE query in the database """ search_terms = self._expand_text_search_aliases(text) if not search_terms: return [] text = text.strip() # structured text search first cre_id_search = r"CRE(:| )(?P\d+-\d+)" cre_naked_id_search = r"\d\d\d-\d\d\d" cre_name_search = r"CRE(:| )(?P\w+)" types = "|".join([v.value for v in cre_defs.Credoctypes]) node_search = ( r"(Node|(?P" + types + "))?((:| )?(?Phttps?://\S+))?((:| )(?P.+$))?" ) match = re.search(cre_id_search, text, re.IGNORECASE) if match: return self.get_CREs(external_id=match.group("id")) match = re.search(cre_naked_id_search, text, re.IGNORECASE) if match: return self.get_CREs(external_id=match.group()) match = re.search(cre_name_search, text, re.IGNORECASE) if match: return self.get_CREs(name=match.group("name")) match = re.search(node_search, text, re.IGNORECASE) if match: link = match.group("link") ntype = match.group("ntype") txt = match.group("val") results: List[cre_defs.Document] = [] if txt: args = txt.split(":") if ":" in txt else txt.split(" ") if len(args) < 4: args += [""] * (4 - len(args)) s = set([p for p in permutations(args, 4)]) for combo in s: nodes = self.get_nodes( name=combo[0], section=combo[1], subsection=combo[2], link=link, ntype=ntype, sectionID=combo[3], ) if nodes: results.extend(nodes) elif link or ntype: nodes = self.get_nodes(link=link, ntype=ntype) if nodes: results.extend(nodes) if results: return list(set(results)) # fuzzy matches second results = {} for search_term in search_terms: args = [f"%{search_term}%", "", "", "", "", ""] s = set([p for p in permutations(args, 6)]) for combo in s: nodes = self.get_nodes( name=combo[0], section=combo[1], subsection=combo[2], link=combo[3], description=combo[4], partial=True, ntype=None, # type: ignore sectionID=combo[5], ) if nodes: for node in nodes: node_key = f"{node.name}:{node.version}:{node.section}:{node.sectionID}:{node.subsection}:" results[node_key] = node args = [f"%{search_term}%", None, None] for combo in permutations(args, 3): cres = self.get_CREs( name=combo[0], external_id=combo[1], description=combo[2], partial=True, ) if cres: for cre in cres: results[cre.id] = cre return list(results.values()) def _expand_text_search_aliases(self, text: str) -> List[str]: normalized = text.strip() if not normalized: return [] lowered = normalized.lower() aliases = { "xxe": [ "XXE", "XML External Entity", "XML External Entity Reference", ], "xss": [ "XSS", "Cross Site Scripting", "Cross-Site Scripting", ], "ssrf": [ "SSRF", "Server Side Request Forgery", "Server-Side Request Forgery", ], "csrf": [ "CSRF", "Cross Site Request Forgery", "Cross-Site Request Forgery", ], "rce": [ "RCE", "Remote Code Execution", ], } expanded = [normalized] expanded.extend(aliases.get(lowered, [])) deduped = [] seen = set() for entry in expanded: key = entry.lower() if key in seen: continue seen.add(key) deduped.append(entry) return deduped def get_root_cres(self): """Returns CRES that only have "Contains" links""" # select name from cre where cre.id not in (select cre from cre_links where type="Contains") and cre.id not in (select "group" from cre_links where type="Is Part OF"); cres = ( self.session.query(CRE) .filter( ~CRE.id.in_( self.session.query(InternalLinks.cre).filter( InternalLinks.type == cre_defs.LinkTypes.Contains, ) ) ) .filter( ~CRE.id.in_( self.session.query(InternalLinks.group).filter( InternalLinks.type == cre_defs.LinkTypes.PartOf, ) ) ) .all() ) return self._hydrate_cres_batch(list(cres)) def health_check(self) -> Dict[str, Any]: """Lightweight liveness/readiness probe for the serving database. Intended for use by a deploy/uptime health endpoint, NOT for deep operational checks (GA completeness, mapping coverage, etc.) which are slow and belong in ops tooling. Performs cheap COUNT queries and never raises: connectivity failures are reported as ``ok=False`` so the caller can return an appropriate status code. Returns a dict with: - ``ok``: True only if the DB is reachable AND holds a non-empty dataset (at least one CRE and one standard/node). - ``db_reachable``: True if the COUNT queries executed. - ``cre_count`` / ``standards_count``: populated when reachable. - ``reason``: short human-readable explanation when ``ok`` is False. """ try: cre_count = self.session.query(func.count(CRE.id)).scalar() or 0 standards_count = self.session.query(func.count(Node.id)).scalar() or 0 except OperationalError: return { "ok": False, "db_reachable": False, "reason": "database unreachable", } except SQLAlchemyError: # pragma: no cover - defensive, never fail open return { "ok": False, "db_reachable": False, "reason": "database health query failed", } if cre_count == 0 or standards_count == 0: return { "ok": False, "db_reachable": True, "cre_count": cre_count, "standards_count": standards_count, "reason": "empty dataset", } return { "ok": True, "db_reachable": True, "cre_count": cre_count, "standards_count": standards_count, } def get_embeddings_by_doc_type(self, doc_type: str) -> Dict[str, List[float]]: from application.database.pgvector_utils import ( parse_stored_embedding_vec, require_embedding_vec_store, ) require_embedding_vec_store( self.session.get_bind(), context="get_embeddings_by_doc_type" ) res = {} embeddings = ( self.session.query(Embeddings).filter(Embeddings.doc_type == doc_type).all() ) if embeddings: for entry in embeddings: vec = parse_stored_embedding_vec(entry.embedding_vec) if not vec: continue if doc_type == cre_defs.Credoctypes.CRE.value: res[entry.cre_id] = vec else: res[entry.node_id] = vec return res def get_embedding_contents_by_doc_type(self, doc_type: str) -> Dict[str, str]: """Return ``{id -> embeddings_content}`` for a doc type. ``embeddings_content`` is the text a vector was built from — the pair text the Module C.2 cross-encoder scores a candidate against. Keyed by cre_id for CREs, node_id otherwise, mirroring ``get_embeddings_by_doc_type``. Rows with no content are skipped. """ res: Dict[str, str] = {} embeddings = ( self.session.query(Embeddings).filter(Embeddings.doc_type == doc_type).all() ) for entry in embeddings: if not entry.embeddings_content: continue if doc_type == cre_defs.Credoctypes.CRE.value: res[entry.cre_id] = entry.embeddings_content else: res[entry.node_id] = entry.embeddings_content return res def get_embeddings_by_doc_type_paginated( self, doc_type: str, page: int = 1, per_page: int = 100 ) -> Tuple[Dict[str, List[float]], int, int]: from application.database.pgvector_utils import ( parse_stored_embedding_vec, require_embedding_vec_store, ) require_embedding_vec_store( self.session.get_bind(), context="get_embeddings_by_doc_type_paginated" ) res = {} embeddings = ( self.session.query(Embeddings) .filter(Embeddings.doc_type == doc_type) .paginate(page=int(page), per_page=per_page, error_out=False) ) total_pages = embeddings.pages if embeddings.items: for entry in embeddings.items: vec = parse_stored_embedding_vec(entry.embedding_vec) if not vec: continue if doc_type == cre_defs.Credoctypes.CRE.value: res[entry.cre_id] = vec else: res[entry.node_id] = vec return res, total_pages, page def get_embeddings_for_doc(self, doc: cre_defs.Node | cre_defs.CRE) -> Embeddings: from application.database.pgvector_utils import require_embedding_vec_store require_embedding_vec_store( self.session.get_bind(), context="get_embeddings_for_doc" ) if doc.doctype == cre_defs.Credoctypes.CRE: obj = self.session.query(CRE).filter(CRE.external_id == doc.id).first() return ( self.session.query(Embeddings) .filter(Embeddings.cre_id == obj.id) .first() ) else: node = dbNodeFromNode(doc) if not node: logger.warning( f"cannot get embeddings for doc {doc.todict()} it's not translatable to a database document" ) return None obj = self.object_select(node) if obj: return ( self.session.query(Embeddings) .filter(Embeddings.node_id == obj[0].id) .first() ) def get_embedding(self, object_id: str) -> Optional[Embeddings]: return ( self.session.query(Embeddings) .filter( sqla.or_( Embeddings.cre_id == object_id, Embeddings.node_id == object_id ) ) .all() ) def delete_all_embeddings(self) -> int: """ Delete every row in ``embeddings`` so a subsequent ``PromptHandler(..., load_all_embeddings=True)`` run rebuilds vectors (e.g. after changing smart-extract policy or embedding model). """ n_deleted = self.session.query(Embeddings).delete(synchronize_session=False) self.session.commit() return int(n_deleted or 0) def add_embedding( self, db_object: CRE | Node, doctype: cre_defs.Credoctypes, embeddings: List[float], embedding_text: str, embeddings_url: Optional[str] = None, ): """ Persist embedding vector and content. For nodes, ``embeddings_url`` is the resolved URL used for fetch/embed alignment (may include a fragment). When ``None``, defaults to ``db_object.link`` (importer hyperlink). """ expected_dim_raw = (os.environ.get("CRE_EMBED_EXPECTED_DIM", "") or "").strip() embedding_model_id = ( os.environ.get("CRE_EMBED_MODEL", "") or "" ).strip() or "gemini/gemini-embedding-001" embedding_dim = len(embeddings) if expected_dim_raw: expected_dim = int(expected_dim_raw) if len(embeddings) != expected_dim: raise ValueError( f"embedding dimension mismatch for {db_object.id}: " f"expected {expected_dim}, got {len(embeddings)}" ) from application.database.pgvector_utils import ( require_embedding_vec_store, to_pgvector_literal, ) require_embedding_vec_store(self.session.get_bind(), context="add_embedding") existing = self.get_embedding(db_object.id) # Sole store is ``embedding_vec`` (pgvector on Postgres; Text literal # on SQLite). Never LLM re-embed here — see pgvector_utils. embedding_vec_literal = to_pgvector_literal(embeddings) resolved_node_url: Optional[str] = None if doctype != cre_defs.Credoctypes.CRE: resolved_node_url = ( embeddings_url if embeddings_url is not None else db_object.link ) if not existing: emb = None if doctype == cre_defs.Credoctypes.CRE: emb = Embeddings( cre_id=db_object.id, doc_type=cre_defs.Credoctypes.CRE.value, embeddings_content=embedding_text, embedding_model_id=embedding_model_id, embedding_dim=embedding_dim, embedding_vec=embedding_vec_literal, ) else: emb = Embeddings( node_id=db_object.id, doc_type=db_object.ntype, embeddings_content=embedding_text, embeddings_url=resolved_node_url, embedding_model_id=embedding_model_id, embedding_dim=embedding_dim, embedding_vec=embedding_vec_literal, ) self.session.add(emb) self.session.commit() return emb else: logger.debug(f"knew of embedding for object {db_object.id} ,updating") existing[0].embeddings_content = embedding_text existing[0].embedding_model_id = embedding_model_id existing[0].embedding_dim = embedding_dim existing[0].embedding_vec = embedding_vec_literal if doctype != cre_defs.Credoctypes.CRE: if embeddings_url is not None: existing[0].embeddings_url = embeddings_url elif getattr(existing[0], "embeddings_url", None) is None: existing[0].embeddings_url = db_object.link self.session.commit() return existing def can_use_pgvector_similarity(self) -> bool: """True when Postgres has ``embeddings.embedding_vec`` for SQL cosine.""" cached = getattr(self, "_pgvector_similarity_ready", None) if cached is not None: return bool(cached) try: bind = self.session.get_bind() from application.database.pgvector_utils import embedding_vec_column_exists ready = getattr( bind.dialect, "name", "" ) == "postgresql" and embedding_vec_column_exists(bind) except Exception: ready = False self._pgvector_similarity_ready = ready return ready def find_most_similar_embedding_id( self, query_embedding: List[float], *, doc_type: str, id_column: str, similarity_threshold: float, ) -> Tuple[Optional[str], Optional[float]]: """Top-1 cosine match via pgvector ``<=>`` (Postgres only). Returns ``(object_id, score)`` or ``(None, None)`` below threshold / when no rows match. Refuses SQLite / missing ``embedding_vec`` with ``SystemExit`` — callers that need a soft fallback must gate on ``can_use_pgvector_similarity()`` first and use sklearn instead. """ from application.database.pgvector_utils import ( fail_pgvector_unavailable, most_similar_id_sql, to_pgvector_literal, ) from sqlalchemy import text as sql_text if not self.can_use_pgvector_similarity(): fail_pgvector_unavailable(context="find_most_similar_embedding_id") sql = most_similar_id_sql(id_column) bind = self.session.get_bind() params = { "q": to_pgvector_literal(query_embedding), "doc_type": doc_type, } try: # Prefer the Session so we stay inside the app transaction; fall # back to a Connection from the Engine for SQLAlchemy 2. if hasattr(self.session, "execute"): row = self.session.execute(sql_text(sql), params).fetchone() else: with bind.connect() as conn: row = conn.execute(sql_text(sql), params).fetchone() if not row: return None, None score = float(row.score) if score < similarity_threshold: return None, None return str(row.object_id), score except SystemExit: raise except Exception as exc: # Match prior sklearn-path resilience: chat/import should degrade # to "no match" rather than 500 on a transient driver/query error. logger.warning( "pgvector similarity query failed (%s); returning no match", exc, ) return None, None def assert_embedding_contract( self, *, expected_model_id: Optional[str], expected_dim: Optional[int], ) -> None: """ Validate persisted embedding metadata consistency. - Fails when multiple dimensions are stored. - Fails when metadata is missing or mismatched against expected model/dimension. """ rows = self.session.query( Embeddings.embedding_dim, Embeddings.embedding_model_id ).all() if not rows: return dims = {int(r[0]) for r in rows if r[0] is not None} model_ids = {str(r[1]) for r in rows if r[1]} has_missing_dim = any(r[0] is None for r in rows) has_missing_model = any(not r[1] for r in rows) if len(dims) > 1: raise RuntimeError( f"multiple embedding dimensions detected in DB: {sorted(dims)}" ) if len(model_ids) > 1: raise RuntimeError( f"multiple embedding models detected in DB: {sorted(model_ids)}" ) if has_missing_dim or has_missing_model: raise RuntimeError( "embedding metadata missing in DB; run metadata migration/backfill" ) if expected_dim is not None and dims: db_dim = next(iter(dims)) if db_dim != expected_dim: raise RuntimeError( f"DB embedding dim {db_dim} does not match expected dim {expected_dim}" ) if expected_model_id and model_ids: db_model = next(iter(model_ids)) if db_model != expected_model_id: raise RuntimeError( f"DB embedding model {db_model} does not match expected model {expected_model_id}" ) def gap_analysis_exists(self, cache_key) -> bool: row = ( self.session.query(GapAnalysisResults) .filter(GapAnalysisResults.cache_key == cache_key) .first() ) if row is None: return False if gap_analysis_cache_key_is_primary(cache_key): return primary_gap_analysis_payload_is_material(row.ga_object) return True def get_gap_analysis_result(self, cache_key) -> str: logger.info(f"looking for gap analysis with cache key: {cache_key}") res = ( self.session.query(GapAnalysisResults) .filter(GapAnalysisResults.cache_key == cache_key) .first() ) if res: logger.info(f"found gap analysis with cache key: {cache_key}") return res.ga_object logger.info(f"did not find gap analysis with cache key: {cache_key}") def add_gap_analysis_result(self, cache_key: str, ga_object: str): existing = ( self.session.query(GapAnalysisResults) .filter(GapAnalysisResults.cache_key == cache_key) .first() ) if gap_analysis_cache_key_is_primary(cache_key): existing_payload = existing.ga_object if existing else None if not should_persist_primary_gap_analysis_cache( ga_object, existing_payload ): if existing is None: logger.info( "Skipping empty primary gap analysis cache insert for %s", cache_key, ) else: logger.warning( "Refusing non-material primary gap analysis update for %s", cache_key, ) return if existing: existing.ga_object = ga_object self.session.add(existing) self.session.commit() else: logger.info(f"adding gap analysis result with cache key: {cache_key}") res = GapAnalysisResults(cache_key=cache_key, ga_object=ga_object) self.session.add(res) self.session.commit() def dbNodeFromNode(doc: cre_defs.Node) -> Optional[Node]: if doc.doctype == cre_defs.Credoctypes.Standard: return dbNodeFromStandard(doc) elif doc.doctype == cre_defs.Credoctypes.Code: return dbNodeFromCode(doc) elif doc.doctype == cre_defs.Credoctypes.Tool: return dbNodeFromTool(doc) else: return None def dbNodeFromCode(code: cre_defs.Node) -> Node: code = cast(cre_defs.Code, code) tags = "" if code.tags: tags = ",".join(tags) return Node( name=code.name, ntype=code.doctype.value, tags=tags, description=code.description, link=code.hyperlink, metadata_json=code.metadata or {}, ) def dbNodeFromStandard(standard: cre_defs.Node) -> Node: standard = cast(cre_defs.Standard, standard) tags = "" if standard.tags: tags = ",".join(standard.tags) return Node( name=standard.name, ntype=standard.doctype.value, tags=tags, description=standard.description, link=standard.hyperlink, section=standard.section, subsection=standard.subsection, version=standard.version, section_id=standard.sectionID, metadata_json=standard.metadata or {}, ) def dbNodeFromTool(tool: cre_defs.Node) -> Node: tool = cast(cre_defs.Tool, tool) tgs = [tool.tooltype.value] if tool.tags: tgs.extend(tool.tags) tags = ",".join(tgs) return Node( name=tool.name, tags=tags, ntype=tool.doctype.value, description=tool.description, link=tool.hyperlink, section=tool.section, section_id=tool.sectionID, metadata_json=tool.metadata or {}, ) def nodeFromDB(dbnode: Node) -> cre_defs.Node: if not dbnode: return None tags = [] if dbnode.tags: tags = list(set(dbnode.tags.split(","))) metadata = dbnode.metadata_json or {} if dbnode.ntype == cre_defs.Standard.__name__: return cre_defs.Standard( name=dbnode.name, section=dbnode.section, subsection=dbnode.subsection, hyperlink=dbnode.link, tags=tags, version=dbnode.version, sectionID=dbnode.section_id, description=dbnode.description or "", metadata=metadata, ) elif dbnode.ntype == cre_defs.Tool.__name__: ttype = cre_defs.ToolTypes.Unknown for tag in tags: if tag in cre_defs.ToolTypes: ttype = cre_defs.ToolTypes[tag] tags.remove(tag) return cre_defs.Tool( name=dbnode.name, hyperlink=dbnode.link, tags=tags, description=dbnode.description, tooltype=ttype, section=dbnode.section, sectionID=dbnode.section_id, metadata=metadata, ) elif dbnode.ntype == cre_defs.Code.__name__: return cre_defs.Code( name=dbnode.name, hyperlink=dbnode.link, tags=tags, description=dbnode.description, metadata=metadata, ) else: raise ValueError( f"Db node {dbnode.name} has an unrecognised ntype {dbnode.ntype}" ) def CREfromDB(dbcre: CRE) -> cre_defs.CRE: if not dbcre: return None tags = [] if dbcre.tags: tags = list(set(dbcre.tags.split(","))) return cre_defs.CRE( name=dbcre.name, description=dbcre.description, id=dbcre.external_id, tags=tags, metadata=dbcre.metadata_json or {}, ) def dbCREfromCRE(cre: cre_defs.CRE) -> CRE: tags = cre.tags if cre.tags else [] return CRE( name=cre.name, description=cre.description, external_id=cre.id, tags=",".join(tags), metadata_json=cre.metadata or {}, ) def gap_analysis( neo_db: NEO_DB, node_names: List[str], cache_key: str = "", ): if neo_db is None: neo_db = NEO_DB.instance() cre_db = Node_collection() base_standard, paths = neo_db.gap_analysis(node_names[0], node_names[1]) logger.info(f"got db gap analysis for {'>>>'.join(node_names)}, calculating paths") if base_standard is None: return None grouped_paths = {} extra_paths_dict = {} GA_STRONG_UPPER_LIMIT = 2 for node in base_standard: key = node.id if not key: logger.error( "key is empty, this is a bug and this gap analysis will not progress" ) continue if key not in grouped_paths: grouped_paths[key] = {"start": node, "paths": {}, "extra": 0} extra_paths_dict[key] = {"paths": {}} # Paths may start from CRE nodes that were not included in ``base_standard`` (or # ``base_standard`` entries were skipped due to empty ids). Seed roots from each # path's start so we never drop Neo paths or hit KeyError in the merge loop below. for path in paths: start_doc = path.get("start") if start_doc is None: continue key = getattr(start_doc, "id", "") or "" if not key: continue if key not in grouped_paths: grouped_paths[key] = {"start": start_doc, "paths": {}, "extra": 0} extra_paths_dict[key] = {"paths": {}} for path in paths: key = path["start"].id end_key = path["end"].id if not end_key: logger.error( "end_key is empty, this is a bug and this gap analysis will not progress" ) continue path["score"] = get_path_score(path) del path["start"] if ( path["score"] <= GA_STRONG_UPPER_LIMIT ): # strong paths, return them in the main object if end_key in extra_paths_dict[key]["paths"]: # if we found a shortest path to a node previously in the weak paths del extra_paths_dict[key]["paths"][end_key] grouped_paths[key]["extra"] -= 1 if end_key in grouped_paths[key]["paths"]: if grouped_paths[key]["paths"][end_key]["score"] > path["score"]: # if we found a shortest path to an existing strong path grouped_paths[key]["paths"][end_key] = path else: grouped_paths[key]["paths"][end_key] = path else: if end_key in grouped_paths[key]["paths"]: continue if end_key in extra_paths_dict[key]["paths"]: if extra_paths_dict[key]["paths"][end_key]["score"] > path["score"]: extra_paths_dict[key]["paths"][end_key] = path else: extra_paths_dict[key]["paths"][end_key] = path grouped_paths[key]["extra"] += 1 if cache_key == "": cache_key = make_resources_key(node_names) if not grouped_paths: if gap_analysis_cache_key_is_primary(cache_key): stale = ( cre_db.session.query(GapAnalysisResults) .filter( or_( GapAnalysisResults.cache_key == cache_key, GapAnalysisResults.cache_key.like(cache_key + "->%"), ) ) .all() ) for row in stale: cre_db.session.delete(row) if stale: cre_db.session.commit() logger.warning( "Not persisting gap analysis for %s: grouped_paths is empty " "(Neo likely had no parseable base standard or no paths). " "Re-run after Neo is populated so the pair is not locked in as an empty cache.", cache_key, ) return (node_names, grouped_paths, extra_paths_dict) logger.info(f"got gap analysis paths for {'>>>'.join(node_names)}, storing result") cre_db.add_gap_analysis_result( cache_key=cache_key, ga_object=flask_json.dumps({"result": grouped_paths}) ) for key in extra_paths_dict: cre_db.add_gap_analysis_result( cache_key=make_subresources_key(node_names, key), ga_object=flask_json.dumps({"result": extra_paths_dict[key]}), ) logger.info(f"stored gapa analysis for {'>>>'.join(node_names)}, successfully") return (node_names, grouped_paths, extra_paths_dict)