feat(ui): add chat-driven dataset review flow

Move dataset review clarification into the assistant workspace and
rework the review page into a chat-centric layout with execution rails.

Add session-scoped assistant actions for mappings, semantic fields,
and SQL preview generation. Introduce optimistic locking for dataset
review mutations, propagate session versions through API responses,
and mask imported filter values before assistant exposure.

Refresh tests, i18n, and spec artifacts to match the new workflow.

BREAKING CHANGE: dataset review mutation endpoints now require the
X-Session-Version header, and clarification is no longer handled
through ClarificationDialog-based flows

backend/src/services/dataset_review/clarification_engine.py

@@ -59,6 +59,8 @@ class ClarificationQuestionPayload:
     priority: int
     state: QuestionState
     options: list[dict[str, object]] = field(default_factory=list)
+
+
 # [/DEF:ClarificationQuestionPayload:Class]
 
 
@@ -71,6 +73,8 @@ class ClarificationStateResult:
     current_question: Optional[ClarificationQuestionPayload]
     session: DatasetReviewSession
     changed_findings: List[ValidationFinding] = field(default_factory=list)
+
+
 # [/DEF:ClarificationStateResult:Class]
 
 
@@ -84,6 +88,8 @@ class ClarificationAnswerCommand:
     answer_kind: AnswerKind
     answer_value: Optional[str]
     user: User
+
+
 # [/DEF:ClarificationAnswerCommand:Class]
 
 
@@ -102,6 +108,7 @@ class ClarificationEngine:
     # @PURPOSE: Bind repository dependency for clarification persistence operations.
     def __init__(self, repository: DatasetReviewSessionRepository) -> None:
         self.repository = repository
+
     # [/DEF:ClarificationEngine_init:Function]
 
     # [DEF:build_question_payload:Function]
@@ -127,10 +134,17 @@ class ClarificationEngine:
                 return None
 
             active_questions = [
-                question for question in clarification_session.questions
+                question
+                for question in clarification_session.questions
                 if question.state == QuestionState.OPEN
             ]
-            active_questions.sort(key=lambda item: (-int(item.priority), item.created_at, item.question_id))
+            active_questions.sort(
+                key=lambda item: (
+                    -int(item.priority),
+                    item.created_at,
+                    item.question_id,
+                )
+            )
 
             if not active_questions:
                 clarification_session.current_question_id = None
@@ -184,7 +198,11 @@ class ClarificationEngine:
                     }
                     for option in sorted(
                         selected_question.options,
-                        key=lambda item: (item.display_order, item.label, item.option_id),
+                        key=lambda item: (
+                            item.display_order,
+                            item.label,
+                            item.option_id,
+                        ),
                     )
                 ],
             )
@@ -197,6 +215,7 @@ class ClarificationEngine:
                 },
             )
             return payload
+
     # [/DEF:build_question_payload:Function]
 
     # [DEF:record_answer:Function]
@@ -208,7 +227,9 @@ class ClarificationEngine:
     # @POST: Answer row is persisted before current-question pointer advances; skipped/expert-review items remain unresolved and visible.
     # @SIDE_EFFECT: Inserts answer row, mutates question/session states, updates clarification findings, and commits.
     # @DATA_CONTRACT: Input[ClarificationAnswerCommand] -> Output[ClarificationStateResult]
-    def record_answer(self, command: ClarificationAnswerCommand) -> ClarificationStateResult:
+    def record_answer(
+        self, command: ClarificationAnswerCommand
+    ) -> ClarificationStateResult:
         with belief_scope("ClarificationEngine.record_answer"):
             session = command.session
             clarification_session = self._get_latest_clarification_session(session)
@@ -223,18 +244,27 @@ class ClarificationEngine:
             if question is None:
                 logger.explore(
                     "Cannot record clarification answer for foreign or missing question",
-                    extra={"session_id": session.session_id, "question_id": command.question_id},
+                    extra={
+                        "session_id": session.session_id,
+                        "question_id": command.question_id,
+                    },
                 )
                 raise ValueError("Clarification question not found")
 
             if question.answer is not None:
                 logger.explore(
                     "Rejected duplicate clarification answer submission",
-                    extra={"session_id": session.session_id, "question_id": command.question_id},
+                    extra={
+                        "session_id": session.session_id,
+                        "question_id": command.question_id,
+                    },
                 )
                 raise ValueError("Clarification question already answered")
 
-            if clarification_session.current_question_id and clarification_session.current_question_id != question.question_id:
+            if (
+                clarification_session.current_question_id
+                and clarification_session.current_question_id != question.question_id
+            ):
                 logger.explore(
                     "Rejected answer for non-active clarification question",
                     extra={
@@ -243,9 +273,13 @@ class ClarificationEngine:
                         "current_question_id": clarification_session.current_question_id,
                     },
                 )
-                raise ValueError("Only the active clarification question can be answered")
+                raise ValueError(
+                    "Only the active clarification question can be answered"
+                )
 
-            normalized_answer_value = self._normalize_answer_value(command.answer_kind, command.answer_value, question)
+            normalized_answer_value = self._normalize_answer_value(
+                command.answer_kind, command.answer_value, question
+            )
 
             logger.reason(
                 "Persisting clarification answer before state advancement",
@@ -260,7 +294,9 @@ class ClarificationEngine:
                 answer_kind=command.answer_kind,
                 answer_value=normalized_answer_value,
                 answered_by_user_id=command.user.id,
-                impact_summary=self._build_impact_summary(question, command.answer_kind, normalized_answer_value),
+                impact_summary=self._build_impact_summary(
+                    question, command.answer_kind, normalized_answer_value
+                ),
             )
             self.repository.db.add(persisted_answer)
             self.repository.db.flush()
@@ -284,15 +320,25 @@ class ClarificationEngine:
             question.updated_at = datetime.utcnow()
             self.repository.db.flush()
 
-            clarification_session.resolved_count = self._count_resolved_questions(clarification_session)
-            clarification_session.remaining_count = self._count_remaining_questions(clarification_session)
-            clarification_session.summary_delta = self.summarize_progress(clarification_session)
+            clarification_session.resolved_count = self._count_resolved_questions(
+                clarification_session
+            )
+            clarification_session.remaining_count = self._count_remaining_questions(
+                clarification_session
+            )
+            clarification_session.summary_delta = self.summarize_progress(
+                clarification_session
+            )
             clarification_session.updated_at = datetime.utcnow()
 
             next_question = self._select_next_open_question(clarification_session)
-            clarification_session.current_question_id = next_question.question_id if next_question else None
+            clarification_session.current_question_id = (
+                next_question.question_id if next_question else None
+            )
             clarification_session.status = (
-                ClarificationStatus.ACTIVE if next_question else ClarificationStatus.COMPLETED
+                ClarificationStatus.ACTIVE
+                if next_question
+                else ClarificationStatus.COMPLETED
             )
             if clarification_session.status == ClarificationStatus.COMPLETED:
                 clarification_session.completed_at = datetime.utcnow()
@@ -304,7 +350,7 @@ class ClarificationEngine:
                 if clarification_session.current_question_id
                 else SessionPhase.REVIEW
             )
-            session.last_activity_at = datetime.utcnow()
+            self.repository.bump_session_version(session)
 
             self.repository.db.commit()
             self.repository.db.refresh(session)
@@ -326,6 +372,7 @@ class ClarificationEngine:
                 session=session,
                 changed_findings=[changed_finding] if changed_finding else [],
             )
+
     # [/DEF:record_answer:Function]
 
     # [DEF:summarize_progress:Function]
@@ -336,6 +383,7 @@ class ClarificationEngine:
         resolved = self._count_resolved_questions(clarification_session)
         remaining = self._count_remaining_questions(clarification_session)
         return f"{resolved} resolved, {remaining} unresolved"
+
     # [/DEF:summarize_progress:Function]
 
     # [DEF:_get_latest_clarification_session:Function]
@@ -353,6 +401,7 @@ class ClarificationEngine:
             reverse=True,
         )
         return ordered_sessions[0]
+
     # [/DEF:_get_latest_clarification_session:Function]
 
     # [DEF:_find_question:Function]
@@ -367,6 +416,7 @@ class ClarificationEngine:
             if question.question_id == question_id:
                 return question
         return None
+
     # [/DEF:_find_question:Function]
 
     # [DEF:_select_next_open_question:Function]
@@ -377,35 +427,46 @@ class ClarificationEngine:
         clarification_session: ClarificationSession,
     ) -> Optional[ClarificationQuestion]:
         open_questions = [
-            question for question in clarification_session.questions
+            question
+            for question in clarification_session.questions
             if question.state == QuestionState.OPEN
         ]
         if not open_questions:
             return None
-        open_questions.sort(key=lambda item: (-int(item.priority), item.created_at, item.question_id))
+        open_questions.sort(
+            key=lambda item: (-int(item.priority), item.created_at, item.question_id)
+        )
         return open_questions[0]
+
     # [/DEF:_select_next_open_question:Function]
 
     # [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:
+    def _count_resolved_questions(
+        self, clarification_session: ClarificationSession
+    ) -> int:
         return sum(
             1
             for question in clarification_session.questions
             if question.state == QuestionState.ANSWERED
         )
+
     # [/DEF:_count_resolved_questions:Function]
 
     # [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:
+    def _count_remaining_questions(
+        self, clarification_session: ClarificationSession
+    ) -> int:
         return sum(
             1
             for question in clarification_session.questions
-            if question.state in {QuestionState.OPEN, QuestionState.SKIPPED, QuestionState.EXPERT_REVIEW}
+            if question.state
+            in {QuestionState.OPEN, QuestionState.SKIPPED, QuestionState.EXPERT_REVIEW}
         )
+
     # [/DEF:_count_remaining_questions:Function]
 
     # [DEF:_normalize_answer_value:Function]
@@ -417,18 +478,28 @@ class ClarificationEngine:
         answer_value: Optional[str],
         question: ClarificationQuestion,
     ) -> Optional[str]:
-        normalized_answer_value = str(answer_value).strip() if answer_value is not None else None
-        if answer_kind in {AnswerKind.SELECTED, AnswerKind.CUSTOM} and not normalized_answer_value:
-            raise ValueError("answer_value is required for selected or custom clarification answers")
+        normalized_answer_value = (
+            str(answer_value).strip() if answer_value is not None else None
+        )
+        if (
+            answer_kind in {AnswerKind.SELECTED, AnswerKind.CUSTOM}
+            and not normalized_answer_value
+        ):
+            raise ValueError(
+                "answer_value is required for selected or custom clarification answers"
+            )
         if answer_kind == AnswerKind.SELECTED:
             allowed_values = {option.value for option in question.options}
             if normalized_answer_value not in allowed_values:
-                raise ValueError("answer_value must match one of the current clarification options")
+                raise ValueError(
+                    "answer_value must match one of the current clarification options"
+                )
         if answer_kind == AnswerKind.SKIPPED:
             return normalized_answer_value or "skipped"
         if answer_kind == AnswerKind.EXPERT_REVIEW:
             return normalized_answer_value or "expert_review"
         return normalized_answer_value
+
     # [/DEF:_normalize_answer_value:Function]
 
     # [DEF:_build_impact_summary:Function]
@@ -445,6 +516,7 @@ 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:_build_impact_summary:Function]
 
     # [DEF:_upsert_clarification_finding:Function]
@@ -461,8 +533,10 @@ class ClarificationEngine:
         caused_by_ref = f"clarification:{question.question_id}"
         existing = next(
             (
-                finding for finding in session.findings
-                if finding.area == FindingArea.CLARIFICATION and finding.caused_by_ref == caused_by_ref
+                finding
+                for finding in session.findings
+                if finding.area == FindingArea.CLARIFICATION
+                and finding.caused_by_ref == caused_by_ref
             ),
             None,
         )
@@ -478,7 +552,9 @@ class ClarificationEngine:
         else:
             resolution_state = ResolutionState.EXPERT_REVIEW
             resolved_at = None
-            message = f"Clarification for '{question.topic_ref}' requires expert review."
+            message = (
+                f"Clarification for '{question.topic_ref}' requires expert review."
+            )
 
         if existing is None:
             existing = ValidationFinding(
@@ -513,6 +589,7 @@ class ClarificationEngine:
             existing.title = "Clarification requires expert review"
 
         return existing
+
     # [/DEF:_upsert_clarification_finding:Function]
 
     # [DEF:_derive_readiness_state:Function]
@@ -532,12 +609,15 @@ class ClarificationEngine:
             return ReadinessState.CLARIFICATION_NEEDED
 
         return ReadinessState.REVIEW_READY
+
     # [/DEF:_derive_readiness_state:Function]
 
     # [DEF:_derive_recommended_action:Function]
     # @COMPLEXITY: 2
     # @PURPOSE: Recompute next-action guidance after clarification mutations.
-    def _derive_recommended_action(self, session: DatasetReviewSession) -> RecommendedAction:
+    def _derive_recommended_action(
+        self, session: DatasetReviewSession
+    ) -> RecommendedAction:
         clarification_session = self._get_latest_clarification_session(session)
         if clarification_session is None:
             return session.recommended_action
@@ -546,7 +626,10 @@ class ClarificationEngine:
         if clarification_session.remaining_count > 0:
             return RecommendedAction.START_CLARIFICATION
         return RecommendedAction.REVIEW_DOCUMENTATION
+
     # [/DEF:_derive_recommended_action:Function]
+
+
 # [/DEF:ClarificationEngine:Class]
 
-# [/DEF:ClarificationEngine:Module]
+# [/DEF:ClarificationEngine:Module]

backend/src/services/dataset_review/orchestrator.py

@@ -109,6 +109,7 @@ class StartSessionResult:
 class PreparePreviewCommand:
     user: User
     session_id: str
+    expected_version: Optional[int] = None
 
 
 # [/DEF:PreparePreviewCommand:Class]
@@ -134,6 +135,7 @@ class PreparePreviewResult:
 class LaunchDatasetCommand:
     user: User
     session_id: str
+    expected_version: Optional[int] = None
 
 
 # [/DEF:LaunchDatasetCommand:Class]
@@ -344,6 +346,7 @@ class DatasetReviewOrchestrator:
             )
             if active_task_id:
                 persisted_session.active_task_id = active_task_id
+                self.repository.bump_session_version(persisted_session)
                 self.repository.db.commit()
                 self.repository.db.refresh(persisted_session)
                 self.repository.event_logger.log_event(
@@ -410,6 +413,11 @@ class DatasetReviewOrchestrator:
                 )
                 raise ValueError("Session not found")
 
+            if command.expected_version is not None:
+                self.repository.require_session_version(
+                    session, command.expected_version
+                )
+
             if session.dataset_id is None:
                 raise ValueError("Preview requires a resolved dataset_id")
 
@@ -521,6 +529,11 @@ class DatasetReviewOrchestrator:
                 )
                 raise ValueError("Session not found")
 
+            if command.expected_version is not None:
+                self.repository.require_session_version(
+                    session, command.expected_version
+                )
+
             if session.dataset_id is None:
                 raise ValueError("Launch requires a resolved dataset_id")
 
@@ -761,6 +774,7 @@ class DatasetReviewOrchestrator:
                 filter_name=str(item.get("filter_name") or f"imported_filter_{index}"),
                 display_name=item.get("display_name"),
                 raw_value=item.get("raw_value"),
+                raw_value_masked=bool(item.get("raw_value_masked", False)),
                 normalized_value=item.get("normalized_value"),
                 source=FilterSource(
                     str(item.get("source") or FilterSource.SUPERSET_URL.value)
@@ -788,9 +802,11 @@ class DatasetReviewOrchestrator:
 
         if session.dataset_id is not None:
             try:
-                dataset_payload = extractor.client.get_dataset_detail(
-                    session_record.dataset_id
-                )
+                dataset_payload = parsed_context.dataset_payload
+                if not isinstance(dataset_payload, dict):
+                    dataset_payload = extractor.client.get_dataset_detail(
+                        session_record.dataset_id
+                    )
                 discovered_variables = extractor.discover_template_variables(
                     dataset_payload
                 )

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

@@ -1,6 +1,7 @@
 import pytest
-from sqlalchemy import create_engine
+from sqlalchemy import create_engine, inspect, text
 from sqlalchemy.orm import sessionmaker
+from src.core.database import _ensure_dataset_review_session_columns
 from src.models.mapping import Base, Environment
 from src.models.auth import User
 from src.models.dataset_review import (
@@ -9,6 +10,7 @@ from src.models.dataset_review import (
     ValidationFinding,
     CompiledPreview,
     DatasetRunContext,
+    ImportedFilter,
     BusinessSummarySource,
     ConfidenceState,
     FindingArea,
@@ -20,6 +22,7 @@ from src.models.dataset_review import (
 )
 from src.services.dataset_review.repositories.session_repository import (
     DatasetReviewSessionRepository,
+    DatasetReviewSessionVersionConflictError,
 )
 
 # [DEF:SessionRepositoryTests:Module]
@@ -77,11 +80,103 @@ def test_create_session(db_session):
         .first()
     )
     assert loaded.user_id == "user1"
+    assert loaded.version == 0
 
 
 # [/DEF:test_create_session:Function]
 
 
+# [DEF:test_require_session_version_conflict:Function]
+# @RELATION: BINDS_TO -> SessionRepositoryTests
+# @PURPOSE: Verify optimistic-lock conflict is raised when caller version is stale.
+def test_require_session_version_conflict(db_session):
+    repo = DatasetReviewSessionRepository(db_session)
+    session = DatasetReviewSession(
+        user_id="user1",
+        environment_id="env1",
+        source_kind="superset_link",
+        source_input="http://link",
+        dataset_ref="dataset1",
+        version=2,
+    )
+    repo.create_session(session)
+
+    with pytest.raises(DatasetReviewSessionVersionConflictError) as exc_info:
+        repo.require_session_version(session, 1)
+
+    assert exc_info.value.expected_version == 1
+    assert exc_info.value.actual_version == 2
+
+
+# [/DEF:test_require_session_version_conflict:Function]
+
+
+# [DEF:test_bump_session_version_updates_last_activity:Function]
+# @RELATION: BINDS_TO -> SessionRepositoryTests
+# @PURPOSE: Verify repository version bump increments monotonically and refreshes last activity.
+def test_bump_session_version_updates_last_activity(db_session):
+    repo = DatasetReviewSessionRepository(db_session)
+    session = DatasetReviewSession(
+        user_id="user1",
+        environment_id="env1",
+        source_kind="superset_link",
+        source_input="http://link",
+        dataset_ref="dataset1",
+    )
+    repo.create_session(session)
+
+    before_activity = session.last_activity_at
+    next_version = repo.bump_session_version(session)
+
+    assert next_version == 1
+    assert session.version == 1
+    assert session.last_activity_at >= before_activity
+
+
+# [/DEF:test_bump_session_version_updates_last_activity:Function]
+
+
+# [DEF:test_save_recovery_state_preserves_raw_value_masked_flag:Function]
+# @RELATION: BINDS_TO -> SessionRepositoryTests
+# @PURPOSE: Verify imported-filter masking metadata persists with recovery bootstrap state.
+def test_save_recovery_state_preserves_raw_value_masked_flag(db_session):
+    repo = DatasetReviewSessionRepository(db_session)
+    session = DatasetReviewSession(
+        user_id="user1",
+        environment_id="env1",
+        source_kind="superset_link",
+        source_input="http://link",
+        dataset_ref="dataset1",
+    )
+    repo.create_session(session)
+
+    imported_filter = ImportedFilter(
+        session_id=session.session_id,
+        filter_name="email",
+        raw_value="***@example.com",
+        raw_value_masked=True,
+        normalized_value=None,
+        source="manual",
+        confidence_state="confirmed",
+        requires_confirmation=False,
+        recovery_status="recovered",
+    )
+
+    updated = repo.save_recovery_state(
+        session.session_id,
+        "user1",
+        [imported_filter],
+        [],
+        [],
+    )
+
+    assert updated.imported_filters[0].raw_value_masked is True
+    assert updated.version == 1
+
+
+# [/DEF:test_save_recovery_state_preserves_raw_value_masked_flag:Function]
+
+
 # [DEF:test_load_session_detail_ownership:Function]
 # @RELATION: BINDS_TO -> SessionRepositoryTests
 def test_load_session_detail_ownership(db_session):
@@ -178,6 +273,48 @@ def test_save_preview_marks_stale(db_session):
 # [/DEF:test_save_preview_marks_stale:Function]
 
 
+# [DEF:test_save_preview_increments_session_version_once_per_call:Function]
+# @RELATION: BINDS_TO -> SessionRepositoryTests
+# @PURPOSE: Verify preview persistence itself contributes exactly one optimistic-lock version increment so higher orchestration layers do not need to bump again for the same preview mutation.
+def test_save_preview_increments_session_version_once_per_call(db_session):
+    repo = DatasetReviewSessionRepository(db_session)
+    session = DatasetReviewSession(
+        user_id="user1",
+        environment_id="env1",
+        source_kind="superset_link",
+        source_input="http://link",
+        dataset_ref="dataset1",
+    )
+    repo.create_session(session)
+
+    repo.save_preview(
+        session.session_id,
+        "user1",
+        CompiledPreview(
+            session_id=session.session_id,
+            preview_status="ready",
+            preview_fingerprint="fp-ready",
+        ),
+    )
+    db_session.refresh(session)
+    assert session.version == 1
+
+    repo.save_preview(
+        session.session_id,
+        "user1",
+        CompiledPreview(
+            session_id=session.session_id,
+            preview_status="pending",
+            preview_fingerprint="fp-pending",
+        ),
+    )
+    db_session.refresh(session)
+    assert session.version == 2
+
+
+# [/DEF:test_save_preview_increments_session_version_once_per_call:Function]
+
+
 # [DEF:test_save_profile_and_findings:Function]
 # @RELATION: BINDS_TO -> SessionRepositoryTests
 def test_save_profile_and_findings(db_session):
@@ -272,6 +409,173 @@ def test_save_run_context(db_session):
 # [/DEF:test_save_run_context:Function]
 
 
+# [DEF:test_ensure_dataset_review_session_columns_adds_missing_legacy_columns:Function]
+# @RELATION: BINDS_TO -> SessionRepositoryTests
+# @PURPOSE: Verify additive dataset review migration creates missing legacy columns for session and imported-filter tables without dropping rows.
+def test_ensure_dataset_review_session_columns_adds_missing_legacy_columns():
+    engine = create_engine("sqlite:///:memory:")
+    with engine.begin() as connection:
+        connection.execute(
+            text(
+                """
+                CREATE TABLE dataset_review_sessions (
+                    session_id VARCHAR PRIMARY KEY,
+                    user_id VARCHAR NOT NULL,
+                    environment_id VARCHAR NOT NULL,
+                    source_kind VARCHAR NOT NULL,
+                    source_input VARCHAR NOT NULL,
+                    dataset_ref VARCHAR NOT NULL,
+                    dataset_id INTEGER,
+                    dashboard_id INTEGER,
+                    readiness_state VARCHAR NOT NULL,
+                    recommended_action VARCHAR NOT NULL,
+                    status VARCHAR NOT NULL,
+                    current_phase VARCHAR NOT NULL,
+                    active_task_id VARCHAR,
+                    last_preview_id VARCHAR,
+                    last_run_context_id VARCHAR,
+                    created_at DATETIME NOT NULL,
+                    updated_at DATETIME NOT NULL,
+                    last_activity_at DATETIME NOT NULL,
+                    closed_at DATETIME
+                )
+                """
+            )
+        )
+        connection.execute(
+            text(
+                """
+                CREATE TABLE imported_filters (
+                    filter_id VARCHAR PRIMARY KEY,
+                    session_id VARCHAR NOT NULL,
+                    filter_name VARCHAR NOT NULL,
+                    display_name VARCHAR,
+                    raw_value JSON NOT NULL,
+                    normalized_value JSON,
+                    source VARCHAR NOT NULL,
+                    confidence_state VARCHAR NOT NULL,
+                    requires_confirmation BOOLEAN NOT NULL DEFAULT FALSE,
+                    recovery_status VARCHAR NOT NULL,
+                    notes TEXT,
+                    created_at DATETIME NOT NULL,
+                    updated_at DATETIME NOT NULL
+                )
+                """
+            )
+        )
+        connection.execute(
+            text(
+                """
+                INSERT INTO dataset_review_sessions (
+                    session_id,
+                    user_id,
+                    environment_id,
+                    source_kind,
+                    source_input,
+                    dataset_ref,
+                    dataset_id,
+                    dashboard_id,
+                    readiness_state,
+                    recommended_action,
+                    status,
+                    current_phase,
+                    active_task_id,
+                    last_preview_id,
+                    last_run_context_id,
+                    created_at,
+                    updated_at,
+                    last_activity_at,
+                    closed_at
+                ) VALUES (
+                    'sess-legacy',
+                    'user1',
+                    'env1',
+                    'superset_link',
+                    'http://link',
+                    'dataset1',
+                    NULL,
+                    NULL,
+                    'EMPTY',
+                    'IMPORT_FROM_SUPERSET',
+                    'ACTIVE',
+                    'INTAKE',
+                    NULL,
+                    NULL,
+                    NULL,
+                    '2026-03-21 00:00:00',
+                    '2026-03-21 00:00:00',
+                    '2026-03-21 00:00:00',
+                    NULL
+                )
+                """
+            )
+        )
+        connection.execute(
+            text(
+                """
+                INSERT INTO imported_filters (
+                    filter_id,
+                    session_id,
+                    filter_name,
+                    display_name,
+                    raw_value,
+                    normalized_value,
+                    source,
+                    confidence_state,
+                    requires_confirmation,
+                    recovery_status,
+                    notes,
+                    created_at,
+                    updated_at
+                ) VALUES (
+                    'filter-legacy',
+                    'sess-legacy',
+                    'country',
+                    'Country',
+                    '"DE"',
+                    NULL,
+                    'MANUAL',
+                    'CONFIRMED',
+                    FALSE,
+                    'RECOVERED',
+                    NULL,
+                    '2026-03-21 00:00:00',
+                    '2026-03-21 00:00:00'
+                )
+                """
+            )
+        )
+
+    _ensure_dataset_review_session_columns(engine)
+
+    inspector = inspect(engine)
+    session_column_names = {
+        column["name"] for column in inspector.get_columns("dataset_review_sessions")
+    }
+    imported_filter_column_names = {
+        column["name"] for column in inspector.get_columns("imported_filters")
+    }
+    assert "version" in session_column_names
+    assert "raw_value_masked" in imported_filter_column_names
+
+    with engine.connect() as connection:
+        version_value = connection.execute(
+            text(
+                "SELECT version FROM dataset_review_sessions WHERE session_id = 'sess-legacy'"
+            )
+        ).scalar_one()
+        raw_value_masked = connection.execute(
+            text(
+                "SELECT raw_value_masked FROM imported_filters WHERE filter_id = 'filter-legacy'"
+            )
+        ).scalar_one()
+    assert version_value == 0
+    assert raw_value_masked in (False, 0)
+
+
+# [/DEF:test_ensure_dataset_review_session_columns_adds_missing_legacy_columns:Function]
+
+
 # [DEF:test_list_sessions_for_user:Function]
 # @RELATION: BINDS_TO -> SessionRepositoryTests
 def test_list_sessions_for_user(db_session):

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

@@ -12,7 +12,8 @@
 # @DATA_CONTRACT: Input[SessionMutation] -> Output[PersistedSessionAggregate]
 # @INVARIANT: answers, mapping approvals, preview artifacts, and launch snapshots are never attributed to the wrong user or session.
 
-from typing import Optional, List
+from datetime import datetime
+from typing import Any, Optional, List, cast
 from sqlalchemy import or_
 from sqlalchemy.orm import Session, joinedload
 from src.models.dataset_review import (
@@ -33,6 +34,24 @@ from src.models.dataset_review import (
 from src.core.logger import belief_scope, logger
 from src.services.dataset_review.event_logger import SessionEventLogger
 
+logger = cast(Any, logger)
+
+
+# [DEF:DatasetReviewSessionVersionConflictError:Class]
+# @COMPLEXITY: 2
+# @PURPOSE: Signal optimistic-lock conflicts for dataset review session mutations.
+class DatasetReviewSessionVersionConflictError(ValueError):
+    def __init__(self, session_id: str, expected_version: int, actual_version: int):
+        self.session_id = session_id
+        self.expected_version = expected_version
+        self.actual_version = actual_version
+        super().__init__(
+            f"Session version conflict: expected {expected_version}, actual {actual_version}"
+        )
+
+
+# [/DEF:DatasetReviewSessionVersionConflictError:Class]
+
 
 # [DEF:DatasetReviewSessionRepository:Class]
 # @COMPLEXITY: 4
@@ -120,6 +139,70 @@ class DatasetReviewSessionRepository:
 
     # [/DEF:create_sess:Function]
 
+    # [DEF:require_session_version:Function]
+    # @COMPLEXITY: 4
+    # @PURPOSE: Enforce optimistic-lock version matching before a session mutation is persisted.
+    # @RELATION: [DEPENDS_ON] -> [DatasetReviewSession]
+    # @PRE: session belongs to the current owner mutation scope and expected_version is the caller's last observed version.
+    # @POST: returns the same session when versions match; otherwise raises deterministic conflict error.
+    # @SIDE_EFFECT: none.
+    # @DATA_CONTRACT: Input[DatasetReviewSession,int] -> Output[DatasetReviewSession|DatasetReviewSessionVersionConflictError]
+    def require_session_version(
+        self, session: DatasetReviewSession, expected_version: int
+    ) -> DatasetReviewSession:
+        with belief_scope("DatasetReviewSessionRepository.require_session_version"):
+            session_record = cast(Any, session)
+            actual_version = int(getattr(session_record, "version", 0) or 0)
+            logger.reason(
+                "Checking optimistic-lock version for dataset review mutation",
+                extra={
+                    "session_id": session.session_id,
+                    "expected_version": expected_version,
+                    "actual_version": actual_version,
+                },
+            )
+            if actual_version != expected_version:
+                logger.explore(
+                    "Rejected dataset review mutation due to stale session version",
+                    extra={
+                        "session_id": session.session_id,
+                        "expected_version": expected_version,
+                        "actual_version": actual_version,
+                    },
+                )
+                raise DatasetReviewSessionVersionConflictError(
+                    str(session_record.session_id), expected_version, actual_version
+                )
+            logger.reflect(
+                "Optimistic-lock version accepted for dataset review mutation",
+                extra={"session_id": session.session_id, "version": actual_version},
+            )
+            return session
+
+    # [/DEF:require_session_version:Function]
+
+    # [DEF:bump_session_version:Function]
+    # @COMPLEXITY: 4
+    # @PURPOSE: Increment optimistic-lock version after a successful session mutation is assembled.
+    # @RELATION: [DEPENDS_ON] -> [DatasetReviewSession]
+    # @PRE: session mutation has passed guards and will be committed in the current transaction.
+    # @POST: session version increments monotonically and last_activity_at reflects the mutation time.
+    # @SIDE_EFFECT: mutates the in-memory session aggregate before commit.
+    # @DATA_CONTRACT: Input[DatasetReviewSession] -> Output[int]
+    def bump_session_version(self, session: DatasetReviewSession) -> int:
+        with belief_scope("DatasetReviewSessionRepository.bump_session_version"):
+            session_record = cast(Any, session)
+            next_version = int(getattr(session_record, "version", 0) or 0) + 1
+            session_record.version = next_version
+            session_record.last_activity_at = datetime.utcnow()
+            logger.reflect(
+                "Prepared incremented dataset review session version",
+                extra={"session_id": session.session_id, "version": next_version},
+            )
+            return next_version
+
+    # [/DEF:bump_session_version:Function]
+
     # [DEF:load_detail:Function]
     # @COMPLEXITY: 4
     # @PURPOSE: Return the full session aggregate for API and frontend resume flows.
@@ -200,6 +283,7 @@ class DatasetReviewSessionRepository:
     ) -> DatasetReviewSession:
         with belief_scope("DatasetReviewSessionRepository.save_profile_and_findings"):
             session = self._get_owned_session(session_id, user_id)
+            session_record = cast(Any, session)
             logger.reason(
                 "Persisting dataset profile and replacing validation findings",
                 extra={
@@ -225,14 +309,17 @@ class DatasetReviewSessionRepository:
             ).delete()
 
             for finding in findings:
-                finding.session_id = session_id
+                finding_record = cast(Any, finding)
+                finding_record.session_id = session_id
                 self.db.add(finding)
 
+            self.bump_session_version(session)
             self.db.commit()
             logger.reflect(
                 "Dataset profile and validation findings committed",
                 extra={
                     "session_id": session.session_id,
+                    "version": session_record.version,
                     "user_id": user_id,
                     "findings_count": len(findings),
                 },
@@ -257,6 +344,7 @@ class DatasetReviewSessionRepository:
     ) -> DatasetReviewSession:
         with belief_scope("DatasetReviewSessionRepository.save_recovery_state"):
             session = self._get_owned_session(session_id, user_id)
+            session_record = cast(Any, session)
             logger.reason(
                 "Persisting dataset review recovery bootstrap state",
                 extra={
@@ -279,24 +367,29 @@ class DatasetReviewSessionRepository:
             ).delete()
 
             for imported_filter in imported_filters:
-                imported_filter.session_id = session_id
+                imported_filter_record = cast(Any, imported_filter)
+                imported_filter_record.session_id = session_id
                 self.db.add(imported_filter)
 
             for template_variable in template_variables:
-                template_variable.session_id = session_id
+                template_variable_record = cast(Any, template_variable)
+                template_variable_record.session_id = session_id
                 self.db.add(template_variable)
 
             self.db.flush()
 
             for execution_mapping in execution_mappings:
-                execution_mapping.session_id = session_id
+                execution_mapping_record = cast(Any, execution_mapping)
+                execution_mapping_record.session_id = session_id
                 self.db.add(execution_mapping)
 
+            self.bump_session_version(session)
             self.db.commit()
             logger.reflect(
                 "Dataset review recovery bootstrap state committed",
                 extra={
                     "session_id": session.session_id,
+                    "version": session_record.version,
                     "user_id": user_id,
                     "imported_filters_count": len(imported_filters),
                     "template_variables_count": len(template_variables),
@@ -321,6 +414,7 @@ class DatasetReviewSessionRepository:
     ) -> CompiledPreview:
         with belief_scope("DatasetReviewSessionRepository.save_preview"):
             session = self._get_owned_session(session_id, user_id)
+            session_record = cast(Any, session)
             logger.reason(
                 "Persisting compiled preview and staling previous preview snapshots",
                 extra={"session_id": session_id, "user_id": user_id},
@@ -332,14 +426,16 @@ class DatasetReviewSessionRepository:
 
             self.db.add(preview)
             self.db.flush()
-            session.last_preview_id = preview.preview_id
+            session_record.last_preview_id = preview.preview_id
 
+            self.bump_session_version(session)
             self.db.commit()
             self.db.refresh(preview)
             logger.reflect(
                 "Compiled preview committed as latest session preview",
                 extra={
                     "session_id": session.session_id,
+                    "version": session_record.version,
                     "preview_id": preview.preview_id,
                     "user_id": user_id,
                 },
@@ -362,6 +458,7 @@ class DatasetReviewSessionRepository:
     ) -> DatasetRunContext:
         with belief_scope("DatasetReviewSessionRepository.save_run_context"):
             session = self._get_owned_session(session_id, user_id)
+            session_record = cast(Any, session)
             logger.reason(
                 "Persisting dataset run context audit snapshot",
                 extra={"session_id": session_id, "user_id": user_id},
@@ -369,14 +466,16 @@ class DatasetReviewSessionRepository:
 
             self.db.add(run_context)
             self.db.flush()
-            session.last_run_context_id = run_context.run_context_id
+            session_record.last_run_context_id = run_context.run_context_id
 
+            self.bump_session_version(session)
             self.db.commit()
             self.db.refresh(run_context)
             logger.reflect(
                 "Dataset run context committed as latest launch snapshot",
                 extra={
                     "session_id": session.session_id,
+                    "version": session_record.version,
                     "run_context_id": run_context.run_context_id,
                     "user_id": user_id,
                 },