From ba2a43f3e3a347414fe9ae5481032bbd2f81d166 Mon Sep 17 00:00:00 2001 From: Georgie Kennedy Date: Sat, 12 Sep 2026 10:51:18 +1000 Subject: [PATCH] cleanup --- docs/development/plugins.md | 12 +- docs/services/vocab.md | 18 ++ src/groundworkers/app.py | 57 +++- .../application/setup/databases.py | 18 +- .../setup/performance_maintenance.py | 16 +- .../application/setup/runtime_setup.py | 1 - src/groundworkers/base/sql.py | 44 ++- src/groundworkers/plugins.py | 67 +++++ src/groundworkers/services/mapping.py | 268 ++++++++++++++---- src/groundworkers/services/vocab.py | 266 +++++++++++------ src/groundworkers/tools/mapping_tools.py | 14 +- src/groundworkers/tui/pages/setup.py | 44 ++- src/groundworkers/tui/presenters/database.py | 9 +- .../tui/presenters/performance.py | 88 ++++-- src/groundworkers/tui/routes.py | 2 +- .../tui/wizards/performance_maintenance.py | 39 ++- tests/unit/test_mapping_context.py | 90 ++++++ tests/unit/test_mapping_service.py | 28 +- tests/unit/test_mapping_tools.py | 1 + tests/unit/test_performance.py | 5 +- .../test_performance_maintenance_wizard.py | 24 ++ tests/unit/test_setup_databases.py | 4 +- tests/unit/test_vocab_service.py | 81 +++++- 23 files changed, 962 insertions(+), 234 deletions(-) create mode 100644 tests/unit/test_mapping_context.py create mode 100644 tests/unit/test_performance_maintenance_wizard.py diff --git a/docs/development/plugins.md b/docs/development/plugins.md index b09d4cc..04b6398 100644 --- a/docs/development/plugins.md +++ b/docs/development/plugins.md @@ -76,12 +76,18 @@ The concept-embedding row and the free-text-embedding row look similar but are n ### Assumptions this table is making -- **Vocabulary only, so far.** Every example above reads OMOP *vocabulary* i.e. operations over concepts, concept relationships, concept embeddings. Nothing about `context.cdm_engine` restricts a plugin to vocabulary tables specifically, however. These tables are available over the same CDM connection groundworkers itself uses, and a plugin *could* query clinical/person-level tables (`condition_occurrence`, `drug_exposure`, etc.) through it just as easily, subject to whatever environmental and procedural access controls actually govern that data. -- **Plugin graph access is TODO.** `PluginContext` does not expose graph traversal at this time, which means that a plugin that wants graph traversal would have to depend on `omop-graph` itself and build its own adapter, reusing `context.cdm_engine` for the connection but not groundworkers' own shared instance. Recommend not doing it that way, however, as it would be easier to expose the core graph the same way `vector_store` and `embedding_backend_factory` already are. +- **Vocabulary only, so far.** Every example above reads OMOP *vocabulary* i.e. operations over concepts, concept relationships, concept embeddings, or the host's shared grounding cascade. Nothing about `context.cdm_engine` restricts a plugin to vocabulary tables specifically, however. These tables are available over the same CDM connection groundworkers itself uses, and a plugin *could* query clinical/person-level tables (`condition_occurrence`, `drug_exposure`, etc.) through it just as easily, subject to whatever environmental and procedural access controls actually govern that data. +- **Shared mapping access.** `PluginContext.mapping_context` is an optional, + read-only capability façade over the host's mapping and grounding policy. + Use its operations when a plugin needs candidate bundles, text grounding, or + concept context; the underlying vocabulary, graph, and service objects are + deliberately not exposed. Do not construct a second `omop-graph` or + vocabulary adapter from `context.cdm_engine`. The field is `None` only when + the host did not compose services for the plugin context. ## Build and register a new plugin -`build()` receives `PluginContext`. The context provides the resolved CDM database and engine, vector store, shared lazy model backend factories, and a deliberately narrow resolver for independent named OA resources. Return `None` when a prerequisite is unavailable. Groundworkers then keeps serving its core capabilities and reports why the plugin was not activated. +`build()` receives `PluginContext`. The context provides the resolved CDM database and engine, vector store, shared lazy model backend factories, the optional read-only `mapping_context`, and a deliberately narrow resolver for independent named OA resources. Return `None` when a prerequisite is unavailable. Groundworkers then keeps serving its core capabilities and reports why the plugin was not activated. `register()` adds MCP tools, prompts, or resources against the state returned by `build()`. The shared MCP registration wrapper supplies the same safe error translation used by core tools. Code-defined prompts can continue to derive their arguments from the callable signature. A plugin publishing data-defined prompts can pass explicit string metadata with `PromptArgument`; this keeps dynamic pack fields out of Python closure signatures: diff --git a/docs/services/vocab.md b/docs/services/vocab.md index 3371f63..41e9fbd 100644 --- a/docs/services/vocab.md +++ b/docs/services/vocab.md @@ -48,11 +48,29 @@ vocab.search_normalized( vocabulary_id: str | None = None, standard_only: bool = False, limit: int = 20, + allow_unindexed_scan: bool = False, ) -> list[ConceptMatch] ``` Normalized search: lowercases and strips punctuation from both the query and concept names before matching. Catches common surface-form differences (abbreviations, spacing, case) that exact search misses without the false-positive risk of full-text search. +#### Why this one needs a selective filter + +Comparison is on a *computed* expression, so an ordinary index on `concept_name` cannot serve it. Without a selective filter the planner scans every concept. Measured against a 1.7M-concept CDM: + +| Call | Time | +| --- | --- | +| unconstrained | 40s | +| `domain='Drug'` | 25s | +| `vocabulary_id='MedDRA'` | 0.4s | + +So the call is **refused** unless one of the following holds: + +- `vocabulary_id` or `parent_ids` is given. Note that `domain` does **not** count — it looks like a narrowing filter but the largest domains hold millions of concepts, which is why `domain='Drug'` still took 25s. +- `allow_unindexed_scan=True`, which accepts the scan deliberately. + +It raises rather than warning because the unindexed query does not fail — it succeeds slowly, so a warning gets absorbed into a caller that merely looks sluggish. + ### `search_fulltext` ```python diff --git a/src/groundworkers/app.py b/src/groundworkers/app.py index e56782c..eb224fd 100644 --- a/src/groundworkers/app.py +++ b/src/groundworkers/app.py @@ -15,6 +15,7 @@ from groundworkers.plugins import ( GroundworkersPlugin, GroundworkersPluginReadiness, + MappingContext, PluginConfigResolver, PluginContext, PluginReadinessResult, @@ -52,8 +53,12 @@ class Adapters: # _build_plugin_context) so a plugin reuses the one backend connection # instead of opening a second one. None when the corresponding backend # is unconfigured, same gating as the adapters above. - embedding_backend_factory: Callable[[], Any] | None = field(default=None, repr=False) - embedding_model_backend_factory: Callable[[], Any] | None = field(default=None, repr=False) + embedding_backend_factory: Callable[[], Any] | None = field( + default=None, repr=False + ) + embedding_model_backend_factory: Callable[[], Any] | None = field( + default=None, repr=False + ) chat_backend_factory: Callable[[], Any] | None = field(default=None, repr=False) @@ -104,7 +109,9 @@ def get_model_backend() -> ModelBackend: nonlocal shared_model_backend if shared_model_backend is None: if resolved_model is None: - raise RuntimeError("No embedding model is configured for Groundworkers.") + raise RuntimeError( + "No embedding model is configured for Groundworkers." + ) from omop_llm import build_model_backend_from_resolved shared_model_backend = build_model_backend_from_resolved(resolved_model) @@ -180,9 +187,14 @@ def get_chat_backend() -> ModelBackend: def build_services(config: AppConfig, adapters: Adapters) -> Services: services = Services() assisted_classifier = None - if adapters.llm is not None and config.groundworkers.source_planning_llm_assisted_enabled: + if ( + adapters.llm is not None + and config.groundworkers.source_planning_llm_assisted_enabled + ): assisted_classifier = AssistedColumnRoleClassifier(adapters.llm) - services.source_planning = SourcePlanningService(assisted_classifier=assisted_classifier) + services.source_planning = SourcePlanningService( + assisted_classifier=assisted_classifier + ) if adapters.omop_graph is not None: # cdm is optional; only the classified-edge traversals # (concept_associations / concept_extended_inheritance) require it. @@ -206,7 +218,27 @@ def build_services(config: AppConfig, adapters: Adapters) -> Services: return services -def _build_plugin_context(config: AppConfig, adapters: Adapters) -> PluginContext: +def build_mapping_context( + config: AppConfig, + *, + adapters: Adapters | None = None, + services: Services | None = None, +) -> MappingContext: + """Build the mapping capabilities shared by hosted and standalone plugins.""" + + shared_adapters = adapters or build_adapters(config) + shared_services = services or build_services(config, shared_adapters) + return MappingContext( + mapping_service=shared_services.mapping, + grounding_service=shared_services.grounding, + ) + + +def _build_plugin_context( + config: AppConfig, + adapters: Adapters, + services: Services | None = None, +) -> PluginContext: """Assemble the resolved handles every plugin is given. `cdm_database`/`vector_store` come straight off `AppConfig`, already @@ -223,6 +255,11 @@ def _build_plugin_context(config: AppConfig, adapters: Adapters) -> PluginContex embedding_backend_factory=adapters.embedding_backend_factory, embedding_model_backend_factory=adapters.embedding_model_backend_factory, chat_backend_factory=adapters.chat_backend_factory, + mapping_context=( + build_mapping_context(config, adapters=adapters, services=services) + if services is not None + else None + ), ) @@ -292,7 +329,8 @@ def verify_plugin_readiness( summary="This plugin does not expose readiness verification.", ) adapters = build_adapters(config) - context = _build_plugin_context(config, adapters) + services = build_services(config, adapters) + context = _build_plugin_context(config, adapters, services) state, issue = _build_plugin(config, context, plugin) if state is None: return PluginReadinessResult( @@ -342,13 +380,14 @@ def load_plugin_readiness( def build_application(config: AppConfig) -> GroundworkersApp: adapters = build_adapters(config) - context = _build_plugin_context(config, adapters) + services = build_services(config, adapters) + context = _build_plugin_context(config, adapters, services) plugin_definitions = tuple(discover_plugins()) plugins, plugin_issues = _build_plugins(config, context, plugin_definitions) return GroundworkersApp( config=config, adapters=adapters, - services=build_services(config, adapters), + services=services, plugins=plugins, plugin_definitions=plugin_definitions, plugin_issues=plugin_issues, diff --git a/src/groundworkers/application/setup/databases.py b/src/groundworkers/application/setup/databases.py index 298a852..ac25acd 100644 --- a/src/groundworkers/application/setup/databases.py +++ b/src/groundworkers/application/setup/databases.py @@ -111,6 +111,9 @@ def resolve_database_targets( cdm = resolver.resolve_database(groundworkers.cdm_db) if not isinstance(cdm, ResolvedCDMDatabase): return () + cdm_schema = cdm.schema_name or ( + "main" if cdm.connection.url.startswith("sqlite") else "public" + ) embedding_target: DatabaseTarget | None = None expected_embedding_model_name: str | None = None embedding_safe_url: str | None = None @@ -146,7 +149,7 @@ def resolve_database_targets( database_entry_name=cdm.name, connection_name=cdm.connection.name, safe_url=cdm.connection.safe_url, - cdm_schema=cdm.schema_name or "main", + cdm_schema=cdm_schema, vocabulary_schema=cdm.vocab_schema, connection_url=cdm.connection.url, role="cdm", @@ -157,7 +160,7 @@ def resolve_database_targets( database_entry_name=cdm.name, connection_name=cdm.connection.name, safe_url=cdm.connection.safe_url, - cdm_schema=cdm.schema_name or "main", + cdm_schema=cdm_schema, vocabulary_schema=cdm.vocab_schema, connection_url=cdm.connection.url, role="graph", @@ -168,7 +171,7 @@ def resolve_database_targets( database_entry_name=cdm.name, connection_name=cdm.connection.name, safe_url=cdm.connection.safe_url, - cdm_schema=cdm.schema_name or "main", + cdm_schema=cdm_schema, vocabulary_schema=cdm.vocab_schema, connection_url=cdm.connection.url, role="groundworkers", @@ -478,9 +481,7 @@ def _groundworkers_embedding_model_diagnostics( "Groundworkers grounding model could not be checked because the omop-emb model registry table is missing.", ), ) - rows = _embedding_registry_rows( - connection, schema=target.embedding_schema - ) + rows = _embedding_registry_rows(connection, schema=target.embedding_schema) except Exception as exc: # Broad except: reported as a redacted warning. failure = classify_connection_error(exc) @@ -903,10 +904,7 @@ def _column_has_null_values( column = quote_identifier(connection, column_name) return bool( connection.execute( - text( - f"SELECT EXISTS (SELECT 1 FROM {qualified} " - f"WHERE {column} IS NULL)" - ) + text(f"SELECT EXISTS (SELECT 1 FROM {qualified} WHERE {column} IS NULL)") ).scalar() ) diff --git a/src/groundworkers/application/setup/performance_maintenance.py b/src/groundworkers/application/setup/performance_maintenance.py index f118957..4426712 100644 --- a/src/groundworkers/application/setup/performance_maintenance.py +++ b/src/groundworkers/application/setup/performance_maintenance.py @@ -38,7 +38,11 @@ def build_performance_commands( ) -> tuple[Command, ...]: """Build safe, out-of-process commands for the selected indexes.""" - environment = () if config_path is None else (("OA_CONFIG_PATH", str(Path(config_path).expanduser())),) + environment = ( + () + if config_path is None + else (("OA_CONFIG_PATH", str(Path(config_path).expanduser())),) + ) commands: list[Command] = [] selected = set(actions) if PerformanceRemediation.TRIGRAM_INDEXES in selected: @@ -55,7 +59,9 @@ def build_performance_commands( ) if PerformanceRemediation.EMBEDDING_INDEX in selected: if not embedding_model: - raise ValueError("A registered embedding model is required to build its index.") + raise ValueError( + "A registered embedding model is required to build its index." + ) commands.append( Command( argv=( @@ -114,7 +120,11 @@ def populate_trigram_indexes(snapshot: ConfigurationSnapshot) -> None: connection.execute(text("CREATE EXTENSION IF NOT EXISTS pg_trgm")) for table, column, index_name in ( ("concept", "concept_name", "idx_concept_lower_name_trgm"), - ("concept_synonym", "concept_synonym_name", "idx_concept_synonym_lower_name_trgm"), + ( + "concept_synonym", + "concept_synonym_name", + "idx_concept_synonym_lower_name_trgm", + ), ): connection.execute( text( diff --git a/src/groundworkers/application/setup/runtime_setup.py b/src/groundworkers/application/setup/runtime_setup.py index 758ad1a..bd71e17 100644 --- a/src/groundworkers/application/setup/runtime_setup.py +++ b/src/groundworkers/application/setup/runtime_setup.py @@ -498,4 +498,3 @@ def _optional_str(value: object) -> str | None: def _effective_tool(stack: StackConfig, name: str): return stack.tools.get(name) - diff --git a/src/groundworkers/base/sql.py b/src/groundworkers/base/sql.py index 8d6461a..2a293ae 100644 --- a/src/groundworkers/base/sql.py +++ b/src/groundworkers/base/sql.py @@ -2,9 +2,30 @@ from __future__ import annotations +from enum import StrEnum + +from sqlalchemy import func from sqlalchemy.engine import Connection, Engine -__all__ = ["effective_schema", "quote_identifier"] +__all__ = [ + "STOP_PHRASE_SQL_PATTERN", + "NormalizationProfile", + "effective_schema", + "normalization_expression", + "quote_identifier", +] + + +class NormalizationProfile(StrEnum): + """Supported normalization contracts shared by Python and SQL callers.""" + + VERBATIM = "verbatim" + AGGRESSIVE = "aggressive" + DRUG_NAME = "drug_name" + + +#: Stop phrases stripped before comparison, as a PostgreSQL regex. +STOP_PHRASE_SQL_PATTERN = r"\m(?:nos|nec|nfs|unspecified|unknown|other|w/?o|w/)\M" def quote_identifier(bind: Engine | Connection, name: str) -> str: @@ -28,3 +49,24 @@ def effective_schema(bind: Engine | Connection) -> str | None: comes from, so reflection and generated SQL cannot disagree. """ return bind.get_execution_options().get("schema_translate_map", {}).get(None) + + +def normalization_expression( + column, + *, + profile: str | NormalizationProfile, + remove_stop_phrases: bool, +): + """Build the SQL expression used for normalized comparison.""" + profile = NormalizationProfile(profile) + space = " " + globally = "g" + expr = func.lower(column) + if remove_stop_phrases: + expr = func.regexp_replace(expr, STOP_PHRASE_SQL_PATTERN, space, globally) + expr = func.regexp_replace(expr, r"[^a-z0-9]+", space, globally) + if profile == "drug_name": + expr = func.replace(expr, " extended release ", space) + expr = func.replace(expr, " modified release ", space) + expr = func.regexp_replace(expr, r"\s+", space, globally) + return func.btrim(expr) diff --git a/src/groundworkers/plugins.py b/src/groundworkers/plugins.py index 425b394..9c98fef 100644 --- a/src/groundworkers/plugins.py +++ b/src/groundworkers/plugins.py @@ -37,6 +37,9 @@ MutationOperation, ) + from groundworkers.services.grounding import ConceptGroundingService + from groundworkers.services.mapping import MappingService + logger = logging.getLogger(__name__) _PLUGIN_SETUP_FIELD_KINDS = frozenset( @@ -167,6 +170,69 @@ def resolve_vector_store(self, name: str) -> ResolvedVectorStore: return self._resolver.resolve_vector_store(name) +class MappingContext: + """Stable, read-only access to Groundworkers' shared mapping services. + + The services are deliberately held behind this host-owned façade. Plugins + can reuse the configured vocabulary, graph, and mapping orchestration + without constructing a second adapter or depending on Groundworkers' + application container. All exposed operations are query-only. + """ + + resolver_identity = "groundworkers-grounding-v1" + + def __init__( + self, + *, + mapping_service: MappingService | None = None, + grounding_service: ConceptGroundingService | None = None, + ) -> None: + self._mapping_service = mapping_service + self._grounding_service = grounding_service + + @property + def available(self) -> bool: + """Whether the host built the candidate-mapping service.""" + + return self._mapping_service is not None + + @property + def grounding_available(self) -> bool: + """Whether the shared tiered grounding capability is configured.""" + + return self._grounding_service is not None + + def candidate_bundle(self, query: str, **kwargs: Any) -> dict[str, Any]: + """Build a shared candidate/evidence bundle for *query*.""" + + if self._mapping_service is None: + raise RuntimeError("Groundworkers mapping service is not configured") + return self._mapping_service.concept_candidate_bundle(query, **kwargs) + + async def async_candidate_bundle(self, query: str, **kwargs: Any) -> dict[str, Any]: + """Build a candidate bundle using the service's async retrieval path.""" + + if self._mapping_service is None: + raise RuntimeError("Groundworkers mapping service is not configured") + return await self._mapping_service.async_concept_candidate_bundle( + query, **kwargs + ) + + def concept_mapping_context(self, concept_id: int, **kwargs: Any) -> dict[str, Any]: + """Build the shared context packet for one known OMOP concept.""" + + if self._mapping_service is None: + raise RuntimeError("Groundworkers mapping service is not configured") + return self._mapping_service.concept_mapping_context(concept_id, **kwargs) + + def ground(self, query: str, **kwargs: Any) -> dict[str, Any]: + """Resolve text through Groundworkers' shared tiered grounding policy.""" + + if self._grounding_service is None: + raise RuntimeError("Groundworkers grounding service is not configured") + return self._grounding_service.ground(query, **kwargs) + + @dataclass(frozen=True) class PluginContext: """Resolved, read-only handles passed to every plugin. @@ -190,6 +256,7 @@ class PluginContext: embedding_backend_factory: Callable[[], Any] | None embedding_model_backend_factory: Callable[[], Any] | None chat_backend_factory: Callable[[], Any] | None + mapping_context: MappingContext | None = None class GroundworkersPlugin(Protocol): diff --git a/src/groundworkers/services/mapping.py b/src/groundworkers/services/mapping.py index d91c819..137ab7b 100644 --- a/src/groundworkers/services/mapping.py +++ b/src/groundworkers/services/mapping.py @@ -5,6 +5,7 @@ from groundworkers.adapters.omop_emb import OmopEmbAdapter from groundworkers.base.errors import GroundworkersError +from groundworkers.base.sql import NormalizationProfile from groundworkers.services.graph import GraphService from groundworkers.services.grounding import ConceptGroundingService from groundworkers.services.vocab import ( @@ -45,9 +46,11 @@ def concept_search_normalized( vocabulary_id: str | None = None, standard_only: bool = False, include_synonyms: bool = False, - normalization_profile: str = "verbatim", + normalization_profile: str + | NormalizationProfile = NormalizationProfile.VERBATIM, remove_stop_phrases: bool = True, limit: int = 20, + allow_unindexed_scan: bool = False, ) -> dict[str, Any]: stripped = query.strip() if not stripped: @@ -58,7 +61,9 @@ def concept_search_normalized( remove_stop_phrases=remove_stop_phrases, ) if not normalized_query: - raise ValueError("query must contain searchable content after normalization") + raise ValueError( + "query must contain searchable content after normalization" + ) results = self._vocab.search_normalized( normalized_query, @@ -69,6 +74,7 @@ def concept_search_normalized( normalization_profile=normalization_profile, remove_stop_phrases=remove_stop_phrases, limit=limit, + allow_unindexed_scan=allow_unindexed_scan, ) serialised = [] for match in results: @@ -82,7 +88,9 @@ def concept_search_normalized( row["matched_text"] = matched_text row["matched_text_normalized"] = matched_text_normalized row["match_mode"] = ( - "synonym_exact_normalized" if match.match_source == "synonym" else "label_exact_normalized" + "synonym_exact_normalized" + if match.match_source == "synonym" + else "label_exact_normalized" ) serialised.append(row) return { @@ -93,6 +101,52 @@ def concept_search_normalized( "results": serialised, } + def _normalized_channel( + self, + query: str, + *, + domain: str | None, + vocabulary_id: str | None, + standard_only: bool, + active_only: bool, + parent_ids: list[int] | None, + limit: int, + ) -> tuple[dict[str, Any], str | None]: + """The bundle's normalized channel, dropped when it would scan the table. + + A bundle surveys what each channel can offer, so an unaffordable + normalized query is omitted the way an absent full-text sidecar is. + Letting the underlying guard raise would fail the whole bundle over one + optional channel. + """ + available = bool(vocabulary_id or parent_ids) + if not available: + return ( + {"available": False, "results": [], "retrieval_notes": []}, + "normalized channel omitted: it would scan the whole concept table " + "without a vocabulary_id or parent_ids filter", + ) + results = self._vocab.search_normalized( + query, + domain=domain or None, + vocabulary_id=vocabulary_id or None, + standard_only=standard_only, + active_only=active_only, + include_synonyms=False, + parent_ids=parent_ids, + limit=limit, + ) + return ( + { + "available": True, + "results": [serialise_concept_match(r) for r in results], + "retrieval_notes": [ + "deterministic normalized equality over concept labels" + ], + }, + None, + ) + def concept_candidate_bundle( self, query: str, @@ -112,7 +166,9 @@ def concept_candidate_bundle( per_channel_limit: int = 10, overall_limit: int = 30, model_name: str | None = None, - _embedding_result: dict[str, Any] | GroundworkersError | object = _SYNC_EMBEDDING, + _embedding_result: dict[str, Any] + | GroundworkersError + | object = _SYNC_EMBEDDING, ) -> dict[str, Any]: stripped = query.strip() if not stripped: @@ -144,25 +200,24 @@ def concept_candidate_bundle( channels["exact"] = { "available": True, "results": [serialise_concept_match(r) for r in exact_results], - "retrieval_notes": ["case-insensitive exact match over concept_name and optional synonyms"], + "retrieval_notes": [ + "case-insensitive exact match over concept_name and optional synonyms" + ], } if include_normalized: - normalized_results = self._vocab.search_normalized( + channel, warning = self._normalized_channel( query, - domain=domain or None, - vocabulary_id=vocabulary_id or None, + domain=domain, + vocabulary_id=vocabulary_id, standard_only=standard_only, active_only=active_only, - include_synonyms=False, parent_ids=parent_ids, limit=per_channel_limit, ) - channels["normalized"] = { - "available": True, - "results": [serialise_concept_match(r) for r in normalized_results], - "retrieval_notes": ["deterministic normalized equality over concept labels"], - } + channels["normalized"] = channel + if warning: + warnings.append(warning) if include_fulltext: fts_results, fts_available = self._vocab.search_fulltext( @@ -178,15 +233,25 @@ def concept_candidate_bundle( channels["fulltext"] = { "available": fts_available, "results": [serialise_concept_match(r) for r in fts_results], - "retrieval_notes": ["ranked PostgreSQL full-text retrieval"] if fts_available else [], + "retrieval_notes": ["ranked PostgreSQL full-text retrieval"] + if fts_available + else [], } if not fts_available: - warnings.append("full-text sidecar columns unavailable; fulltext channel omitted") + warnings.append( + "full-text sidecar columns unavailable; fulltext channel omitted" + ) if include_embedding: if self._emb is None: - channels["embedding"] = {"available": False, "results": [], "retrieval_notes": []} - warnings.append("embedding adapter not configured; embedding channel omitted") + channels["embedding"] = { + "available": False, + "results": [], + "retrieval_notes": [], + } + warnings.append( + "embedding adapter not configured; embedding channel omitted" + ) else: try: if _embedding_result is _SYNC_EMBEDDING: @@ -206,14 +271,20 @@ def concept_candidate_bundle( embedding_result = _embedding_result emb_notes = ["semantic retrieval from omop-emb"] if parent_ids: - emb_notes.append("parent_ids hierarchy filter was not applied at the embedding level") + emb_notes.append( + "parent_ids hierarchy filter was not applied at the embedding level" + ) channels["embedding"] = { "available": True, "results": embedding_result.get("results", []), "retrieval_notes": emb_notes, } except GroundworkersError as exc: - channels["embedding"] = {"available": False, "results": [], "retrieval_notes": []} + channels["embedding"] = { + "available": False, + "results": [], + "retrieval_notes": [], + } warnings.append(f"embedding channel unavailable: {exc.message}") candidate_union = self._build_candidate_union(channels, overall_limit) @@ -221,18 +292,31 @@ def concept_candidate_bundle( standardized_candidates: list[dict[str, Any]] = [] if include_standard_mappings and candidate_union: - concept_ids = [row["concept_id"] for row in candidate_union if not row.get("standard_concept")] + concept_ids = [ + row["concept_id"] + for row in candidate_union + if not row.get("standard_concept") + ] if concept_ids: mappings = self._vocab.navigate_to_standard(concept_ids) - standardized_candidates = [serialise_standard_mapping(m) for m in mappings] - mapping_index = {m["source_concept_id"]: m["standard_concepts"] for m in standardized_candidates} + standardized_candidates = [ + serialise_standard_mapping(m) for m in mappings + ] + mapping_index = { + m["source_concept_id"]: m["standard_concepts"] + for m in standardized_candidates + } for row in candidate_union: - row["mapped_standard_concepts"] = mapping_index.get(row["concept_id"], []) + row["mapped_standard_concepts"] = mapping_index.get( + row["concept_id"], [] + ) if include_hierarchy_context and self._graph is not None: for row in candidate_union[: min(5, len(candidate_union))]: try: - row["ancestor_preview"] = self._graph.get_ancestors(row["concept_id"], 2)[:3] + row["ancestor_preview"] = self._graph.get_ancestors( + row["concept_id"], 2 + )[:3] except Exception: row["ancestor_preview"] = [] elif include_hierarchy_context: @@ -332,13 +416,17 @@ def concept_nearest_standard_ancestor( _grounded_result: dict[str, Any] | object = _SYNC_GROUNDING, ) -> dict[str, Any]: if self._graph is None: - raise GroundworkersError("BACKEND_UNAVAIL", "omop_graph backend is not configured") + raise GroundworkersError( + "BACKEND_UNAVAIL", "omop_graph backend is not configured" + ) if (query is None) == (concept_id is None): raise ValueError("exactly one of query or concept_id must be provided") if query is not None: if self._grounding is None: - raise GroundworkersError("BACKEND_UNAVAIL", "grounding service is not configured") + raise GroundworkersError( + "BACKEND_UNAVAIL", "grounding service is not configured" + ) if _grounded_result is _SYNC_GROUNDING: grounded = self._grounding.ground( query.strip(), @@ -364,8 +452,12 @@ def concept_nearest_standard_ancestor( seed = results[0] seed_concept = self._graph.get_concept(seed["concept_id"]) if seed_concept is None: - raise GroundworkersError("NOT_FOUND", f"Concept {seed['concept_id']} was not found") - seed_is_standard = seed_concept is not None and seed_concept.get("standard_concept") + raise GroundworkersError( + "NOT_FOUND", f"Concept {seed['concept_id']} was not found" + ) + seed_is_standard = seed_concept is not None and seed_concept.get( + "standard_concept" + ) selection_reason = ( "exact_standard_match" if seed.get("match_kind") == "EXACT" and seed_is_standard @@ -375,8 +467,14 @@ def concept_nearest_standard_ancestor( assert concept_id is not None seed_concept = self._graph.get_concept(concept_id) if seed_concept is None: - raise GroundworkersError("NOT_FOUND", f"Concept {concept_id} was not found") - seed = {"concept_id": concept_id, "concept_name": seed_concept["concept_name"], "match_kind": "DIRECT"} + raise GroundworkersError( + "NOT_FOUND", f"Concept {concept_id} was not found" + ) + seed = { + "concept_id": concept_id, + "concept_name": seed_concept["concept_name"], + "match_kind": "DIRECT", + } selection_reason = "direct_concept_input" warnings: list[str] = [] @@ -387,7 +485,7 @@ def concept_nearest_standard_ancestor( ancestors = self._graph.get_ancestors(seed_concept["concept_id"], max_depth) standard_ancestors = [a for a in ancestors if a.get("standard_concept")] selected_parent = standard_ancestors[0] if standard_ancestors else None - alternatives = standard_ancestors[1: min(5, len(standard_ancestors))] + alternatives = standard_ancestors[1 : min(5, len(standard_ancestors))] if selected_parent is None: # No standard ancestor. Before giving up, follow "Maps to": a @@ -409,7 +507,10 @@ def concept_nearest_standard_ancestor( ) path_payload = [] - if selected_parent and selected_parent["concept_id"] != seed_concept["concept_id"]: + if ( + selected_parent + and selected_parent["concept_id"] != seed_concept["concept_id"] + ): path_info = self._graph.find_path( seed_concept["concept_id"], selected_parent["concept_id"], @@ -484,7 +585,9 @@ def concept_mapping_context( model_name: str | None = None, ) -> dict[str, Any]: if self._graph is None: - raise GroundworkersError("BACKEND_UNAVAIL", "omop_graph backend is not configured") + raise GroundworkersError( + "BACKEND_UNAVAIL", "omop_graph backend is not configured" + ) if concept_id <= 0: raise ValueError("concept_id must be a positive integer") concept = self._graph.get_concept(concept_id) @@ -498,11 +601,17 @@ def concept_mapping_context( concept["concept_code"], ) if include_ancestors: - result["ancestors"] = self._graph.get_ancestors(concept_id, max(1, min(ancestor_limit, 10)))[:ancestor_limit] + result["ancestors"] = self._graph.get_ancestors( + concept_id, max(1, min(ancestor_limit, 10)) + )[:ancestor_limit] if include_descendants: - result["descendants"] = self._graph.get_descendants(concept_id, max(1, min(descendant_limit, 10)))[:descendant_limit] + result["descendants"] = self._graph.get_descendants( + concept_id, max(1, min(descendant_limit, 10)) + )[:descendant_limit] if include_relationship_summary: - result["relationship_summary"] = self._summarise_edges(self._graph.get_edges(concept_id)) + result["relationship_summary"] = self._summarise_edges( + self._graph.get_edges(concept_id) + ) if include_neighbors: neighbors = self._graph.get_neighbors( concept_id=concept_id, @@ -515,7 +624,9 @@ def concept_mapping_context( if include_embedding_neighbors: if self._emb is None: result["embedding_neighbors"] = [] - result.setdefault("warnings", []).append("embedding adapter not configured") + result.setdefault("warnings", []).append( + "embedding adapter not configured" + ) else: result["embedding_neighbors"] = self._emb.get_neighbours( concept_id=concept_id, @@ -530,20 +641,28 @@ def concept_map_to_value( concept_code: str, ) -> dict[str, Any]: if self._graph is None: - raise GroundworkersError("BACKEND_UNAVAIL", "omop_graph backend is not configured") + raise GroundworkersError( + "BACKEND_UNAVAIL", "omop_graph backend is not configured" + ) if not vocabulary_id.strip(): raise ValueError("vocabulary_id must be a non-empty string") if not concept_code.strip(): raise ValueError("concept_code must be a non-empty string") source_list = self._graph.get_concept_by_code(vocabulary_id, concept_code) if not source_list: - raise GroundworkersError("NOT_FOUND", f"Concept {vocabulary_id}:{concept_code} was not found") + raise GroundworkersError( + "NOT_FOUND", f"Concept {vocabulary_id}:{concept_code} was not found" + ) source = source_list[0] mappings = self._vocab.navigate_to_value([source["concept_id"]]) mapping = mappings[0] if mappings else None return { "source_concept": source, - "maps_to_value": serialise_related_concept_mapping(mapping)["related_concepts"] if mapping else [], + "maps_to_value": serialise_related_concept_mapping(mapping)[ + "related_concepts" + ] + if mapping + else [], } def concept_resolve_mapping_expression( @@ -555,9 +674,17 @@ def concept_resolve_mapping_expression( resolve_to_standard: bool = True, ) -> dict[str, Any]: if self._graph is None: - raise GroundworkersError("BACKEND_UNAVAIL", "omop_graph backend is not configured") + raise GroundworkersError( + "BACKEND_UNAVAIL", "omop_graph backend is not configured" + ) if not items: - return {"expression_items": [], "resolved_concept_ids": [], "resolved_concepts": [], "excluded_concepts": [], "counts": {"resolved": 0, "excluded": 0}} + return { + "expression_items": [], + "resolved_concept_ids": [], + "resolved_concepts": [], + "excluded_concepts": [], + "counts": {"resolved": 0, "excluded": 0}, + } resolved: dict[int, dict[str, Any]] = {} excluded: dict[int, dict[str, Any]] = {} try: @@ -568,7 +695,9 @@ def concept_resolve_mapping_expression( continue concepts_to_apply = [concept] if resolve_to_standard and not concept.get("standard_concept"): - mapped = self._graph.map_to_standard(concept["vocabulary_id"], concept["concept_code"]) + mapped = self._graph.map_to_standard( + concept["vocabulary_id"], concept["concept_code"] + ) mapped_standards = mapped.get("standard_concepts", []) if mapped_standards: concepts_to_apply = mapped_standards @@ -576,10 +705,16 @@ def concept_resolve_mapping_expression( for base in concepts_to_apply: expanded.append(base) if item.get("include_descendants"): - expanded.extend(self._graph.get_descendants(base["concept_id"], 2)) + expanded.extend( + self._graph.get_descendants(base["concept_id"], 2) + ) target = excluded if item.get("exclude") else resolved for entry in expanded: - if domain and entry.get("domain_id") and str(entry["domain_id"]).lower() != domain.lower(): + if ( + domain + and entry.get("domain_id") + and str(entry["domain_id"]).lower() != domain.lower() + ): continue target[int(entry["concept_id"])] = entry if deduplicate: @@ -628,14 +763,22 @@ def mapping_evaluate_candidates( ref = reference_index.pop(key, None) predicted_ids = self._extract_predicted_ids(pred, top_k=top_k) if ref is None: - extra_prediction_cases.append({"source_key": key, "predicted_concept_ids": predicted_ids}) + extra_prediction_cases.append( + {"source_key": key, "predicted_concept_ids": predicted_ids} + ) continue reference_id = int(ref["reference_standard_concept_id"]) - domain_name = str(ref.get("domain_id") or pred.get("domain_id") or "UNKNOWN") + domain_name = str( + ref.get("domain_id") or pred.get("domain_id") or "UNKNOWN" + ) bucket = by_domain_counts.setdefault(domain_name, Counter()) if reference_id in predicted_ids: agreement_cases.append( - {"source_key": key, "reference_standard_concept_id": reference_id, "predicted_concept_ids": predicted_ids} + { + "source_key": key, + "reference_standard_concept_id": reference_id, + "predicted_concept_ids": predicted_ids, + } ) bucket["agreement"] += 1 else: @@ -651,7 +794,9 @@ def mapping_evaluate_candidates( missing_reference_cases = [ { "source_key": key, - "reference_standard_concept_id": int(row["reference_standard_concept_id"]), + "reference_standard_concept_id": int( + row["reference_standard_concept_id"] + ), "domain_id": row.get("domain_id"), } for key, row in reference_index.items() @@ -659,8 +804,12 @@ def mapping_evaluate_candidates( total_compared = len(agreement_cases) + len(disagreement_cases) summary_metrics = { - "accuracy": round(len(agreement_cases) / total_compared, 6) if total_compared else 0.0, - "coverage": round(total_compared / len(reference_mappings), 6) if reference_mappings else 0.0, + "accuracy": round(len(agreement_cases) / total_compared, 6) + if total_compared + else 0.0, + "coverage": round(total_compared / len(reference_mappings), 6) + if reference_mappings + else 0.0, "agreement_count": len(agreement_cases), "disagreement_count": len(disagreement_cases), "missing_reference_count": len(missing_reference_cases), @@ -689,10 +838,11 @@ def mapping_evaluate_candidates( def _backfill_candidate_metadata( self, candidate_union: list[dict[str, Any]], warnings: list[str] ) -> None: - """Fill identity metadata for candidates surfaced only via embedding. - """ + """Fill identity metadata for candidates surfaced only via embedding.""" incomplete_ids = [ - row["concept_id"] for row in candidate_union if row.get("vocabulary_id") is None + row["concept_id"] + for row in candidate_union + if row.get("vocabulary_id") is None ] if not incomplete_ids: return @@ -719,7 +869,9 @@ def _backfill_candidate_metadata( row["standard_concept"] = view.get("standard_concept") @staticmethod - def _build_candidate_union(channels: dict[str, dict[str, Any]], overall_limit: int) -> list[dict[str, Any]]: + def _build_candidate_union( + channels: dict[str, dict[str, Any]], overall_limit: int + ) -> list[dict[str, Any]]: union: dict[int, dict[str, Any]] = {} channel_order = ("exact", "normalized", "fulltext", "embedding") for channel_name in channel_order: @@ -737,7 +889,9 @@ def _build_candidate_union(channels: dict[str, dict[str, Any]], overall_limit: i "vocabulary_id": item.get("vocabulary_id"), "domain_id": item.get("domain_id"), "concept_class_id": item.get("concept_class_id"), - "standard_concept": item.get("standard_concept", item.get("is_standard")), + "standard_concept": item.get( + "standard_concept", item.get("is_standard") + ), "retrieved_by": [], }, ) diff --git a/src/groundworkers/services/vocab.py b/src/groundworkers/services/vocab.py index e1d6322..dd9e36d 100644 --- a/src/groundworkers/services/vocab.py +++ b/src/groundworkers/services/vocab.py @@ -20,7 +20,11 @@ from groundworkers.base.concept_payload import serialise_concept_view from groundworkers.base.domain_names import DomainNameResolver from groundworkers.base.errors import GroundworkersError -from groundworkers.base.sql import effective_schema +from groundworkers.base.sql import ( + NormalizationProfile, + effective_schema, + normalization_expression, +) logger = logging.getLogger(__name__) @@ -32,16 +36,18 @@ # Return types # --------------------------------------------------------------------------- + @dataclass class ConceptMatch: """A single candidate returned by search_exact, search_normalized, or search_fulltext.""" + #: Shared concept payload from ``base.concept_payload``, so search results #: report the same flags as every other concept-returning tool. Previously a #: parallel field list carrying ``standard_concept`` and raw #: ``invalid_reason`` but no ``classification_concept``, which meant search #: could not distinguish a classification concept from an unflagged one. concept: dict[str, Any] - match_source: str # "name" | "synonym" + match_source: str # "name" | "synonym" matched_synonym: str | None = None ts_rank: float | None = None @@ -99,10 +105,24 @@ class ConceptMappingResult: RelatedConceptMapping = ConceptMappingResult +def _require_bounded_normalized_query( + vocabulary_id: str | None, + parent_ids: list[int] | None, +) -> None: + if not (vocabulary_id or parent_ids): + raise GroundworkersError( + "INVALID_INPUT", + "search_normalized would scan the whole concept table. Pass " + "vocabulary_id or parent_ids (domain alone is not selective enough), " + "or pass allow_unindexed_scan=True to accept the scan.", + ) + + # --------------------------------------------------------------------------- # Service # --------------------------------------------------------------------------- + class VocabService: """Direct Python API for OMOP vocabulary search and concept navigation. @@ -244,7 +264,9 @@ def search_exact( Concept.domain_id, Concept.concept_class_id, Concept.is_standard_expr().label("standard_concept"), - Concept.is_classification_expr().label("classification_concept"), + Concept.is_classification_expr().label( + "classification_concept" + ), Concept.is_valid_expr().label("is_active"), ).where(func.lower(Concept.concept_name) == q.lower()), domain=domain, @@ -270,12 +292,20 @@ def search_exact( Concept.domain_id, Concept.concept_class_id, Concept.is_standard_expr().label("standard_concept"), - Concept.is_classification_expr().label("classification_concept"), + Concept.is_classification_expr().label( + "classification_concept" + ), Concept.is_valid_expr().label("is_active"), Concept_Synonym.concept_synonym_name, ) - .join(Concept_Synonym, Concept_Synonym.concept_id == Concept.concept_id) - .where(func.lower(Concept_Synonym.concept_synonym_name) == q.lower()), + .join( + Concept_Synonym, + Concept_Synonym.concept_id == Concept.concept_id, + ) + .where( + func.lower(Concept_Synonym.concept_synonym_name) + == q.lower() + ), domain=domain, vocabulary_id=vocabulary_id, standard_only=standard_only, @@ -284,10 +314,16 @@ def search_exact( ).limit(remaining) if seen_ids: - syn_stmt = syn_stmt.where(Concept.concept_id.not_in(list(seen_ids))) + syn_stmt = syn_stmt.where( + Concept.concept_id.not_in(list(seen_ids)) + ) for row in session.execute(syn_stmt).all(): - results.append(_row_to_match(row, "synonym", row.concept_synonym_name, None)) + results.append( + _row_to_match( + row, "synonym", row.concept_synonym_name, None + ) + ) except GroundworkersError: raise @@ -311,15 +347,21 @@ def search_normalized( standard_only: bool = False, active_only: bool = False, include_synonyms: bool = False, - normalization_profile: str = "verbatim", + normalization_profile: str + | NormalizationProfile = NormalizationProfile.VERBATIM, parent_ids: list[int] | None = None, remove_stop_phrases: bool = True, limit: int = 20, + allow_unindexed_scan: bool = False, ) -> list[ConceptMatch]: """Deterministic near-verbatim search after text normalization. Both the query and candidate text are normalized before comparison. Distinct from full-text search: deterministic equality, not ranked retrieval. + + Requires ``vocabulary_id`` or ``parent_ids`` because ``domain`` alone is + not selective enough to avoid a whole-table scan. Pass + ``allow_unindexed_scan=True`` to accept that scan explicitly. """ normalized_query, _steps = normalize_text_for_matching( query, @@ -329,13 +371,16 @@ def search_normalized( if not normalized_query: raise ValueError("query must be a non-empty string after normalization") + if not allow_unindexed_scan: + _require_bounded_normalized_query(vocabulary_id, parent_ids) + results: list[ConceptMatch] = [] seen_ids: set[int] = set() try: - name_expr = _normalized_sql_expr( + name_expr = normalization_expression( Concept.concept_name, - normalization_profile=normalization_profile, + profile=normalization_profile, remove_stop_phrases=remove_stop_phrases, ) @@ -349,7 +394,9 @@ def search_normalized( Concept.domain_id, Concept.concept_class_id, Concept.is_standard_expr().label("standard_concept"), - Concept.is_classification_expr().label("classification_concept"), + Concept.is_classification_expr().label( + "classification_concept" + ), Concept.is_valid_expr().label("is_active"), ).where(name_expr == normalized_query), domain=domain, @@ -366,9 +413,9 @@ def search_normalized( if include_synonyms: remaining = limit - len(results) if remaining > 0: - syn_expr = _normalized_sql_expr( + syn_expr = normalization_expression( Concept_Synonym.concept_synonym_name, - normalization_profile=normalization_profile, + profile=normalization_profile, remove_stop_phrases=remove_stop_phrases, ) syn_stmt = self._apply_concept_filters( @@ -380,11 +427,16 @@ def search_normalized( Concept.domain_id, Concept.concept_class_id, Concept.is_standard_expr().label("standard_concept"), - Concept.is_classification_expr().label("classification_concept"), + Concept.is_classification_expr().label( + "classification_concept" + ), Concept.is_valid_expr().label("is_active"), Concept_Synonym.concept_synonym_name, ) - .join(Concept_Synonym, Concept_Synonym.concept_id == Concept.concept_id) + .join( + Concept_Synonym, + Concept_Synonym.concept_id == Concept.concept_id, + ) .where(syn_expr == normalized_query), domain=domain, vocabulary_id=vocabulary_id, @@ -394,10 +446,16 @@ def search_normalized( ).limit(remaining) if seen_ids: - syn_stmt = syn_stmt.where(Concept.concept_id.not_in(list(seen_ids))) + syn_stmt = syn_stmt.where( + Concept.concept_id.not_in(list(seen_ids)) + ) for row in session.execute(syn_stmt).all(): - results.append(_row_to_match(row, "synonym", row.concept_synonym_name, None)) + results.append( + _row_to_match( + row, "synonym", row.concept_synonym_name, None + ) + ) except GroundworkersError: raise @@ -451,25 +509,31 @@ def search_fulltext( name_rank = func.ts_rank(sa_col("concept_name_tsvector"), tsquery) with self._cdm.session() as session: - name_stmt = self._apply_concept_filters( - select( - Concept.concept_id, - Concept.concept_name, - Concept.concept_code, - Concept.vocabulary_id, - Concept.domain_id, - Concept.concept_class_id, - Concept.is_standard_expr().label("standard_concept"), - Concept.is_classification_expr().label("classification_concept"), - Concept.is_valid_expr().label("is_active"), - name_rank.label("ts_rank"), - ).where(sa_col("concept_name_tsvector").op("@@")(tsquery)), - domain=domain, - vocabulary_id=vocabulary_id, - standard_only=standard_only, - active_only=active_only, - parent_ids=parent_ids, - ).order_by(name_rank.desc()).limit(limit) + name_stmt = ( + self._apply_concept_filters( + select( + Concept.concept_id, + Concept.concept_name, + Concept.concept_code, + Concept.vocabulary_id, + Concept.domain_id, + Concept.concept_class_id, + Concept.is_standard_expr().label("standard_concept"), + Concept.is_classification_expr().label( + "classification_concept" + ), + Concept.is_valid_expr().label("is_active"), + name_rank.label("ts_rank"), + ).where(sa_col("concept_name_tsvector").op("@@")(tsquery)), + domain=domain, + vocabulary_id=vocabulary_id, + standard_only=standard_only, + active_only=active_only, + parent_ids=parent_ids, + ) + .order_by(name_rank.desc()) + .limit(limit) + ) if min_rank > 0.0: name_stmt = name_stmt.where(name_rank >= min_rank) @@ -481,37 +545,63 @@ def search_fulltext( if include_synonyms and self._fts_synonym_sidecar: remaining = limit - len(results) if remaining > 0: - syn_rank = func.ts_rank(sa_col("concept_synonym_name_tsvector"), tsquery) - syn_stmt = self._apply_concept_filters( - select( - Concept.concept_id, - Concept.concept_name, - Concept.concept_code, - Concept.vocabulary_id, - Concept.domain_id, - Concept.concept_class_id, - Concept.is_standard_expr().label("standard_concept"), - Concept.is_classification_expr().label("classification_concept"), - Concept.is_valid_expr().label("is_active"), - Concept_Synonym.concept_synonym_name, - syn_rank.label("ts_rank"), + syn_rank = func.ts_rank( + sa_col("concept_synonym_name_tsvector"), tsquery + ) + syn_stmt = ( + self._apply_concept_filters( + select( + Concept.concept_id, + Concept.concept_name, + Concept.concept_code, + Concept.vocabulary_id, + Concept.domain_id, + Concept.concept_class_id, + Concept.is_standard_expr().label( + "standard_concept" + ), + Concept.is_classification_expr().label( + "classification_concept" + ), + Concept.is_valid_expr().label("is_active"), + Concept_Synonym.concept_synonym_name, + syn_rank.label("ts_rank"), + ) + .join( + Concept_Synonym, + Concept_Synonym.concept_id == Concept.concept_id, + ) + .where( + sa_col("concept_synonym_name_tsvector").op("@@")( + tsquery + ) + ), + domain=domain, + vocabulary_id=vocabulary_id, + standard_only=standard_only, + active_only=active_only, + parent_ids=parent_ids, ) - .join(Concept_Synonym, Concept_Synonym.concept_id == Concept.concept_id) - .where(sa_col("concept_synonym_name_tsvector").op("@@")(tsquery)), - domain=domain, - vocabulary_id=vocabulary_id, - standard_only=standard_only, - active_only=active_only, - parent_ids=parent_ids, - ).order_by(syn_rank.desc()).limit(remaining) + .order_by(syn_rank.desc()) + .limit(remaining) + ) if min_rank > 0.0: syn_stmt = syn_stmt.where(syn_rank >= min_rank) if seen_ids: - syn_stmt = syn_stmt.where(Concept.concept_id.not_in(list(seen_ids))) + syn_stmt = syn_stmt.where( + Concept.concept_id.not_in(list(seen_ids)) + ) for row in session.execute(syn_stmt).all(): - results.append(_row_to_match(row, "synonym", row.concept_synonym_name, float(row.ts_rank))) + results.append( + _row_to_match( + row, + "synonym", + row.concept_synonym_name, + float(row.ts_rank), + ) + ) except GroundworkersError: raise @@ -551,7 +641,9 @@ def navigate_to_standard(self, concept_ids: list[int]) -> list[StandardMapping]: Concept.is_valid_expr().label("is_active"), ).where(Concept.concept_id.in_(concept_ids)) - source_rows = {int(r.concept_id): r for r in session.execute(source_stmt).all()} + source_rows = { + int(r.concept_id): r for r in session.execute(source_stmt).all() + } non_standard_ids = [ cid for cid, r in source_rows.items() if not r.standard_concept @@ -570,13 +662,20 @@ def navigate_to_standard(self, concept_ids: list[int]) -> list[StandardMapping]: Concept.domain_id, Concept.concept_class_id, Concept.is_standard_expr().label("standard_concept"), - Concept.is_classification_expr().label("classification_concept"), + Concept.is_classification_expr().label( + "classification_concept" + ), Concept.is_valid_expr().label("is_active"), ) - .join(Concept, Concept.concept_id == Concept_Relationship.concept_id_2) + .join( + Concept, + Concept.concept_id == Concept_Relationship.concept_id_2, + ) .where( Concept_Relationship.concept_id_1.in_(non_standard_ids), - Concept_Relationship.relationship_id.in_(self.IDENTITY_RELATIONSHIP_IDS), + Concept_Relationship.relationship_id.in_( + self.IDENTITY_RELATIONSHIP_IDS + ), Concept_Relationship.is_valid_expr(), Concept.is_standard_expr(), ) @@ -720,7 +819,9 @@ def _navigate_relationship( Concept.is_valid_expr().label("is_active"), ).where(Concept.concept_id.in_(concept_ids)) - source_rows = {int(r.concept_id): r for r in session.execute(source_stmt).all()} + source_rows = { + int(r.concept_id): r for r in session.execute(source_stmt).all() + } related: dict[int, list[MappedConcept]] = {} if source_rows: @@ -735,12 +836,19 @@ def _navigate_relationship( Concept.domain_id, Concept.concept_class_id, Concept.is_standard_expr().label("standard_concept"), - Concept.is_classification_expr().label("classification_concept"), + Concept.is_classification_expr().label( + "classification_concept" + ), Concept.is_valid_expr().label("is_active"), ) - .join(Concept, Concept.concept_id == Concept_Relationship.concept_id_2) + .join( + Concept, + Concept.concept_id == Concept_Relationship.concept_id_2, + ) .where( - Concept_Relationship.concept_id_1.in_(list(source_rows.keys())), + Concept_Relationship.concept_id_1.in_( + list(source_rows.keys()) + ), Concept_Relationship.relationship_id.in_(relationship_ids), Concept_Relationship.is_valid_expr(), ) @@ -783,6 +891,7 @@ def _navigate_relationship( # Serialisation helpers # --------------------------------------------------------------------------- + def serialise_concept_match(match: ConceptMatch) -> dict: """Serialise a ConceptMatch to a JSON-safe dict for MCP tool responses.""" result: dict = { @@ -838,10 +947,11 @@ def serialise_related_concept_mapping(mapping: ConceptMappingResult) -> dict: def normalize_text_for_matching( text: str, *, - profile: str = "verbatim", + profile: str | NormalizationProfile = NormalizationProfile.VERBATIM, remove_stop_phrases: bool = True, ) -> tuple[str, list[str]]: """Normalize free text into a deterministic matching form.""" + profile = NormalizationProfile(profile) steps: list[str] = ["strip", "lowercase", "collapse_whitespace"] normalised = _WHITESPACE_RE.sub(" ", text.strip().lower()) @@ -862,19 +972,9 @@ def normalize_text_for_matching( return normalised, steps -def _normalized_sql_expr(column, *, normalization_profile: str, remove_stop_phrases: bool): - expr = func.lower(column) - if remove_stop_phrases: - expr = func.regexp_replace(expr, r'\m(?:nos|nec|nfs|unspecified|unknown|other|w/?o|w/)\M', " ", "g") - expr = func.regexp_replace(expr, r"[^a-z0-9]+", " ", "g") - if normalization_profile == "drug_name": - expr = func.replace(expr, " extended release ", " ") - expr = func.replace(expr, " modified release ", " ") - expr = func.regexp_replace(expr, r"\s+", " ", "g") - return func.btrim(expr) - - -def _row_to_match(row, match_source: str, matched_synonym: str | None, ts_rank: float | None) -> ConceptMatch: +def _row_to_match( + row, match_source: str, matched_synonym: str | None, ts_rank: float | None +) -> ConceptMatch: return ConceptMatch( concept=serialise_concept_view(row, detail="flags"), match_source=match_source, diff --git a/src/groundworkers/tools/mapping_tools.py b/src/groundworkers/tools/mapping_tools.py index 565f036..bdca936 100644 --- a/src/groundworkers/tools/mapping_tools.py +++ b/src/groundworkers/tools/mapping_tools.py @@ -4,10 +4,13 @@ from groundworkers.base.errors import GroundworkersError from groundworkers.base.server import GroundworkersMCPServer +from groundworkers.base.sql import NormalizationProfile from groundworkers.services import MappingService -def register_mapping_tools(server: GroundworkersMCPServer, mapping_service: MappingService) -> None: +def register_mapping_tools( + server: GroundworkersMCPServer, mapping_service: MappingService +) -> None: @server.tool("concept_search_normalized") def concept_search_normalized( query: str, @@ -15,15 +18,21 @@ def concept_search_normalized( vocabulary_id: str | None = None, standard_only: bool = False, include_synonyms: bool = False, - normalization_profile: str = "verbatim", + normalization_profile: NormalizationProfile = NormalizationProfile.VERBATIM, remove_stop_phrases: bool = True, limit: int = 20, + allow_unindexed_scan: bool = False, ) -> dict[str, Any]: """Deterministic near-verbatim concept search after text normalization. Both the query and candidate concept names are normalized before comparison, making this more robust to punctuation and whitespace variation than concept_search_exact while remaining fully deterministic (no ranking). + + Comparison is on a computed expression, so this tool needs a vocabulary_id + filter; otherwise it scans every concept and the call is refused. A domain + filter is not selective enough to qualify. Set allow_unindexed_scan to + accept the scan deliberately. """ safe_limit = max(1, min(limit, 50)) try: @@ -36,6 +45,7 @@ def concept_search_normalized( normalization_profile=normalization_profile, remove_stop_phrases=remove_stop_phrases, limit=safe_limit, + allow_unindexed_scan=allow_unindexed_scan, ) except ValueError as exc: return {"error": True, "code": "INVALID_INPUT", "message": str(exc)} diff --git a/src/groundworkers/tui/pages/setup.py b/src/groundworkers/tui/pages/setup.py index df75090..8bd323a 100644 --- a/src/groundworkers/tui/pages/setup.py +++ b/src/groundworkers/tui/pages/setup.py @@ -406,7 +406,9 @@ def row_selected(self, row_key: str, context: PageContext) -> None: row = next((item for item in view.rows if item.key == row_key), None) if row is None or row.detail is None: return - detail = key_value_detail(f"{self._selected_section_title()} detail", row.detail) + detail = key_value_detail( + f"{self._selected_section_title()} detail", row.detail + ) if detail is not None: context.surface.show_detail(self.route.key, detail) @@ -425,17 +427,24 @@ def action_selected(self, action_key: str, context: PageContext) -> None: self._start_verify_all(context) return if action_key == "overview.integration": - output = self._integration_output() if self._session.databases_connected else None + output = ( + self._integration_output() + if self._session.databases_connected + else None + ) if output is None: - context.notify("Verify the CDM before showing integration commands.", severity="warning") + context.notify( + "Verify the CDM before showing integration commands.", + severity="warning", + ) else: - context.notify(f"stdio: {output.stdio_command}\nHTTP: {output.http_command}") + context.notify( + f"stdio: {output.stdio_command}\nHTTP: {output.http_command}" + ) return if action_key == "graph.prepare": context.open_wizard( - GraphMaintenanceWizardController( - self._session, self._graph_readiness() - ) + GraphMaintenanceWizardController(self._session, self._graph_readiness()) ) return if action_key == "performance.refresh": @@ -622,7 +631,9 @@ def verify() -> None: self.run_worker(verify, thread=True, exclusive=True) - def _finish_performance_checks(self, results, coverage, context: PageContext) -> None: + def _finish_performance_checks( + self, results, coverage, context: PageContext + ) -> None: self._session.connection_results = tuple(results) self._session.embedding_coverage = coverage self._show_current_state(context) @@ -698,7 +709,11 @@ def _trigram_available(self) -> bool: def _graph_target_result(self, target_key: str): return next( - (result for result in self._session.connection_results if result.target_key == target_key), + ( + result + for result in self._session.connection_results + if result.target_key == target_key + ), None, ) @@ -716,7 +731,9 @@ def _integration_output(self): return build_integration_output(self._session.configuration) def _integration_ready(self) -> bool: - return self._session.databases_connected and self._integration_output() is not None + return ( + self._session.databases_connected and self._integration_output() is not None + ) def _start_llm_provider_check(self, context: PageContext) -> None: context.notify("Checking LLM provider endpoint and model.") @@ -855,8 +872,7 @@ def _show_section_detail(self, context: PageContext) -> None: ) tail = collapse_tqdm_tail(tail) step_label = ( - f"{step_index + 1}/{run.total} · " - f"{run.steps[step_index].spec.key}" + f"{step_index + 1}/{run.total} · {run.steps[step_index].spec.key}" if step_index is not None else "No step has started" ) @@ -911,7 +927,9 @@ def _copy_selected_run_log(self, context: PageContext) -> None: step_index = _selected_log_step(run) path = None if step_index is None else run.steps[step_index].log_path if path is None: - context.notify("No log output is available for this run.", severity="warning") + context.notify( + "No log output is available for this run.", severity="warning" + ) return try: text = read_log_tail(path) diff --git a/src/groundworkers/tui/presenters/database.py b/src/groundworkers/tui/presenters/database.py index b02ab47..7238203 100644 --- a/src/groundworkers/tui/presenters/database.py +++ b/src/groundworkers/tui/presenters/database.py @@ -99,10 +99,7 @@ def landing( ), ) - result_by_key = { - item.target_key: _database_result(item) - for item in results - } + result_by_key = {item.target_key: _database_result(item) for item in results} rows = [ _target_row(target, result_by_key.get(target.key)) for target in targets ] @@ -171,7 +168,9 @@ def _database_result(result: ConnectionResult) -> ConnectionResult: ) -def _database_results(results: Sequence[ConnectionResult]) -> tuple[ConnectionResult, ...]: +def _database_results( + results: Sequence[ConnectionResult], +) -> tuple[ConnectionResult, ...]: return tuple(_database_result(result) for result in results) diff --git a/src/groundworkers/tui/presenters/performance.py b/src/groundworkers/tui/presenters/performance.py index 1840161..a3a3cdb 100644 --- a/src/groundworkers/tui/presenters/performance.py +++ b/src/groundworkers/tui/presenters/performance.py @@ -18,6 +18,37 @@ ) from groundworkers.tui.presenters.base import SetupPresenterBase +_INDEX_DIAGNOSTICS = ( + ( + "performance.graph.fulltext", + "Graph", + "Full-text indexes", + "database.graph", + ( + "fulltext_sidecar_missing", + "fulltext_indexes_missing", + "fulltext_sidecar_unpopulated", + ), + "fulltext_indexes_present", + ), + ( + "performance.graph.functional", + "Graph", + "Functional text indexes", + "database.graph", + ("functional_indexes_missing",), + "functional_indexes_present", + ), + ( + "performance.groundworkers.trigram", + "Groundworkers", + "Trigram indexes", + "database.groundworkers", + ("trigram_indexes_missing", "trigram_indexes_unchecked"), + "trigram_indexes_present", + ), +) + class PerformancePresenter(SetupPresenterBase): """Present index readiness without mixing it into database setup.""" @@ -87,7 +118,8 @@ def landing( ), message=( "Graph indexes are prepared from Graph Setup. Groundworkers trigram " - "and embedding indexes can be prepared here when their backends support it." + "and embedding indexes can be prepared here when their backends " + "support it." ), ) @@ -99,39 +131,40 @@ def _statuses( embedding_coverage: EmbeddingCoverageReport | None, ) -> tuple[tuple[tuple[str, str, str, str], SemanticStatus], ...]: by_key = {result.target_key: result for result in connections} - graph = by_key.get("database.graph") - groundworkers = by_key.get("database.groundworkers") - graph_fulltext = _diagnostic_outcome( - graph, - ( - "fulltext_sidecar_missing", - "fulltext_indexes_missing", - "fulltext_sidecar_unpopulated", - ), - present_code="fulltext_indexes_present", - ) - graph_functional = _diagnostic_outcome( - graph, - ("functional_indexes_missing",), - present_code="functional_indexes_present", - ) - trigram = _diagnostic_outcome( - groundworkers, - ("trigram_indexes_missing", "trigram_indexes_unchecked"), - present_code="trigram_indexes_present", + index_rows = tuple( + _status_row( + key, + area, + index, + _diagnostic_outcome( + by_key.get(target_key), + missing_codes, + present_code=present_code, + ), + ) + for key, area, index, target_key, missing_codes, present_code in _INDEX_DIAGNOSTICS ) embedding = _embedding_index_outcome( embedding_configuration, embedding_coverage, ) return ( - (("performance.graph.fulltext", "Graph", "Full-text indexes", graph_fulltext[0]), graph_fulltext[1]), - (("performance.graph.functional", "Graph", "Functional text indexes", graph_functional[0]), graph_functional[1]), - (("performance.groundworkers.trigram", "Groundworkers", "Trigram indexes", trigram[0]), trigram[1]), - (("performance.embeddings.index", "Embeddings", "Vector index", embedding[0]), embedding[1]), + *index_rows, + _status_row( + "performance.embeddings.index", "Embeddings", "Vector index", embedding + ), ) +def _status_row( + key: str, + area: str, + index: str, + outcome: tuple[str, SemanticStatus], +) -> tuple[tuple[str, str, str, str], SemanticStatus]: + return ((key, area, index, outcome[0]), outcome[1]) + + def _diagnostic_outcome( result: ConnectionResult | None, missing_codes: tuple[str, ...], @@ -143,7 +176,10 @@ def _diagnostic_outcome( if not result.connected: return "Connection failed", SemanticStatus.ERROR codes = {diagnostic.code: diagnostic for diagnostic in result.diagnostics} - if any(code in codes and codes[code].severity is DiagnosticSeverity.WARNING for code in missing_codes): + if any( + code in codes and codes[code].severity is DiagnosticSeverity.WARNING + for code in missing_codes + ): return "Missing", SemanticStatus.WARNING if present_code in codes: return "Ready", SemanticStatus.OK diff --git a/src/groundworkers/tui/routes.py b/src/groundworkers/tui/routes.py index 8239436..7aaa43b 100644 --- a/src/groundworkers/tui/routes.py +++ b/src/groundworkers/tui/routes.py @@ -6,4 +6,4 @@ purpose="Configure and verify the services Groundworkers uses.", ) -__all__ = ["SETUP_ROUTE"] \ No newline at end of file +__all__ = ["SETUP_ROUTE"] diff --git a/src/groundworkers/tui/wizards/performance_maintenance.py b/src/groundworkers/tui/wizards/performance_maintenance.py index 848f8c7..cd71443 100644 --- a/src/groundworkers/tui/wizards/performance_maintenance.py +++ b/src/groundworkers/tui/wizards/performance_maintenance.py @@ -103,8 +103,14 @@ def review(self) -> WizardTransition: def apply(self) -> WizardResult: try: commands = build_performance_commands( - tuple(remediation for remediation in PerformanceRemediation if remediation in self._selected), - embedding_model=(self._coverage.index.model_name if self._coverage else None), + tuple( + remediation + for remediation in PerformanceRemediation + if remediation in self._selected + ), + embedding_model=( + self._coverage.index.model_name if self._coverage else None + ), config_path=self._session.configuration.path, ) if not commands: @@ -171,11 +177,15 @@ def _review_step(self) -> ReviewStep: title="Confirm", review=WizardReview( changes=tuple( - ReviewChange(field=_LABELS[item], before="available", after="will run") + ReviewChange( + field=_LABELS[item], before="available", after="will run" + ) for item in PerformanceRemediation if item in self._selected ), - effects=("Each index build runs as a background command with its own log.",), + effects=( + "Each index build runs as a background command with its own log.", + ), warnings=( "Index builds can be slow and temporarily use substantial database resources.", ), @@ -192,14 +202,18 @@ def _snapshot(self, *, issues: tuple[ValidationIssue, ...] = ()) -> WizardSnapsh step=step, step_index=self._step_index, step_count=len(steps), - values={} if isinstance(step, ReviewStep) else { + values={} + if isinstance(step, ReviewStep) + else { key: remediation in self._selected for remediation, key in _FIELD_BY_REMEDIATION.items() }, issues=issues, can_back=self._step_index > 0, can_next=not isinstance(step, ReviewStep), - can_apply=isinstance(step, ReviewStep) and step.review.ready_to_apply and not issues, + can_apply=isinstance(step, ReviewStep) + and step.review.ready_to_apply + and not issues, expected_revision=self._session.configuration.revision, ) @@ -210,12 +224,21 @@ def _submit_form(self, values: Mapping[str, object]) -> tuple[ValidationIssue, . else: self._selected.discard(remediation) if not self._selected: - return (ValidationIssue("Select at least one index to prepare.", "create_trigram_indexes"),) + return ( + ValidationIssue( + "Select at least one index to prepare.", "create_trigram_indexes" + ), + ) return () def _as_bool(value: object) -> bool: - return (isinstance(value, bool) and value) or str(value).strip().lower() in {"true", "yes", "1", "on"} + return (isinstance(value, bool) and value) or str(value).strip().lower() in { + "true", + "yes", + "1", + "on", + } __all__ = ["PerformanceMaintenanceWizardController"] diff --git a/tests/unit/test_mapping_context.py b/tests/unit/test_mapping_context.py new file mode 100644 index 0000000..4b84f5b --- /dev/null +++ b/tests/unit/test_mapping_context.py @@ -0,0 +1,90 @@ +import asyncio + +import pytest + +from groundworkers.app import build_application +from groundworkers.bootstrap import build_app_config_from_stack +from groundworkers.plugins import MappingContext +from tests.support.stack_config import build_cdm_stack + + +class StubMappingService: + def concept_candidate_bundle(self, query: str, **kwargs): + return {"query": query, "kwargs": kwargs} + + async def async_concept_candidate_bundle(self, query: str, **kwargs): + return {"query": query, "kwargs": kwargs, "async": True} + + def concept_mapping_context(self, concept_id: int, **kwargs): + return {"concept_id": concept_id, "kwargs": kwargs} + + +class StubGroundingService: + def ground(self, query: str, **kwargs): + return {"query": query, "kwargs": kwargs} + + +def test_mapping_context_delegates_to_the_host_service(): + context = MappingContext(mapping_service=StubMappingService()) + + assert context.available is True + assert context.candidate_bundle("diabetes", limit=3) == { + "query": "diabetes", + "kwargs": {"limit": 3}, + } + assert context.concept_mapping_context(201826, include_ancestors=False) == { + "concept_id": 201826, + "kwargs": {"include_ancestors": False}, + } + + +def test_mapping_context_exposes_async_candidate_retrieval(): + context = MappingContext(mapping_service=StubMappingService()) + + assert asyncio.run(context.async_candidate_bundle("diabetes")) == { + "query": "diabetes", + "kwargs": {}, + "async": True, + } + + +def test_mapping_context_reports_unconfigured_service(): + context = MappingContext() + + assert context.available is False + with pytest.raises(RuntimeError, match="mapping service is not configured"): + context.candidate_bundle("diabetes") + + +def test_mapping_context_exposes_grounding_without_exposing_service_internals(): + context = MappingContext(grounding_service=StubGroundingService()) + + assert context.grounding_available is True + assert context.ground("diabetes", limit=3) == { + "query": "diabetes", + "kwargs": {"limit": 3}, + } + assert not hasattr(context, "grounding_service") + + +class ContextPlugin: + name = "context_plugin" + config_cls = None + + def build(self, context, config): + assert config is None + return context.mapping_context + + def register(self, server, state): + del server, state + + +def test_application_wires_one_shared_mapping_context_into_plugins(monkeypatch): + monkeypatch.setattr("groundworkers.app.discover_plugins", lambda: [ContextPlugin()]) + + app = build_application(build_app_config_from_stack(build_cdm_stack())) + + context = app.plugins["context_plugin"] + assert isinstance(context, MappingContext) + assert context.available is True + assert context.grounding_available is True diff --git a/tests/unit/test_mapping_service.py b/tests/unit/test_mapping_service.py index 5b4cebe..fc14c7f 100644 --- a/tests/unit/test_mapping_service.py +++ b/tests/unit/test_mapping_service.py @@ -29,7 +29,7 @@ def search_exact(self, query: str, *, domain=None, vocabulary_id=None, standard_ ) ] - def search_normalized(self, query: str, *, domain=None, vocabulary_id=None, standard_only=False, active_only=False, include_synonyms=False, normalization_profile="verbatim", parent_ids=None, remove_stop_phrases=True, limit=20): + def search_normalized(self, query: str, *, domain=None, vocabulary_id=None, standard_only=False, active_only=False, include_synonyms=False, normalization_profile="verbatim", parent_ids=None, remove_stop_phrases=True, limit=20, allow_unindexed_scan=False): return [ ConceptMatch( concept={ @@ -349,6 +349,7 @@ def test_concept_candidate_bundle_combines_channels_and_standard_mappings(): result = service.concept_candidate_bundle( "type 2 diabetes", + vocabulary_id="ICD10CM", include_hierarchy_context=True, include_relationship_summary=True, ) @@ -531,7 +532,7 @@ def _service_with_recording_vocab(emb=None): def test_active_only_true_propagated_to_all_lexical_channels(): service, vocab = _service_with_recording_vocab(emb=StubEmbAdapter()) - service.concept_candidate_bundle("diabetes", active_only=True) + service.concept_candidate_bundle("diabetes", vocabulary_id="ICD10CM", active_only=True) assert vocab.received["exact"]["active_only"] is True assert vocab.received["normalized"]["active_only"] is True @@ -541,7 +542,7 @@ def test_active_only_true_propagated_to_all_lexical_channels(): def test_active_only_false_propagated_to_all_lexical_channels(): service, vocab = _service_with_recording_vocab(emb=StubEmbAdapter()) - service.concept_candidate_bundle("diabetes", active_only=False) + service.concept_candidate_bundle("diabetes", vocabulary_id="ICD10CM", active_only=False) assert vocab.received["exact"]["active_only"] is False assert vocab.received["normalized"]["active_only"] is False @@ -565,7 +566,7 @@ def test_parent_ids_propagated_to_all_lexical_channels(): def test_parent_ids_none_propagated_when_not_specified(): service, vocab = _service_with_recording_vocab(emb=StubEmbAdapter()) - service.concept_candidate_bundle("diabetes") + service.concept_candidate_bundle("diabetes", vocabulary_id="ICD10CM") assert vocab.received["exact"]["parent_ids"] is None assert vocab.received["normalized"]["parent_ids"] is None @@ -724,3 +725,22 @@ def test_nearest_standard_ancestor_still_succeeds_through_ancestry(): assert result["found"] is True assert result["selected_parent"]["concept_id"] == 301 assert result["warnings"] == [] + + +def test_candidate_bundle_omits_the_normalized_channel_when_it_would_scan(): + service = MappingService(StubVocabAdapter(), graph_service=StubGraphAdapter()) + + result = service.concept_candidate_bundle("diabetes") + + assert result["channels"]["normalized"]["available"] is False + assert result["channels"]["normalized"]["results"] == [] + assert any("normalized channel omitted" in warning for warning in result["warnings"]) + + +def test_candidate_bundle_keeps_the_normalized_channel_when_a_vocabulary_bounds_it(): + service = MappingService(StubVocabAdapter(), graph_service=StubGraphAdapter()) + + result = service.concept_candidate_bundle("diabetes", vocabulary_id="SNOMED") + + assert result["channels"]["normalized"]["available"] is True + assert not any("normalized channel omitted" in warning for warning in result["warnings"]) diff --git a/tests/unit/test_mapping_tools.py b/tests/unit/test_mapping_tools.py index cb2d906..1c54d0f 100644 --- a/tests/unit/test_mapping_tools.py +++ b/tests/unit/test_mapping_tools.py @@ -69,6 +69,7 @@ def test_concept_search_normalized_clamps_limit_and_calls_service(): "normalization_profile": "verbatim", "remove_stop_phrases": True, "limit": 50, + "allow_unindexed_scan": False, }, ) ] diff --git a/tests/unit/test_performance.py b/tests/unit/test_performance.py index 6b8c5c6..9d7bf09 100644 --- a/tests/unit/test_performance.py +++ b/tests/unit/test_performance.py @@ -62,4 +62,7 @@ def test_performance_commands_run_trigram_before_embedding_index() -> None: assert commands[0].argv[-1] == "trigram" assert commands[1].argv[1:3] == ("maintenance", "rebuild-index") assert "--index-type" in commands[1].argv - assert all(command.environment == (("OA_CONFIG_PATH", "/tmp/config.toml"),) for command in commands) + assert all( + command.environment == (("OA_CONFIG_PATH", "/tmp/config.toml"),) + for command in commands + ) diff --git a/tests/unit/test_performance_maintenance_wizard.py b/tests/unit/test_performance_maintenance_wizard.py new file mode 100644 index 0000000..fa58483 --- /dev/null +++ b/tests/unit/test_performance_maintenance_wizard.py @@ -0,0 +1,24 @@ +from pathlib import Path + +from oa_configurator import save_stack_config + +from groundworkers.tui.state import SetupSession +from groundworkers.tui.wizards.performance_maintenance import ( + PerformanceMaintenanceWizardController, +) +from tests.support.stack_config import build_cdm_stack + + +def test_postgres_trigram_action_is_available(tmp_path: Path) -> None: + config_path = tmp_path / "config.toml" + save_stack_config(build_cdm_stack(), config_path) + controller = PerformanceMaintenanceWizardController( + SetupSession(config_path=config_path), + embedding_coverage=None, + trigram_available=True, + ) + + values = controller.start().values + + assert values["create_trigram_indexes"] is True + assert values["create_embedding_index"] is False diff --git a/tests/unit/test_setup_databases.py b/tests/unit/test_setup_databases.py index 4ace03a..71797ee 100644 --- a/tests/unit/test_setup_databases.py +++ b/tests/unit/test_setup_databases.py @@ -83,9 +83,7 @@ def test_distinct_vocabulary_connection_is_refused_rather_than_targeted( snapshot = load_configuration(config_path=path) assert snapshot.state is ConfigurationState.INCOMPLETE - assert [issue.code for issue in snapshot.issues] == [ - "vocabulary_connection_split" - ] + assert [issue.code for issue in snapshot.issues] == ["vocabulary_connection_split"] assert resolve_database_targets(snapshot) == () diff --git a/tests/unit/test_vocab_service.py b/tests/unit/test_vocab_service.py index 7e84d9b..363bbf8 100644 --- a/tests/unit/test_vocab_service.py +++ b/tests/unit/test_vocab_service.py @@ -3,17 +3,21 @@ import re from datetime import date +import pytest from omop_alchemy.cdm.model.vocabulary import ( Concept, Concept_Ancestor, Concept_Relationship, Concept_Synonym, ) +from sqlalchemy import column as sa_column from sqlalchemy import create_engine, event from groundworkers.adapters.cdm import CDMAdapter +from groundworkers.base.errors import GroundworkersError from groundworkers.base.server import GroundworkersMCPServer -from groundworkers.services.vocab import VocabService +from groundworkers.base.sql import normalization_expression +from groundworkers.services.vocab import VocabService, normalize_text_for_matching from groundworkers.tools.search_tools import register_search_tools @@ -48,7 +52,13 @@ def regexp_replace(value, pattern, replacement, _flags): ) connection.execute( Concept_Synonym.__table__.insert(), - ({"concept_id": 1, "concept_synonym_name": "Diabetes", "language_concept_id": 0},), + ( + { + "concept_id": 1, + "concept_synonym_name": "Diabetes", + "language_concept_id": 0, + }, + ), ) connection.execute( Concept_Relationship.__table__.insert(), @@ -88,7 +98,9 @@ def _concept( } -def test_vocab_service_exact_normalized_sidecar_and_standard_navigation(tmp_path) -> None: +def test_vocab_service_exact_normalized_sidecar_and_standard_navigation( + tmp_path, +) -> None: service = _service(tmp_path) exact = service.search_exact("diabetes", include_synonyms=True) @@ -116,7 +128,9 @@ def test_vocab_service_exact_normalized_sidecar_and_standard_navigation(tmp_path assert "concept_code" in target and "is_active" in target -def test_fulltext_probe_uses_the_first_available_concept_label(tmp_path, monkeypatch) -> None: +def test_fulltext_probe_uses_the_first_available_concept_label( + tmp_path, monkeypatch +) -> None: service = _service(tmp_path) queried: list[str] = [] @@ -146,3 +160,62 @@ def test_search_tools_expose_the_characterized_vocab_service(tmp_path) -> None: "results": [], } assert navigation["results"][0]["standard_concepts"][0]["concept_id"] == 1 + + +# --------------------------------------------------------------------------- +# Normalized-search affordability guard +# --------------------------------------------------------------------------- + + +def test_normalized_search_is_refused_when_it_would_scan_the_whole_table( + tmp_path, +) -> None: + """No index and no selective filter is a table scan, so it is refused. + + The unindexed query succeeds slowly rather than failing, so returning + results here would hide the cost instead of reporting it. + """ + service = _service(tmp_path) + + with pytest.raises(GroundworkersError) as excinfo: + service.search_normalized("diabetes mellitus") + + assert "vocabulary_id or parent_ids" in excinfo.value.message + assert "allow_unindexed_scan" in excinfo.value.message + + +def test_a_domain_filter_alone_does_not_satisfy_the_guard(tmp_path) -> None: + """domain looks selective and is not: measured at 25s against a real CDM.""" + service = _service(tmp_path) + + with pytest.raises(GroundworkersError): + service.search_normalized("diabetes mellitus", domain="Condition") + + +@pytest.mark.parametrize( + "kwargs", + [ + {"vocabulary_id": "ICD10CM"}, + {"parent_ids": [1]}, + {"allow_unindexed_scan": True}, + ], + ids=["vocabulary_id", "parent_ids", "explicit_opt_in"], +) +def test_normalized_search_runs_when_bounded_or_explicitly_allowed( + tmp_path, kwargs +) -> None: + service = _service(tmp_path) + + # Any of these makes the call affordable or accepts the cost deliberately. + assert isinstance(service.search_normalized("diabetes mellitus", **kwargs), list) + + +def test_unknown_normalization_profile_is_rejected_by_python_and_sql() -> None: + with pytest.raises(ValueError, match="NormalizationProfile"): + normalize_text_for_matching("diabetes", profile="invented") + with pytest.raises(ValueError, match="NormalizationProfile"): + normalization_expression( + sa_column("concept_name"), + profile="invented", + remove_stop_phrases=True, + )