fix(mcp): hygiene issues review feedback (TG-1029)

Round of review fixes applied on top of the initial implementation:

* Drop the legacy `profile_run_id` from the MCP layer — model methods take
  a `job_execution_id` and JOIN `ProfilingRun` to filter. New
  `ProfilingRun.get_latest_complete_je_id_for_table_group` avoids reading the
  `table_groups.last_complete_profile_run_id` cache (which still points at the
  internal run PK; schema-level cleanup tracked separately).
* Trust the run/JE backfill migration — drop `if x is None` guards and the
  descriptive-string fallback in `_resolve_profile_run`.
* StrEnums for fixed string sets: `Disposition`, `IssueLikelihood`, `PiiRisk`
  (in `common/models/hygiene_issue.py`) and `QualityDimension` (new shared
  `common/enums.py`). Parsers return enum types; coalesce defaults reference
  enum members; sentinel sets dropped.
* Tighten labels: heading uses "for profiling run"; `get_hygiene_issue` no
  longer triples "Run / Profiling Run / Run Date".
* Empty states render with title + italic marker.
* `dq_prevalence` removed from output and dataclasses (no precedent label,
  unhelpful score-engine internal).

Author: rboni-dk

testgen/common/enums.py

@@ -0,0 +1,29 @@
+"""Shared enums used across multiple models, services, and surfaces.
+
+Add an enum here when its values are referenced by more than one model file or by
+both the model layer and an outer surface (MCP, API, UI). Single-model enums live
+in their model file.
+"""
+from enum import StrEnum
+
+
+class QualityDimension(StrEnum):
+    """Stored ``dq_dimension`` values shared by ``profile_anomaly_types`` and ``test_types``.
+    Surfaced to users as "Quality Dimension"."""
+    ACCURACY = "Accuracy"
+    COMPLETENESS = "Completeness"
+    CONSISTENCY = "Consistency"
+    RECENCY = "Recency"
+    TIMELINESS = "Timeliness"
+    UNIQUENESS = "Uniqueness"
+    VALIDITY = "Validity"
+
+
+class ImpactDimension(StrEnum):
+    """Stored ``impact_dimension`` values shared by ``profile_anomaly_types`` /
+    ``profile_anomaly_results`` and ``test_types``. The primary dimension breakdown
+    used by scorecards."""
+    RELIABILITY = "Reliability"
+    CONFORMANCE = "Conformance"
+    REGULARITY = "Regularity"
+    USABILITY = "Usability"

testgen/common/models/hygiene_issue.py

@@ -2,6 +2,7 @@
 from collections.abc import Iterable
 from dataclasses import dataclass
 from datetime import datetime
+from enum import StrEnum
 from typing import Self
 from uuid import UUID, uuid4
 
@@ -21,6 +22,28 @@
 PII_RISK_RE = re.compile(r"Risk: (MODERATE|HIGH),")
 
 
+class Disposition(StrEnum):
+    """Stored disposition values for ``profile_anomaly_results.disposition`` and
+    ``test_results.disposition``. The user-facing label for ``INACTIVE`` is "Muted"."""
+    CONFIRMED = "Confirmed"
+    DISMISSED = "Dismissed"
+    INACTIVE = "Inactive"
+
+
+class IssueLikelihood(StrEnum):
+    """Stored ``profile_anomaly_types.issue_likelihood`` values."""
+    DEFINITE = "Definite"
+    LIKELY = "Likely"
+    POSSIBLE = "Possible"
+    POTENTIAL_PII = "Potential PII"
+
+
+class PiiRisk(StrEnum):
+    """Risk level extracted from PII issue ``detail`` strings via ``priority`` hybrid."""
+    HIGH = "High"
+    MODERATE = "Moderate"
+
+
 @dataclass
 class IssueLikelihoodCounts:
     """Counts of hygiene issues by likelihood category, with dismissed/inactive separated."""
@@ -51,6 +74,7 @@ class HygieneIssueListRow:
     schema_name: str
     table_name: str
     column_name: str
+    impact_dimension: str | None
     dq_dimension: str | None
     disposition: str
     priority: str | None
@@ -72,6 +96,7 @@ class HygieneIssueSearchRow:
     schema_name: str
     table_name: str
     column_name: str
+    impact_dimension: str | None
     dq_dimension: str | None
     disposition: str
     priority: str | None
@@ -92,7 +117,6 @@ class HygieneIssueDetail:
     schema_name: str
     table_name: str
     column_name: str
-    db_data_type: str | None
     dq_dimension: str | None
     impact_dimension: str | None
     disposition: str
@@ -150,13 +174,12 @@ class HygieneIssue(Entity):
     schema_name: str = Column(String, nullable=False)
     table_name: str = Column(String, nullable=False)
     column_name: str = Column(String, nullable=False)
-    db_data_type: str = Column(String)
 
     detail: str = Column(String, nullable=False)
     disposition: str = Column(String)
     impact_dimension: str = Column(String)
 
-    # Unmapped: column_type, dq_prevalence.
+    # Unmapped: column_type, db_data_type, dq_prevalence.
 
     @hybrid_property
     def priority(self):
@@ -237,12 +260,12 @@ def _priority_order(cls):
     @classmethod
     def list_for_run(
         cls,
-        profile_run_id: UUID,
+        job_execution_id: UUID,
         *clauses,
         page: int = 1,
         limit: int = 50,
     ) -> tuple[list[HygieneIssueListRow], int]:
-        """Paginated hygiene issues for a single profiling run.
+        """Paginated hygiene issues for a single profiling run, scoped by its job_execution_id.
 
         Caller-supplied ``*clauses`` carry every WHERE filter (project scoping, disposition,
         likelihood / pii_risk, table / column / dq_dimension / issue_type filters).
@@ -255,14 +278,16 @@ def list_for_run(
                 cls.schema_name.label("schema_name"),
                 cls.table_name.label("table_name"),
                 cls.column_name.label("column_name"),
+                cls.impact_dimension.label("impact_dimension"),
                 HygieneIssueType.dq_dimension.label("dq_dimension"),
-                func.coalesce(cls.disposition, "Confirmed").label("disposition"),
+                func.coalesce(cls.disposition, Disposition.CONFIRMED).label("disposition"),
                 cls.priority.label("priority"),
                 cls.detail.label("detail"),
                 HygieneIssueType.detail_redactable.label("detail_redactable"),
                 ProfileResult.pii_flag.label("pii_flag"),
             )
             .join(HygieneIssueType, HygieneIssueType.id == cls.type_id)
+            .join(ProfilingRun, ProfilingRun.id == cls.profile_run_id)
             .outerjoin(
                 ProfileResult,
                 and_(
@@ -272,7 +297,7 @@ def list_for_run(
                     ProfileResult.column_name == cls.column_name,
                 ),
             )
-            .where(cls.profile_run_id == profile_run_id, *clauses)
+            .where(ProfilingRun.job_execution_id == job_execution_id, *clauses)
             .order_by(cls._priority_order(), cls.table_name, cls.column_name, cls.id)
         )
         return cls._paginate(query, page=page, limit=limit, data_class=HygieneIssueListRow)
@@ -301,8 +326,9 @@ def search(
                 cls.schema_name.label("schema_name"),
                 cls.table_name.label("table_name"),
                 cls.column_name.label("column_name"),
+                cls.impact_dimension.label("impact_dimension"),
                 HygieneIssueType.dq_dimension.label("dq_dimension"),
-                func.coalesce(cls.disposition, "Confirmed").label("disposition"),
+                func.coalesce(cls.disposition, Disposition.CONFIRMED).label("disposition"),
                 cls.priority.label("priority"),
                 cls.detail.label("detail"),
                 HygieneIssueType.detail_redactable.label("detail_redactable"),
@@ -347,10 +373,9 @@ def get_with_context(cls, issue_id: UUID, *clauses) -> HygieneIssueDetail | None
                 cls.schema_name.label("schema_name"),
                 cls.table_name.label("table_name"),
                 cls.column_name.label("column_name"),
-                cls.db_data_type.label("db_data_type"),
                 HygieneIssueType.dq_dimension.label("dq_dimension"),
                 cls.impact_dimension.label("impact_dimension"),
-                func.coalesce(cls.disposition, "Confirmed").label("disposition"),
+                func.coalesce(cls.disposition, Disposition.CONFIRMED).label("disposition"),
                 cls.priority.label("priority"),
                 cls.detail.label("detail"),
                 HygieneIssueType.detail_redactable.label("detail_redactable"),

testgen/common/models/profiling_run.py

@@ -162,6 +162,23 @@ def get_latest_run(cls, project_code: str) -> LatestProfilingRun | None:
             return LatestProfilingRun(str(result["id"]), result["run_time"])
         return None
 
+    @classmethod
+    def get_latest_complete_je_id_for_table_group(cls, table_groups_id: UUID) -> UUID | None:
+        """Return the ``job_execution_id`` of the latest completed profiling run for a table group.
+
+        Computed live from ``profiling_runs`` joined to ``job_executions`` — does not read the
+        legacy ``table_groups.last_complete_profile_run_id`` cache, which points at the internal
+        run PK rather than the JE id.
+        """
+        query = (
+            select(cls.job_execution_id)
+            .join(JobExecution, cls.job_execution_id == JobExecution.id)
+            .where(cls.table_groups_id == table_groups_id, JobExecution.status == JobStatus.COMPLETED)
+            .order_by(desc(JobExecution.started_at))
+            .limit(1)
+        )
+        return get_current_session().scalar(query)
+
     @classmethod
     @st.cache_data(show_spinner=False, hash_funcs=ENTITY_HASH_FUNCS)
     def select_minimal_where(

testgen/mcp/prompts/workflows.py

@@ -95,16 +95,15 @@ def hygiene_triage(table_group_id: str | None = None) -> str:
         steps.append(
             "Call `get_data_inventory()` to see projects and table groups, with profiling status per group."
         )
-    steps.append(f"Call `list_profiling_summaries(table_group_id={tg})` to see hygiene issue counts per run.")
     steps.append(f"Call `list_hygiene_issues(table_group_id={tg}, disposition='Confirmed')` for the issues to review.")
     steps.append(
         "For each top issue (ordered by priority), call `get_hygiene_issue(issue_id='...')` for full context — "
         "issue type description, suggested action, column profile."
     )
     steps.append("For unfamiliar issue types, read `testgen://hygiene-issue-types` once for the reference table.")
     steps.append(
-        "For each issue: explain what was found, then ask the user whether to dismiss the issue "
-        "(call `update_hygiene_issue(issue_id='...', disposition='Dismissed')`) or investigate further."
+        "For each issue: explain what was found, then help the user decide a disposition — **Confirmed**, **Dismissed**, or **Muted**. "
+        "Apply via `update_hygiene_issue(issue_id='...', disposition='...')`, or leave it open for further investigation."
     )
     steps.append("Summarize the triage results and any patterns noted across the issues.")
 

testgen/mcp/tools/common.py

@@ -2,20 +2,24 @@
 from uuid import UUID
 
 from testgen.common.date_service import parse_since
-from testgen.common.models.hygiene_issue import HygieneIssueType
+from testgen.common.enums import ImpactDimension, QualityDimension
+from testgen.common.models.hygiene_issue import Disposition, HygieneIssueType, IssueLikelihood, PiiRisk
 from testgen.common.models.table_group import TableGroup
 from testgen.common.models.test_definition import TestType
 from testgen.common.models.test_result import TestResultStatus
 from testgen.common.models.test_suite import TestSuite
 from testgen.mcp.exceptions import MCPResourceNotAccessible, MCPUserError
 from testgen.mcp.permissions import get_project_permissions
 
-VALID_DQ_DIMENSIONS = {"Accuracy", "Completeness", "Consistency", "Recency", "Timeliness", "Uniqueness", "Validity"}
-# DB stores "Inactive"; user-facing label is "Muted".
-_DISPOSITION_USER_TO_DB = {"Confirmed": "Confirmed", "Dismissed": "Dismissed", "Muted": "Inactive"}
-_DISPOSITION_DB_TO_USER = {v: k for k, v in _DISPOSITION_USER_TO_DB.items()}
-_VALID_ISSUE_LIKELIHOODS = {"Definite", "Likely", "Possible"}
-_VALID_PII_RISKS = {"High", "Moderate"}
+# User-facing label for ``Disposition.INACTIVE`` is "Muted" — accept that label on input.
+_DISPOSITION_USER_TO_DB: dict[str, Disposition] = {
+    "Confirmed": Disposition.CONFIRMED,
+    "Dismissed": Disposition.DISMISSED,
+    "Muted": Disposition.INACTIVE,
+}
+_DISPOSITION_DB_TO_USER: dict[Disposition, str] = {v: k for k, v in _DISPOSITION_USER_TO_DB.items()}
+# Filter accepts only the regular likelihoods — PII rows are filtered separately via ``pii_risk``.
+_FILTERABLE_LIKELIHOODS = frozenset({IssueLikelihood.DEFINITE, IssueLikelihood.LIKELY, IssueLikelihood.POSSIBLE})
 
 
 def parse_uuid(value: str, label: str = "ID") -> UUID:
@@ -50,44 +54,74 @@ def parse_since_arg(value: str, label: str = "since", *, today: date | None = No
         raise MCPUserError(f"Invalid `{label}`: {err}") from err
 
 
-def parse_quality_dimension(value: str) -> str:
-    if value not in VALID_DQ_DIMENSIONS:
-        valid = ", ".join(sorted(VALID_DQ_DIMENSIONS))
-        raise MCPUserError(f"Invalid quality_dimension `{value}`. Valid values: {valid}")
-    return value
+def parse_impact_dimension(value: str) -> ImpactDimension:
+    try:
+        return ImpactDimension(value)
+    except ValueError as err:
+        valid = ", ".join(d.value for d in ImpactDimension)
+        raise MCPUserError(f"Invalid impact_dimension `{value}`. Valid values: {valid}") from err
+
 
+def parse_quality_dimension(value: str) -> QualityDimension:
+    try:
+        return QualityDimension(value)
+    except ValueError as err:
+        valid = ", ".join(d.value for d in QualityDimension)
+        raise MCPUserError(f"Invalid quality_dimension `{value}`. Valid values: {valid}") from err
 
-def parse_disposition(value: str) -> str:
-    """Validate a user-facing disposition label and return the DB value.
 
-    Accepts ``Confirmed``, ``Dismissed``, ``Muted`` and returns ``Confirmed``,
-    ``Dismissed``, ``Inactive`` respectively (the DB encodes the legacy ``Inactive``).
+def parse_disposition(value: str) -> Disposition:
+    """Validate a user-facing disposition label and return the stored ``Disposition``.
+
+    Accepts ``Confirmed``, ``Dismissed``, ``Muted`` (user-facing labels). The DB encodes
+    ``INACTIVE`` for "Muted" — see ``Disposition``.
     """
-    if value not in _DISPOSITION_USER_TO_DB:
+    db_value = _DISPOSITION_USER_TO_DB.get(value)
+    if db_value is None:
         valid = ", ".join(sorted(_DISPOSITION_USER_TO_DB))
         raise MCPUserError(f"Invalid disposition `{value}`. Valid values: {valid}")
-    return _DISPOSITION_USER_TO_DB[value]
-
+    return db_value
 
-def format_disposition(value: str) -> str:
-    """Map a DB disposition value to its user-facing label (``Inactive`` → ``Muted``)."""
-    return _DISPOSITION_DB_TO_USER.get(value, value)
 
-
-def parse_issue_likelihood_list(values: list[str]) -> list[str]:
-    invalid = [v for v in values if v not in _VALID_ISSUE_LIKELIHOODS]
+def format_disposition(value: Disposition | str) -> str:
+    """Map a stored disposition to its user-facing label (``INACTIVE`` → "Muted")."""
+    try:
+        return _DISPOSITION_DB_TO_USER[Disposition(value)]
+    except ValueError:
+        return str(value)
+
+
+def parse_issue_likelihood_list(values: list[str]) -> list[IssueLikelihood]:
+    parsed: list[IssueLikelihood] = []
+    invalid: list[str] = []
+    for value in values:
+        try:
+            likelihood = IssueLikelihood(value)
+        except ValueError:
+            invalid.append(value)
+            continue
+        if likelihood not in _FILTERABLE_LIKELIHOODS:
+            invalid.append(value)
+            continue
+        parsed.append(likelihood)
     if invalid:
-        valid = ", ".join(sorted(_VALID_ISSUE_LIKELIHOODS))
+        valid = ", ".join(sorted(v.value for v in _FILTERABLE_LIKELIHOODS))
         raise MCPUserError(f"Invalid issue_likelihood values {invalid}. Valid values: {valid}")
-    return values
+    return parsed
 
 
-def parse_pii_risk_list(values: list[str]) -> list[str]:
-    invalid = [v for v in values if v not in _VALID_PII_RISKS]
+def parse_pii_risk_list(values: list[str]) -> list[PiiRisk]:
+    parsed: list[PiiRisk] = []
+    invalid: list[str] = []
+    for value in values:
+        try:
+            parsed.append(PiiRisk(value))
+        except ValueError:
+            invalid.append(value)
     if invalid:
-        valid = ", ".join(sorted(_VALID_PII_RISKS))
+        valid = ", ".join(r.value for r in PiiRisk)
         raise MCPUserError(f"Invalid pii_risk values {invalid}. Valid values: {valid}")
-    return values
+    return parsed
 
 
 def resolve_test_type(short_name: str) -> str: