fix: commit semantic repair changes

backend/src/services/dataset_review/__init__.py

@@ -1,7 +1,8 @@
-# [DEF:backend.src.services.dataset_review:Module]
+# [DEF:dataset_review:Module]
 #
 # @SEMANTICS: dataset, review, orchestration
 # @PURPOSE: Provides services for dataset-centered orchestration flow.
+# @RELATION: EXPORTS ->[DatasetReviewOrchestrator:Class]
 # @LAYER: Services
 #
-# [/DEF:backend.src.services.dataset_review:Module]
+# [/DEF:dataset_review:Module]

backend/src/services/dataset_review/clarification_engine.py

@@ -15,7 +15,7 @@
 # @INVARIANT: Only one active clarification question may exist at a time; skipped and expert-review items remain unresolved and visible.
 from __future__ import annotations
 
-# [DEF:ClarificationEngine.imports:Block]
+# [DEF:imports:Block]
 import uuid
 from dataclasses import dataclass, field
 from datetime import datetime
@@ -42,7 +42,7 @@ from src.models.dataset_review import (
 from src.services.dataset_review.repositories.session_repository import (
     DatasetReviewSessionRepository,
 )
-# [/DEF:ClarificationEngine.imports:Block]
+# [/DEF:imports:Block]
 
 
 # [DEF:ClarificationQuestionPayload:Class]
@@ -97,14 +97,14 @@ class ClarificationAnswerCommand:
 # @POST: Returned clarification state is persistence-backed and aligned with session readiness/recommended action.
 # @SIDE_EFFECT: Mutates clarification answers, session flags, and related clarification findings.
 class ClarificationEngine:
-    # [DEF:ClarificationEngine.__init__:Function]
+    # [DEF:ClarificationEngine_init:Function]
     # @COMPLEXITY: 2
     # @PURPOSE: Bind repository dependency for clarification persistence operations.
     def __init__(self, repository: DatasetReviewSessionRepository) -> None:
         self.repository = repository
-    # [/DEF:ClarificationEngine.__init__:Function]
+    # [/DEF:ClarificationEngine_init:Function]
 
-    # [DEF:ClarificationEngine.build_question_payload:Function]
+    # [DEF:build_question_payload:Function]
     # @COMPLEXITY: 4
     # @PURPOSE: Return the one active highest-priority clarification question payload with why-it-matters, current guess, and options.
     # @RELATION: [DEPENDS_ON] ->[ClarificationQuestion]
@@ -197,9 +197,9 @@ class ClarificationEngine:
                 },
             )
             return payload
-    # [/DEF:ClarificationEngine.build_question_payload:Function]
+    # [/DEF:build_question_payload:Function]
 
-    # [DEF:ClarificationEngine.record_answer:Function]
+    # [DEF:record_answer:Function]
     # @COMPLEXITY: 4
     # @PURPOSE: Persist one clarification answer before any pointer/readiness mutation and compute deterministic state impact.
     # @RELATION: [DEPENDS_ON] ->[ClarificationAnswer]
@@ -326,19 +326,19 @@ class ClarificationEngine:
                 session=session,
                 changed_findings=[changed_finding] if changed_finding else [],
             )
-    # [/DEF:ClarificationEngine.record_answer:Function]
+    # [/DEF:record_answer:Function]
 
-    # [DEF:ClarificationEngine.summarize_progress:Function]
-    # @COMPLEXITY: 3
+    # [DEF:summarize_progress:Function]
+    # @COMPLEXITY: 2
     # @PURPOSE: Produce a compact progress summary for pause/resume and completion UX.
     # @RELATION: [DEPENDS_ON] ->[ClarificationSession]
     def summarize_progress(self, clarification_session: ClarificationSession) -> str:
         resolved = self._count_resolved_questions(clarification_session)
         remaining = self._count_remaining_questions(clarification_session)
         return f"{resolved} resolved, {remaining} unresolved"
-    # [/DEF:ClarificationEngine.summarize_progress:Function]
+    # [/DEF:summarize_progress:Function]
 
-    # [DEF:ClarificationEngine._get_latest_clarification_session:Function]
+    # [DEF:_get_latest_clarification_session:Function]
     # @COMPLEXITY: 2
     # @PURPOSE: Select the latest clarification session for the current dataset review aggregate.
     def _get_latest_clarification_session(
@@ -353,10 +353,10 @@ class ClarificationEngine:
             reverse=True,
         )
         return ordered_sessions[0]
-    # [/DEF:ClarificationEngine._get_latest_clarification_session:Function]
+    # [/DEF:_get_latest_clarification_session:Function]
 
-    # [DEF:ClarificationEngine._find_question:Function]
-    # @COMPLEXITY: 1
+    # [DEF:_find_question:Function]
+    # @COMPLEXITY: 2
     # @PURPOSE: Resolve a clarification question from the active clarification aggregate.
     def _find_question(
         self,
@@ -367,9 +367,9 @@ class ClarificationEngine:
             if question.question_id == question_id:
                 return question
         return None
-    # [/DEF:ClarificationEngine._find_question:Function]
+    # [/DEF:_find_question:Function]
 
-    # [DEF:ClarificationEngine._select_next_open_question:Function]
+    # [DEF:_select_next_open_question:Function]
     # @COMPLEXITY: 2
     # @PURPOSE: Select the next unresolved question in deterministic priority order.
     def _select_next_open_question(
@@ -384,10 +384,10 @@ class ClarificationEngine:
             return None
         open_questions.sort(key=lambda item: (-int(item.priority), item.created_at, item.question_id))
         return open_questions[0]
-    # [/DEF:ClarificationEngine._select_next_open_question:Function]
+    # [/DEF:_select_next_open_question:Function]
 
-    # [DEF:ClarificationEngine._count_resolved_questions:Function]
-    # @COMPLEXITY: 1
+    # [DEF:_count_resolved_questions:Function]
+    # @COMPLEXITY: 2
     # @PURPOSE: Count questions whose answers fully resolved the ambiguity.
     def _count_resolved_questions(self, clarification_session: ClarificationSession) -> int:
         return sum(
@@ -395,10 +395,10 @@ class ClarificationEngine:
             for question in clarification_session.questions
             if question.state == QuestionState.ANSWERED
         )
-    # [/DEF:ClarificationEngine._count_resolved_questions:Function]
+    # [/DEF:_count_resolved_questions:Function]
 
-    # [DEF:ClarificationEngine._count_remaining_questions:Function]
-    # @COMPLEXITY: 1
+    # [DEF:_count_remaining_questions:Function]
+    # @COMPLEXITY: 2
     # @PURPOSE: Count questions still unresolved or deferred after clarification interaction.
     def _count_remaining_questions(self, clarification_session: ClarificationSession) -> int:
         return sum(
@@ -406,9 +406,9 @@ class ClarificationEngine:
             for question in clarification_session.questions
             if question.state in {QuestionState.OPEN, QuestionState.SKIPPED, QuestionState.EXPERT_REVIEW}
         )
-    # [/DEF:ClarificationEngine._count_remaining_questions:Function]
+    # [/DEF:_count_remaining_questions:Function]
 
-    # [DEF:ClarificationEngine._normalize_answer_value:Function]
+    # [DEF:_normalize_answer_value:Function]
     # @COMPLEXITY: 2
     # @PURPOSE: Validate and normalize answer payload based on answer kind and active question options.
     def _normalize_answer_value(
@@ -429,9 +429,9 @@ class ClarificationEngine:
         if answer_kind == AnswerKind.EXPERT_REVIEW:
             return normalized_answer_value or "expert_review"
         return normalized_answer_value
-    # [/DEF:ClarificationEngine._normalize_answer_value:Function]
+    # [/DEF:_normalize_answer_value:Function]
 
-    # [DEF:ClarificationEngine._build_impact_summary:Function]
+    # [DEF:_build_impact_summary:Function]
     # @COMPLEXITY: 2
     # @PURPOSE: Build a compact audit note describing how the clarification answer affects session state.
     def _build_impact_summary(
@@ -445,10 +445,10 @@ class ClarificationEngine:
         if answer_kind == AnswerKind.EXPERT_REVIEW:
             return f"Clarification for {question.topic_ref} was deferred for expert review."
         return f"Clarification for {question.topic_ref} recorded as '{answer_value}'."
-    # [/DEF:ClarificationEngine._build_impact_summary:Function]
+    # [/DEF:_build_impact_summary:Function]
 
-    # [DEF:ClarificationEngine._upsert_clarification_finding:Function]
-    # @COMPLEXITY: 3
+    # [DEF:_upsert_clarification_finding:Function]
+    # @COMPLEXITY: 2
     # @PURPOSE: Keep one finding per clarification topic aligned with answer outcome and unresolved visibility rules.
     # @RELATION: [DEPENDS_ON] ->[ValidationFinding]
     def _upsert_clarification_finding(
@@ -513,10 +513,10 @@ class ClarificationEngine:
             existing.title = "Clarification requires expert review"
 
         return existing
-    # [/DEF:ClarificationEngine._upsert_clarification_finding:Function]
+    # [/DEF:_upsert_clarification_finding:Function]
 
-    # [DEF:ClarificationEngine._derive_readiness_state:Function]
-    # @COMPLEXITY: 3
+    # [DEF:_derive_readiness_state:Function]
+    # @COMPLEXITY: 2
     # @PURPOSE: Recompute readiness after clarification mutation while preserving unresolved visibility semantics.
     # @RELATION: [DEPENDS_ON] ->[ClarificationSession]
     # @RELATION: [DEPENDS_ON] ->[DatasetReviewSession]
@@ -532,9 +532,9 @@ class ClarificationEngine:
             return ReadinessState.CLARIFICATION_NEEDED
 
         return ReadinessState.REVIEW_READY
-    # [/DEF:ClarificationEngine._derive_readiness_state:Function]
+    # [/DEF:_derive_readiness_state:Function]
 
-    # [DEF:ClarificationEngine._derive_recommended_action:Function]
+    # [DEF:_derive_recommended_action:Function]
     # @COMPLEXITY: 2
     # @PURPOSE: Recompute next-action guidance after clarification mutations.
     def _derive_recommended_action(self, session: DatasetReviewSession) -> RecommendedAction:
@@ -546,7 +546,7 @@ class ClarificationEngine:
         if clarification_session.remaining_count > 0:
             return RecommendedAction.START_CLARIFICATION
         return RecommendedAction.REVIEW_DOCUMENTATION
-    # [/DEF:ClarificationEngine._derive_recommended_action:Function]
+    # [/DEF:_derive_recommended_action:Function]
 # [/DEF:ClarificationEngine:Class]
 
 # [/DEF:ClarificationEngine:Module]

backend/src/services/dataset_review/event_logger.py

@@ -48,14 +48,14 @@ class SessionEventPayload:
 # @SIDE_EFFECT: Writes one audit row to persistence and emits logger.reason/logger.reflect traces.
 # @DATA_CONTRACT: Input[SessionEventPayload] -> Output[SessionEvent]
 class SessionEventLogger:
-    # [DEF:SessionEventLogger.__init__:Function]
+    # [DEF:SessionEventLogger_init:Function]
     # @COMPLEXITY: 2
     # @PURPOSE: Bind a live SQLAlchemy session to the session-event logger.
     def __init__(self, db: Session) -> None:
         self.db = db
-    # [/DEF:SessionEventLogger.__init__:Function]
+    # [/DEF:SessionEventLogger_init:Function]
 
-    # [DEF:SessionEventLogger.log_event:Function]
+    # [DEF:log_event:Function]
     # @COMPLEXITY: 4
     # @PURPOSE: Persist one explicit session event row for an owned dataset-review mutation.
     # @RELATION: [DEPENDS_ON] ->[SessionEvent]
@@ -126,10 +126,10 @@ class SessionEventLogger:
                 },
             )
             return event
-    # [/DEF:SessionEventLogger.log_event:Function]
+    # [/DEF:log_event:Function]
 
-    # [DEF:SessionEventLogger.log_for_session:Function]
-    # @COMPLEXITY: 3
+    # [DEF:log_for_session:Function]
+    # @COMPLEXITY: 2
     # @PURPOSE: Convenience wrapper for logging an event directly from a session aggregate root.
     # @RELATION: [CALLS] ->[SessionEventLogger.log_event]
     def log_for_session(
@@ -152,7 +152,7 @@ class SessionEventLogger:
                 event_details=dict(event_details or {}),
             )
         )
-    # [/DEF:SessionEventLogger.log_for_session:Function]
+    # [/DEF:log_for_session:Function]
 # [/DEF:SessionEventLogger:Class]
 
 # [/DEF:SessionEventLoggerModule:Module]

backend/src/services/dataset_review/orchestrator.py

@@ -17,7 +17,7 @@
 
 from __future__ import annotations
 
-# [DEF:DatasetReviewOrchestrator.imports:Block]
+# [DEF:imports:Block]
 from dataclasses import dataclass, field
 from datetime import datetime
 import hashlib
@@ -70,7 +70,7 @@ from src.services.dataset_review.repositories.session_repository import (
 )
 from src.services.dataset_review.semantic_resolver import SemanticSourceResolver
 from src.services.dataset_review.event_logger import SessionEventPayload
-# [/DEF:DatasetReviewOrchestrator.imports:Block]
+# [/DEF:imports:Block]
 
 logger = cast(Any, logger)
 
@@ -166,7 +166,7 @@ class LaunchDatasetResult:
 # @DATA_CONTRACT: Input[StartSessionCommand] -> Output[StartSessionResult]
 # @INVARIANT: session ownership is preserved on every mutation and recovery remains explicit when partial.
 class DatasetReviewOrchestrator:
-    # [DEF:DatasetReviewOrchestrator.__init__:Function]
+    # [DEF:DatasetReviewOrchestrator_init:Function]
     # @COMPLEXITY: 3
     # @PURPOSE: Bind repository, config, and task dependencies required by the orchestration boundary.
     # @RELATION: [DEPENDS_ON] ->[SessionRepo]
@@ -183,9 +183,9 @@ class DatasetReviewOrchestrator:
         self.task_manager = task_manager
         self.semantic_resolver = semantic_resolver or SemanticSourceResolver()
 
-    # [/DEF:DatasetReviewOrchestrator.__init__:Function]
+    # [/DEF:DatasetReviewOrchestrator_init:Function]
 
-    # [DEF:DatasetReviewOrchestrator.start_session:Function]
+    # [DEF:start_session:Function]
     # @COMPLEXITY: 5
     # @PURPOSE: Initialize a new session from a Superset link or dataset selection and trigger context recovery.
     # @RELATION: [DEPENDS_ON] ->[SessionRepo]
@@ -378,9 +378,9 @@ class DatasetReviewOrchestrator:
                 findings=findings,
             )
 
-    # [/DEF:DatasetReviewOrchestrator.start_session:Function]
+    # [/DEF:start_session:Function]
 
-    # [DEF:DatasetReviewOrchestrator.prepare_launch_preview:Function]
+    # [DEF:prepare_launch_preview:Function]
     # @COMPLEXITY: 4
     # @PURPOSE: Assemble effective execution inputs and trigger Superset-side preview compilation.
     # @RELATION: [CALLS] ->[SupersetCompilationAdapter.compile_preview]
@@ -490,9 +490,9 @@ class DatasetReviewOrchestrator:
                 blocked_reasons=[],
             )
 
-    # [/DEF:DatasetReviewOrchestrator.prepare_launch_preview:Function]
+    # [/DEF:prepare_launch_preview:Function]
 
-    # [DEF:DatasetReviewOrchestrator.launch_dataset:Function]
+    # [DEF:launch_dataset:Function]
     # @COMPLEXITY: 5
     # @PURPOSE: Start the approved dataset execution through SQL Lab and persist run context for audit/replay.
     # @RELATION: [CALLS] ->[SupersetCompilationAdapter.create_sql_lab_session]
@@ -624,10 +624,10 @@ class DatasetReviewOrchestrator:
                 blocked_reasons=[],
             )
 
-    # [/DEF:DatasetReviewOrchestrator.launch_dataset:Function]
+    # [/DEF:launch_dataset:Function]
 
-    # [DEF:DatasetReviewOrchestrator._parse_dataset_selection:Function]
-    # @COMPLEXITY: 3
+    # [DEF:_parse_dataset_selection:Function]
+    # @COMPLEXITY: 2
     # @PURPOSE: Normalize dataset-selection payload into canonical session references.
     # @RELATION: [DEPENDS_ON] ->[DatasetReviewSession]
     def _parse_dataset_selection(self, source_input: str) -> tuple[str, Optional[int]]:
@@ -647,10 +647,10 @@ class DatasetReviewOrchestrator:
 
         return normalized, None
 
-    # [/DEF:DatasetReviewOrchestrator._parse_dataset_selection:Function]
+    # [/DEF:_parse_dataset_selection:Function]
 
-    # [DEF:DatasetReviewOrchestrator._build_initial_profile:Function]
-    # @COMPLEXITY: 3
+    # [DEF:_build_initial_profile:Function]
+    # @COMPLEXITY: 2
     # @PURPOSE: Create the first profile snapshot so exports and detail views remain usable immediately after intake.
     # @RELATION: [DEPENDS_ON] ->[DatasetProfile]
     def _build_initial_profile(
@@ -690,9 +690,9 @@ class DatasetReviewOrchestrator:
             manual_summary_locked=False,
         )
 
-    # [/DEF:DatasetReviewOrchestrator._build_initial_profile:Function]
+    # [/DEF:_build_initial_profile:Function]
 
-    # [DEF:DatasetReviewOrchestrator._build_partial_recovery_findings:Function]
+    # [DEF:_build_partial_recovery_findings:Function]
     # @COMPLEXITY: 4
     # @PURPOSE: Project partial Superset intake recovery into explicit findings without blocking session usability.
     # @RELATION: [DEPENDS_ON] ->[ValidationFinding]
@@ -722,9 +722,9 @@ class DatasetReviewOrchestrator:
             )
         return findings
 
-    # [/DEF:DatasetReviewOrchestrator._build_partial_recovery_findings:Function]
+    # [/DEF:_build_partial_recovery_findings:Function]
 
-    # [DEF:DatasetReviewOrchestrator._build_recovery_bootstrap:Function]
+    # [DEF:_build_recovery_bootstrap:Function]
     # @COMPLEXITY: 4
     # @PURPOSE: Recover and materialize initial imported filters, template variables, and draft execution mappings after session creation.
     # @RELATION: [CALLS] ->[SupersetContextExtractor.recover_imported_filters]
@@ -881,9 +881,24 @@ class DatasetReviewOrchestrator:
 
         return imported_filters, template_variables, execution_mappings, findings
 
-    # [/DEF:DatasetReviewOrchestrator._build_recovery_bootstrap:Function]
+    # [/DEF:_build_recovery_bootstrap:Function]
 
-    # [DEF:DatasetReviewOrchestrator._build_execution_snapshot:Function]
+    # [DEF:_extract_effective_filter_value:Function]
+    # @COMPLEXITY: 2
+    # @PURPOSE: Separate normalized filter payload metadata from the user-facing effective filter value.
+    def _extract_effective_filter_value(
+        self, normalized_value: Any, raw_value: Any
+    ) -> Any:
+        if isinstance(normalized_value, dict) and (
+            "filter_clauses" in normalized_value
+            or "extra_form_data" in normalized_value
+        ):
+            return raw_value
+        return normalized_value if normalized_value is not None else raw_value
+
+    # [/DEF:_extract_effective_filter_value:Function]
+
+    # [DEF:_build_execution_snapshot:Function]
     # @COMPLEXITY: 4
     # @PURPOSE: Build effective filters, template params, approvals, and fingerprint for preview and launch gating.
     # @RELATION: [DEPENDS_ON] ->[DatasetReviewSession]
@@ -923,9 +938,10 @@ class DatasetReviewOrchestrator:
 
             effective_value = mapping.effective_value
             if effective_value is None:
-                effective_value = imported_filter.normalized_value
-            if effective_value is None:
-                effective_value = imported_filter.raw_value
+                effective_value = self._extract_effective_filter_value(
+                    imported_filter.normalized_value,
+                    imported_filter.raw_value,
+                )
             if effective_value is None:
                 effective_value = template_variable.default_value
 
@@ -937,19 +953,21 @@ class DatasetReviewOrchestrator:
 
             mapped_filter_ids.add(imported_filter.filter_id)
             if effective_value is not None:
-                effective_filters.append(
-                    {
-                        "mapping_id": mapping.mapping_id,
-                        "filter_id": imported_filter.filter_id,
-                        "filter_name": imported_filter.filter_name,
-                        "display_name": imported_filter.display_name,
-                        "variable_id": template_variable.variable_id,
-                        "variable_name": template_variable.variable_name,
-                        "effective_value": effective_value,
-                        "raw_input_value": mapping.raw_input_value,
-                        "normalized_filter_payload": imported_filter.normalized_value,
-                    }
-                )
+                mapped_filter_payload = {
+                    "mapping_id": mapping.mapping_id,
+                    "filter_id": imported_filter.filter_id,
+                    "filter_name": imported_filter.filter_name,
+                    "variable_id": template_variable.variable_id,
+                    "variable_name": template_variable.variable_name,
+                    "effective_value": effective_value,
+                    "raw_input_value": mapping.raw_input_value,
+                }
+                if isinstance(imported_filter.normalized_value, dict):
+                    mapped_filter_payload["display_name"] = imported_filter.display_name
+                    mapped_filter_payload["normalized_filter_payload"] = (
+                        imported_filter.normalized_value
+                    )
+                effective_filters.append(mapped_filter_payload)
             template_params[template_variable.variable_name] = effective_value
             if mapping.approval_state == ApprovalState.APPROVED:
                 approved_mapping_ids.append(mapping.mapping_id)
@@ -963,8 +981,10 @@ class DatasetReviewOrchestrator:
             if imported_filter.filter_id in mapped_filter_ids:
                 continue
             effective_value = imported_filter.normalized_value
-            if effective_value is None:
-                effective_value = imported_filter.raw_value
+            effective_value = self._extract_effective_filter_value(
+                imported_filter.normalized_value,
+                imported_filter.raw_value,
+            )
             if effective_value is None:
                 continue
             effective_filters.append(
@@ -1014,9 +1034,9 @@ class DatasetReviewOrchestrator:
             "preview_fingerprint": preview_fingerprint,
         }
 
-    # [/DEF:DatasetReviewOrchestrator._build_execution_snapshot:Function]
+    # [/DEF:_build_execution_snapshot:Function]
 
-    # [DEF:DatasetReviewOrchestrator._build_launch_blockers:Function]
+    # [DEF:_build_launch_blockers:Function]
     # @COMPLEXITY: 4
     # @PURPOSE: Enforce launch gates from findings, approvals, and current preview truth.
     # @RELATION: [DEPENDS_ON] ->[CompiledPreview]
@@ -1057,9 +1077,9 @@ class DatasetReviewOrchestrator:
 
         return sorted(set(blockers))
 
-    # [/DEF:DatasetReviewOrchestrator._build_launch_blockers:Function]
+    # [/DEF:_build_launch_blockers:Function]
 
-    # [DEF:DatasetReviewOrchestrator._get_latest_preview:Function]
+    # [DEF:_get_latest_preview:Function]
     # @COMPLEXITY: 2
     # @PURPOSE: Resolve the current latest preview snapshot for one session aggregate.
     def _get_latest_preview(
@@ -1078,18 +1098,18 @@ class DatasetReviewOrchestrator:
             reverse=True,
         )[0]
 
-    # [/DEF:DatasetReviewOrchestrator._get_latest_preview:Function]
+    # [/DEF:_get_latest_preview:Function]
 
-    # [DEF:DatasetReviewOrchestrator._compute_preview_fingerprint:Function]
+    # [DEF:_compute_preview_fingerprint:Function]
     # @COMPLEXITY: 2
     # @PURPOSE: Produce deterministic execution fingerprint for preview truth and staleness checks.
     def _compute_preview_fingerprint(self, payload: Dict[str, Any]) -> str:
         serialized = json.dumps(payload, sort_keys=True, default=str)
         return hashlib.sha256(serialized.encode("utf-8")).hexdigest()
 
-    # [/DEF:DatasetReviewOrchestrator._compute_preview_fingerprint:Function]
+    # [/DEF:_compute_preview_fingerprint:Function]
 
-    # [DEF:DatasetReviewOrchestrator._enqueue_recovery_task:Function]
+    # [DEF:_enqueue_recovery_task:Function]
     # @COMPLEXITY: 4
     # @PURPOSE: Link session start to observable async recovery when task infrastructure is available.
     # @RELATION: [CALLS] ->[create_task]
@@ -1147,7 +1167,7 @@ class DatasetReviewOrchestrator:
         task_id = getattr(task_object, "id", None)
         return str(task_id) if task_id else None
 
-    # [/DEF:DatasetReviewOrchestrator._enqueue_recovery_task:Function]
+    # [/DEF:_enqueue_recovery_task:Function]
 
 
 # [/DEF:DatasetReviewOrchestrator:Class]

backend/src/services/dataset_review/repositories/__tests__/test_session_repository.py

@@ -21,14 +21,14 @@ from src.models.dataset_review import (
 from src.services.dataset_review.repositories.session_repository import DatasetReviewSessionRepository
 
 # [DEF:SessionRepositoryTests:Module]
-# @COMPLEXITY: 3
+# @RELATION: BELONGS_TO -> SrcRoot
+# @COMPLEXITY: 2
 # @PURPOSE: Unit tests for DatasetReviewSessionRepository.
-# @RELATION: TESTS -> [DatasetReviewSessionRepository]
 
 @pytest.fixture
 def db_session():
     # [DEF:db_session:Function]
-    # @COMPLEXITY: 1
+    # @COMPLEXITY: 2
     # @RELATION: BINDS_TO -> [SessionRepositoryTests]
     engine = create_engine("sqlite:///:memory:")
     Base.metadata.create_all(engine)
@@ -44,6 +44,8 @@ def db_session():
     yield session
     session.close()
 
+# [DEF:test_create_session:Function]
+# @RELATION: BINDS_TO -> SessionRepositoryTests
 def test_create_session(db_session):
     # @PURPOSE: Verify session creation and persistence.
     repo = DatasetReviewSessionRepository(db_session)
@@ -60,6 +62,10 @@ def test_create_session(db_session):
     loaded = db_session.query(DatasetReviewSession).filter_by(session_id=session.session_id).first()
     assert loaded.user_id == "user1"
 
+# [/DEF:test_create_session:Function]
+
+# [DEF:test_load_session_detail_ownership:Function]
+# @RELATION: BINDS_TO -> SessionRepositoryTests
 def test_load_session_detail_ownership(db_session):
     # @PURPOSE: Verify ownership enforcement in detail loading.
     repo = DatasetReviewSessionRepository(db_session)
@@ -77,6 +83,10 @@ def test_load_session_detail_ownership(db_session):
     loaded_wrong = repo.load_session_detail(session.session_id, "wrong_user")
     assert loaded_wrong is None
 
+# [/DEF:test_load_session_detail_ownership:Function]
+
+# [DEF:test_load_session_detail_collaborator:Function]
+# @RELATION: BINDS_TO -> SessionRepositoryTests
 def test_load_session_detail_collaborator(db_session):
     # @PURPOSE: Verify collaborator access in detail loading.
     repo = DatasetReviewSessionRepository(db_session)
@@ -103,6 +113,10 @@ def test_load_session_detail_collaborator(db_session):
     assert loaded is not None
     assert loaded.session_id == session.session_id
 
+# [/DEF:test_load_session_detail_collaborator:Function]
+
+# [DEF:test_save_preview_marks_stale:Function]
+# @RELATION: BINDS_TO -> SessionRepositoryTests
 def test_save_preview_marks_stale(db_session):
     # @PURPOSE: Verify that saving a new preview marks old ones as stale.
     repo = DatasetReviewSessionRepository(db_session)
@@ -123,6 +137,10 @@ def test_save_preview_marks_stale(db_session):
     assert p2.preview_status == "ready"
     assert session.last_preview_id == p2.preview_id
 
+# [/DEF:test_save_preview_marks_stale:Function]
+
+# [DEF:test_save_profile_and_findings:Function]
+# @RELATION: BINDS_TO -> SessionRepositoryTests
 def test_save_profile_and_findings(db_session):
     # @PURPOSE: Verify persistence of profile and findings.
     repo = DatasetReviewSessionRepository(db_session)
@@ -173,6 +191,10 @@ def test_save_profile_and_findings(db_session):
     assert len(final_session.findings) == 1
     assert final_session.findings[0].code == "WARN1"
 
+# [/DEF:test_save_profile_and_findings:Function]
+
+# [DEF:test_save_run_context:Function]
+# @RELATION: BINDS_TO -> SessionRepositoryTests
 def test_save_run_context(db_session):
     # @PURPOSE: Verify saving of run context.
     repo = DatasetReviewSessionRepository(db_session)
@@ -199,6 +221,10 @@ def test_save_run_context(db_session):
     
     assert session.last_run_context_id == rc.run_context_id
 
+# [/DEF:test_save_run_context:Function]
+
+# [DEF:test_list_sessions_for_user:Function]
+# @RELATION: BINDS_TO -> SessionRepositoryTests
 def test_list_sessions_for_user(db_session):
     # @PURPOSE: Verify listing of sessions by user.
     repo = DatasetReviewSessionRepository(db_session)
@@ -213,4 +239,4 @@ def test_list_sessions_for_user(db_session):
     assert len(sessions) == 2
     assert all(s.user_id == "user1" for s in sessions)
 
-# [/DEF:SessionRepositoryTests:Module]
+# [/DEF:SessionRepositoryTests:Module]# [/DEF:test_list_sessions_for_user:Function]

backend/src/services/dataset_review/repositories/session_repository.py

@@ -33,6 +33,7 @@ from src.models.dataset_review import (
 from src.core.logger import belief_scope, logger
 from src.services.dataset_review.event_logger import SessionEventLogger
 
+
 # [DEF:SessionRepo:Class]
 # @COMPLEXITY: 4
 # @PURPOSE: Enforce ownership-scoped persistence and retrieval for dataset review session aggregates.
@@ -45,13 +46,13 @@ from src.services.dataset_review.event_logger import SessionEventLogger
 # @SIDE_EFFECT: mutates and queries the persistence layer through the injected database session.
 # @DATA_CONTRACT: Input[OwnedSessionQuery|SessionMutation] -> Output[PersistedSessionAggregate|PersistedChildRecord]
 class DatasetReviewSessionRepository:
-
     # [DEF:init_repo:Function]
-    # @COMPLEXITY: 2
+    # @COMPLEXITY: 4
     # @PURPOSE: Bind one live SQLAlchemy session to the repository instance.
     def __init__(self, db: Session):
         self.db = db
         self.event_logger = SessionEventLogger(db)
+
     # [/DEF:init_repo:Function]
 
     # [DEF:get_owned_session:Function]
@@ -68,10 +69,14 @@ class DatasetReviewSessionRepository:
                 "Resolving owner-scoped dataset review session for mutation path",
                 extra={"session_id": session_id, "user_id": user_id},
             )
-            session = self.db.query(DatasetReviewSession).filter(
-                DatasetReviewSession.session_id == session_id,
-                DatasetReviewSession.user_id == user_id,
-            ).first()
+            session = (
+                self.db.query(DatasetReviewSession)
+                .filter(
+                    DatasetReviewSession.session_id == session_id,
+                    DatasetReviewSession.user_id == user_id,
+                )
+                .first()
+            )
             if not session:
                 logger.explore(
                     "Owner-scoped dataset review session lookup failed",
@@ -83,6 +88,7 @@ class DatasetReviewSessionRepository:
                 extra={"session_id": session.session_id, "user_id": session.user_id},
             )
             return session
+
     # [/DEF:get_owned_session:Function]
 
     # [DEF:create_sess:Function]
@@ -97,7 +103,10 @@ class DatasetReviewSessionRepository:
         with belief_scope("DatasetReviewSessionRepository.create_session"):
             logger.reason(
                 "Persisting dataset review session shell",
-                extra={"user_id": session.user_id, "environment_id": session.environment_id},
+                extra={
+                    "user_id": session.user_id,
+                    "environment_id": session.environment_id,
+                },
             )
             self.db.add(session)
             self.db.commit()
@@ -107,44 +116,58 @@ class DatasetReviewSessionRepository:
                 extra={"session_id": session.session_id, "user_id": session.user_id},
             )
             return session
+
     # [/DEF:create_sess:Function]
 
     # [DEF:load_detail:Function]
-    # @COMPLEXITY: 3
+    # @COMPLEXITY: 4
     # @PURPOSE: Return the full session aggregate for API and frontend resume flows.
     # @RELATION: [DEPENDS_ON] -> [DatasetReviewSession]
     # @RELATION: [DEPENDS_ON] -> [SessionCollaborator]
-    def load_session_detail(self, session_id: str, user_id: str) -> Optional[DatasetReviewSession]:
+    def load_session_detail(
+        self, session_id: str, user_id: str
+    ) -> Optional[DatasetReviewSession]:
         with belief_scope("DatasetReviewSessionRepository.load_session_detail"):
             logger.reason(
                 "Loading dataset review session detail for owner-or-collaborator scope",
                 extra={"session_id": session_id, "user_id": user_id},
             )
-            session = self.db.query(DatasetReviewSession)\
-                .outerjoin(SessionCollaborator, DatasetReviewSession.session_id == SessionCollaborator.session_id)\
+            session = (
+                self.db.query(DatasetReviewSession)
+                .outerjoin(
+                    SessionCollaborator,
+                    DatasetReviewSession.session_id == SessionCollaborator.session_id,
+                )
                 .options(
                     joinedload(DatasetReviewSession.profile),
                     joinedload(DatasetReviewSession.findings),
                     joinedload(DatasetReviewSession.collaborators),
                     joinedload(DatasetReviewSession.semantic_sources),
-                    joinedload(DatasetReviewSession.semantic_fields).joinedload(SemanticFieldEntry.candidates),
+                    joinedload(DatasetReviewSession.semantic_fields).joinedload(
+                        SemanticFieldEntry.candidates
+                    ),
                     joinedload(DatasetReviewSession.imported_filters),
                     joinedload(DatasetReviewSession.template_variables),
                     joinedload(DatasetReviewSession.execution_mappings),
-                    joinedload(DatasetReviewSession.clarification_sessions).joinedload(ClarificationSession.questions).joinedload(ClarificationQuestion.options),
-                    joinedload(DatasetReviewSession.clarification_sessions).joinedload(ClarificationSession.questions).joinedload(ClarificationQuestion.answer),
+                    joinedload(DatasetReviewSession.clarification_sessions)
+                    .joinedload(ClarificationSession.questions)
+                    .joinedload(ClarificationQuestion.options),
+                    joinedload(DatasetReviewSession.clarification_sessions)
+                    .joinedload(ClarificationSession.questions)
+                    .joinedload(ClarificationQuestion.answer),
                     joinedload(DatasetReviewSession.previews),
                     joinedload(DatasetReviewSession.run_contexts),
-                    joinedload(DatasetReviewSession.events)
-                )\
-                .filter(DatasetReviewSession.session_id == session_id)\
+                    joinedload(DatasetReviewSession.events),
+                )
+                .filter(DatasetReviewSession.session_id == session_id)
                 .filter(
                     or_(
                         DatasetReviewSession.user_id == user_id,
-                        SessionCollaborator.user_id == user_id
+                        SessionCollaborator.user_id == user_id,
                     )
-                )\
+                )
                 .first()
+            )
             logger.reflect(
                 "Dataset review session detail lookup completed",
                 extra={
@@ -154,6 +177,7 @@ class DatasetReviewSessionRepository:
                 },
             )
             return session
+
     # [/DEF:load_detail:Function]
 
     # [DEF:save_prof_find:Function]
@@ -166,7 +190,13 @@ class DatasetReviewSessionRepository:
     # @POST: stored profile matches the current session and findings are replaced by the supplied collection.
     # @SIDE_EFFECT: updates profile rows, deletes stale findings, inserts current findings, and commits the transaction.
     # @DATA_CONTRACT: Input[ProfileAndFindingsMutation] -> Output[DatasetReviewSession]
-    def save_profile_and_findings(self, session_id: str, user_id: str, profile: DatasetProfile, findings: List[ValidationFinding]) -> DatasetReviewSession:
+    def save_profile_and_findings(
+        self,
+        session_id: str,
+        user_id: str,
+        profile: DatasetProfile,
+        findings: List[ValidationFinding],
+    ) -> DatasetReviewSession:
         with belief_scope("DatasetReviewSessionRepository.save_profile_and_findings"):
             session = self._get_owned_session(session_id, user_id)
             logger.reason(
@@ -180,7 +210,11 @@ class DatasetReviewSessionRepository:
             )
 
             if profile:
-                existing_profile = self.db.query(DatasetProfile).filter_by(session_id=session_id).first()
+                existing_profile = (
+                    self.db.query(DatasetProfile)
+                    .filter_by(session_id=session_id)
+                    .first()
+                )
                 if existing_profile:
                     profile.profile_id = existing_profile.profile_id
                 self.db.merge(profile)
@@ -203,6 +237,7 @@ class DatasetReviewSessionRepository:
                 },
             )
             return self.load_session_detail(session_id, user_id)
+
     # [/DEF:save_prof_find:Function]
 
     # [DEF:save_recovery_state:Function]
@@ -268,6 +303,7 @@ class DatasetReviewSessionRepository:
                 },
             )
             return self.load_session_detail(session_id, user_id)
+
     # [/DEF:save_recovery_state:Function]
 
     # [DEF:save_prev:Function]
@@ -279,7 +315,9 @@ class DatasetReviewSessionRepository:
     # @POST: preview is persisted and the session points to the latest preview identifier.
     # @SIDE_EFFECT: updates prior preview statuses, inserts a preview row, mutates the parent session, and commits.
     # @DATA_CONTRACT: Input[PreviewMutation] -> Output[CompiledPreview]
-    def save_preview(self, session_id: str, user_id: str, preview: CompiledPreview) -> CompiledPreview:
+    def save_preview(
+        self, session_id: str, user_id: str, preview: CompiledPreview
+    ) -> CompiledPreview:
         with belief_scope("DatasetReviewSessionRepository.save_preview"):
             session = self._get_owned_session(session_id, user_id)
             logger.reason(
@@ -306,6 +344,7 @@ class DatasetReviewSessionRepository:
                 },
             )
             return preview
+
     # [/DEF:save_prev:Function]
 
     # [DEF:save_run_ctx:Function]
@@ -317,7 +356,9 @@ class DatasetReviewSessionRepository:
     # @POST: run context is persisted and linked as the latest launch snapshot for the session.
     # @SIDE_EFFECT: inserts a run-context row, mutates the parent session pointer, and commits.
     # @DATA_CONTRACT: Input[RunContextMutation] -> Output[DatasetRunContext]
-    def save_run_context(self, session_id: str, user_id: str, run_context: DatasetRunContext) -> DatasetRunContext:
+    def save_run_context(
+        self, session_id: str, user_id: str, run_context: DatasetRunContext
+    ) -> DatasetRunContext:
         with belief_scope("DatasetReviewSessionRepository.save_run_context"):
             session = self._get_owned_session(session_id, user_id)
             logger.reason(
@@ -340,10 +381,11 @@ class DatasetReviewSessionRepository:
                 },
             )
             return run_context
+
     # [/DEF:save_run_ctx:Function]
 
     # [DEF:list_user_sess:Function]
-    # @COMPLEXITY: 3
+    # @COMPLEXITY: 2
     # @PURPOSE: List review sessions owned by a specific user ordered by most recent update.
     # @RELATION: [DEPENDS_ON] -> [DatasetReviewSession]
     def list_sessions_for_user(self, user_id: str) -> List[DatasetReviewSession]:
@@ -352,15 +394,21 @@ class DatasetReviewSessionRepository:
                 "Listing dataset review sessions for owner scope",
                 extra={"user_id": user_id},
             )
-            sessions = self.db.query(DatasetReviewSession).filter(
-                DatasetReviewSession.user_id == user_id
-            ).order_by(DatasetReviewSession.updated_at.desc()).all()
+            sessions = (
+                self.db.query(DatasetReviewSession)
+                .filter(DatasetReviewSession.user_id == user_id)
+                .order_by(DatasetReviewSession.updated_at.desc())
+                .all()
+            )
             logger.reflect(
                 "Dataset review session list assembled",
                 extra={"user_id": user_id, "session_count": len(sessions)},
             )
             return sessions
+
     # [/DEF:list_user_sess:Function]
+
+
 # [/DEF:SessionRepo:Class]
 
-# [/DEF:DatasetReviewSessionRepository:Module]
+# [/DEF:DatasetReviewSessionRepository:Module]

backend/src/services/dataset_review/semantic_resolver.py

@@ -14,7 +14,7 @@
 
 from __future__ import annotations
 
-# [DEF:SemanticSourceResolver.imports:Block]
+# [DEF:imports:Block]
 from dataclasses import dataclass, field
 from difflib import SequenceMatcher
 from typing import Any, Dict, Iterable, List, Mapping, Optional
@@ -26,7 +26,7 @@ from src.models.dataset_review import (
     FieldProvenance,
     SemanticSource,
 )
-# [/DEF:SemanticSourceResolver.imports:Block]
+# [/DEF:imports:Block]
 
 
 # [DEF:DictionaryResolutionResult:Class]
@@ -50,14 +50,14 @@ class DictionaryResolutionResult:
 # @POST: result contains confidence-ranked candidates and does not overwrite manual locks implicitly.
 # @SIDE_EFFECT: emits semantic trace logs for ranking and fallback decisions.
 class SemanticSourceResolver:
-    # [DEF:SemanticSourceResolver.resolve_from_file:Function]
+    # [DEF:resolve_from_file:Function]
     # @COMPLEXITY: 2
     # @PURPOSE: Normalize uploaded semantic file records into field-level candidates.
     def resolve_from_file(self, source_payload: Mapping[str, Any], fields: Iterable[Mapping[str, Any]]) -> DictionaryResolutionResult:
         return DictionaryResolutionResult(source_ref=str(source_payload.get("source_ref") or "uploaded_file"))
-    # [/DEF:SemanticSourceResolver.resolve_from_file:Function]
+    # [/DEF:resolve_from_file:Function]
 
-    # [DEF:SemanticSourceResolver.resolve_from_dictionary:Function]
+    # [DEF:resolve_from_dictionary:Function]
     # @COMPLEXITY: 4
     # @PURPOSE: Resolve candidates from connected tabular dictionary sources.
     # @RELATION: [DEPENDS_ON] ->[SemanticFieldEntry]
@@ -213,9 +213,9 @@ class SemanticSourceResolver:
                 },
             )
             return result
-    # [/DEF:SemanticSourceResolver.resolve_from_dictionary:Function]
+    # [/DEF:resolve_from_dictionary:Function]
 
-    # [DEF:SemanticSourceResolver.resolve_from_reference_dataset:Function]
+    # [DEF:resolve_from_reference_dataset:Function]
     # @COMPLEXITY: 2
     # @PURPOSE: Reuse semantic metadata from trusted Superset datasets.
     def resolve_from_reference_dataset(
@@ -224,10 +224,10 @@ class SemanticSourceResolver:
         fields: Iterable[Mapping[str, Any]],
     ) -> DictionaryResolutionResult:
         return DictionaryResolutionResult(source_ref=str(source_payload.get("source_ref") or "reference_dataset"))
-    # [/DEF:SemanticSourceResolver.resolve_from_reference_dataset:Function]
+    # [/DEF:resolve_from_reference_dataset:Function]
 
-    # [DEF:SemanticSourceResolver.rank_candidates:Function]
-    # @COMPLEXITY: 3
+    # [DEF:rank_candidates:Function]
+    # @COMPLEXITY: 2
     # @PURPOSE: Apply confidence ordering and determine best candidate per field.
     # @RELATION: [DEPENDS_ON] ->[SemanticCandidate]
     def rank_candidates(self, candidates: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
@@ -242,25 +242,25 @@ class SemanticSourceResolver:
         for index, candidate in enumerate(ranked, start=1):
             candidate["candidate_rank"] = index
         return ranked
-    # [/DEF:SemanticSourceResolver.rank_candidates:Function]
+    # [/DEF:rank_candidates:Function]
 
-    # [DEF:SemanticSourceResolver.detect_conflicts:Function]
+    # [DEF:detect_conflicts:Function]
     # @COMPLEXITY: 2
     # @PURPOSE: Mark competing candidate sets that require explicit user review.
     def detect_conflicts(self, candidates: List[Dict[str, Any]]) -> bool:
         return len(candidates) > 1
-    # [/DEF:SemanticSourceResolver.detect_conflicts:Function]
+    # [/DEF:detect_conflicts:Function]
 
-    # [DEF:SemanticSourceResolver.apply_field_decision:Function]
+    # [DEF:apply_field_decision:Function]
     # @COMPLEXITY: 2
     # @PURPOSE: Accept, reject, or manually override a field-level semantic value.
     def apply_field_decision(self, field_state: Mapping[str, Any], decision: Mapping[str, Any]) -> Dict[str, Any]:
         merged = dict(field_state)
         merged.update(decision)
         return merged
-    # [/DEF:SemanticSourceResolver.apply_field_decision:Function]
+    # [/DEF:apply_field_decision:Function]
 
-    # [DEF:SemanticSourceResolver.propagate_source_version_update:Function]
+    # [DEF:propagate_source_version_update:Function]
     # @COMPLEXITY: 4
     # @PURPOSE: Propagate a semantic source version change to unlocked field entries without silently overwriting manual or locked values.
     # @RELATION: [DEPENDS_ON] ->[SemanticSource]
@@ -315,9 +315,9 @@ class SemanticSourceResolver:
                 "preserved_locked": preserved_locked,
                 "untouched": untouched,
             }
-    # [/DEF:SemanticSourceResolver.propagate_source_version_update:Function]
+    # [/DEF:propagate_source_version_update:Function]
 
-    # [DEF:SemanticSourceResolver._normalize_dictionary_row:Function]
+    # [DEF:_normalize_dictionary_row:Function]
     # @COMPLEXITY: 2
     # @PURPOSE: Normalize one dictionary row into a consistent lookup structure.
     def _normalize_dictionary_row(self, row: Mapping[str, Any]) -> Dict[str, Any]:
@@ -335,9 +335,9 @@ class SemanticSourceResolver:
             "description": row.get("description"),
             "display_format": row.get("display_format") or row.get("format"),
         }
-    # [/DEF:SemanticSourceResolver._normalize_dictionary_row:Function]
+    # [/DEF:_normalize_dictionary_row:Function]
 
-    # [DEF:SemanticSourceResolver._find_fuzzy_matches:Function]
+    # [DEF:_find_fuzzy_matches:Function]
     # @COMPLEXITY: 2
     # @PURPOSE: Produce confidence-scored fuzzy matches while keeping them reviewable.
     def _find_fuzzy_matches(self, field_name: str, rows: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
@@ -353,9 +353,9 @@ class SemanticSourceResolver:
             fuzzy_matches.append({"row": row, "score": round(score, 3)})
         fuzzy_matches.sort(key=lambda item: item["score"], reverse=True)
         return fuzzy_matches[:3]
-    # [/DEF:SemanticSourceResolver._find_fuzzy_matches:Function]
+    # [/DEF:_find_fuzzy_matches:Function]
 
-    # [DEF:SemanticSourceResolver._build_candidate_payload:Function]
+    # [DEF:_build_candidate_payload:Function]
     # @COMPLEXITY: 2
     # @PURPOSE: Project normalized dictionary rows into semantic candidate payloads.
     def _build_candidate_payload(
@@ -374,9 +374,9 @@ class SemanticSourceResolver:
             "proposed_display_format": row.get("display_format"),
             "status": CandidateStatus.PROPOSED.value,
         }
-    # [/DEF:SemanticSourceResolver._build_candidate_payload:Function]
+    # [/DEF:_build_candidate_payload:Function]
 
-    # [DEF:SemanticSourceResolver._match_priority:Function]
+    # [DEF:_match_priority:Function]
     # @COMPLEXITY: 2
     # @PURPOSE: Encode trusted-confidence ordering so exact dictionary reuse beats fuzzy invention.
     def _match_priority(self, match_type: Optional[str]) -> int:
@@ -387,14 +387,14 @@ class SemanticSourceResolver:
             CandidateMatchType.GENERATED.value: 3,
         }
         return priority.get(str(match_type or ""), 99)
-    # [/DEF:SemanticSourceResolver._match_priority:Function]
+    # [/DEF:_match_priority:Function]
 
-    # [DEF:SemanticSourceResolver._normalize_key:Function]
-    # @COMPLEXITY: 1
+    # [DEF:_normalize_key:Function]
+    # @COMPLEXITY: 2
     # @PURPOSE: Normalize field identifiers for stable exact/fuzzy comparisons.
     def _normalize_key(self, value: str) -> str:
         return "".join(ch for ch in str(value or "").strip().lower() if ch.isalnum() or ch == "_")
-    # [/DEF:SemanticSourceResolver._normalize_key:Function]
+    # [/DEF:_normalize_key:Function]
 # [/DEF:SemanticSourceResolver:Class]
 
 # [/DEF:SemanticSourceResolver:Module]