diff --git a/backend/tests/api/test_assistant_admin_routes.py b/backend/tests/api/test_assistant_admin_routes.py
new file mode 100644
index 00000000..95e5c1b8
--- /dev/null
+++ b/backend/tests/api/test_assistant_admin_routes.py
@@ -0,0 +1,230 @@
+# #region Test.Assistant.AdminRoutes [C:2] [TYPE Module] [SEMANTICS test,assistant,admin,route,audit,conversation]
+# @BRIEF Tests for _admin_routes.py — admin route handlers for assistant.
+# @RELATION BINDS_TO -> [AssistantAdminRoutes]
+# @TEST_EDGE: list_conversations -> Returns empty (deprecated)
+# @TEST_EDGE: delete_conversation -> Deletes from cache and DB
+# @TEST_EDGE: delete_conversation_not_found -> 404 on missing
+# @TEST_EDGE: get_history -> Returns paginated messages
+# @TEST_EDGE: get_assistant_audit -> Returns audit records
+
+from pathlib import Path
+import sys
+
+sys.path.insert(0, str(Path(__file__).parent.parent / "src"))
+
+from datetime import UTC, datetime
+from unittest.mock import MagicMock, patch
+import pytest
+from fastapi import HTTPException
+
+
+@pytest.fixture(autouse=True)
+def _clean_global_state():
+    """Clean global in-memory stores before each test."""
+    from src.api.routes.assistant._schemas import ASSISTANT_AUDIT, CONVERSATIONS, USER_ACTIVE_CONVERSATION
+    ASSISTANT_AUDIT.clear()
+    CONVERSATIONS.clear()
+    USER_ACTIVE_CONVERSATION.clear()
+
+
+class TestListConversations:
+    """list_conversations — deprecated endpoint."""
+
+    def test_returns_empty(self):
+        from src.api.routes.assistant._admin_routes import list_conversations
+
+        result = __import__("asyncio").run(list_conversations(
+            page=1, page_size=20, include_archived=False, archived_only=False,
+            search=None, current_user=None, db=MagicMock(),
+        ))
+        assert result["items"] == []
+        assert result["total"] == 0
+        assert result["page"] == 1
+
+
+class TestDeleteConversation:
+    """delete_conversation — conversation deletion."""
+
+    def test_delete_success(self):
+        from src.api.routes.assistant._schemas import CONVERSATIONS
+        from src.api.routes.assistant._admin_routes import delete_conversation
+        from src.schemas.auth import User
+
+        user = User(id="user-1", username="test", email="test@x.com", is_active=True,
+                    auth_source="internal", created_at=datetime.now(UTC), last_login=datetime.now(UTC), roles=[])
+        db = MagicMock()
+
+        # Add to in-memory cache
+        CONVERSATIONS[("user-1", "conv-1")] = [{"role": "user", "text": "hello"}]
+
+        # Build query chain where .delete() returns int
+        q = MagicMock()
+        q.filter.return_value = q
+        q.delete.return_value = 5
+        db.query.return_value = q
+
+        result = __import__("asyncio").run(delete_conversation(
+            "conv-1", current_user=user, db=db,
+        ))
+        assert result["status"] == "success"
+        assert result["deleted"] == 5
+        assert ("user-1", "conv-1") not in CONVERSATIONS
+
+    def test_delete_not_found_raises_404(self):
+        from src.api.routes.assistant._admin_routes import delete_conversation
+        from src.schemas.auth import User
+
+        user = User(id="user-1", username="test", email="test@x.com", is_active=True,
+                    auth_source="internal", created_at=datetime.now(UTC), last_login=datetime.now(UTC), roles=[])
+        db = MagicMock()
+
+        q = MagicMock()
+        q.filter.return_value = q
+        q.delete.return_value = 0
+        db.query.return_value = q
+
+        with pytest.raises(HTTPException) as exc:
+            __import__("asyncio").run(delete_conversation(
+                "conv-nonexistent", current_user=user, db=db,
+            ))
+        assert exc.value.status_code == 404
+
+    def test_delete_unknown_user_no_cache_entry(self):
+        from src.api.routes.assistant._admin_routes import delete_conversation
+        from src.schemas.auth import User
+
+        user = User(id="user-other", username="test", email="test@x.com", is_active=True,
+                    auth_source="internal", created_at=datetime.now(UTC), last_login=datetime.now(UTC), roles=[])
+        db = MagicMock()
+
+        q = MagicMock()
+        q.filter.return_value = q
+        q.delete.return_value = 0
+        db.query.return_value = q
+
+        with pytest.raises(HTTPException) as exc:
+            __import__("asyncio").run(delete_conversation(
+                "conv-1", current_user=user, db=db,
+            ))
+        assert exc.value.status_code == 404
+
+
+class TestGetHistory:
+    """get_history — conversation history retrieval."""
+
+    def _make_mock_row(self, msg_id="msg-1", conv_id="conv-1", role="user", text="Hello",
+                       state="sent", task_id=None, confirmation_id=None, payload=None):
+        from src.models.assistant import AssistantMessageRecord
+        row = MagicMock(spec=AssistantMessageRecord)
+        row.id = msg_id
+        row.user_id = "user-1"
+        row.conversation_id = conv_id
+        row.role = role
+        row.text = text
+        row.state = state
+        row.task_id = task_id
+        row.confirmation_id = confirmation_id
+        row.payload = payload
+        row.created_at = datetime.now(UTC)
+        return row
+
+    def test_get_history_basic(self):
+        from src.api.routes.assistant._admin_routes import get_history
+        from src.schemas.auth import User
+
+        user = User(id="user-1", username="test", email="test@x.com", is_active=True,
+                    auth_source="internal", created_at=datetime.now(UTC), last_login=datetime.now(UTC), roles=[])
+        db = MagicMock()
+
+        # Build query mock chain that returns real values
+        q = MagicMock()
+        # .filter().filter() -> q
+        q.filter.return_value = q
+        q.count.return_value = 1
+        q.order_by.return_value = q
+        q.offset.return_value = q
+        q.limit.return_value = q
+        q.all.return_value = [self._make_mock_row()]
+        db.query.return_value = q
+
+        with patch("src.api.routes.assistant._admin_routes._resolve_or_create_conversation") as mock_resolve, \
+             patch("src.api.routes.assistant._admin_routes._cleanup_history_ttl") as mock_cleanup:
+            mock_resolve.return_value = "conv-1"
+
+            result = __import__("asyncio").run(get_history(
+                page=1, page_size=20, conversation_id="conv-1",
+                from_latest=False, current_user=user, db=db,
+            ))
+            assert result["total"] == 1
+            assert len(result["items"]) == 1
+            assert result["items"][0]["message_id"] == "msg-1"
+            assert result["conversation_id"] == "conv-1"
+
+    def test_get_history_from_latest(self):
+        from src.api.routes.assistant._admin_routes import get_history
+        from src.schemas.auth import User
+
+        user = User(id="user-1", username="test", email="test@x.com", is_active=True,
+                    auth_source="internal", created_at=datetime.now(UTC), last_login=datetime.now(UTC), roles=[])
+        db = MagicMock()
+
+        q = MagicMock()
+        q.filter.return_value = q
+        q.count.return_value = 1
+        q.order_by.return_value = q
+        q.offset.return_value = q
+        q.limit.return_value = q
+        q.all.return_value = [self._make_mock_row(role="assistant", text="Response", state="success", task_id="task-1")]
+        db.query.return_value = q
+
+        with patch("src.api.routes.assistant._admin_routes._resolve_or_create_conversation") as mock_resolve, \
+             patch("src.api.routes.assistant._admin_routes._cleanup_history_ttl") as mock_cleanup:
+            mock_resolve.return_value = "conv-1"
+
+            result = __import__("asyncio").run(get_history(
+                page=1, page_size=20, conversation_id="conv-1",
+                from_latest=True, current_user=user, db=db,
+            ))
+            assert result["total"] == 1
+            assert result["from_latest"] is True
+
+
+class TestGetAssistantAudit:
+    """get_assistant_audit — audit record retrieval."""
+
+    def _make_mock_row(self):
+        from src.models.assistant import AssistantAuditRecord
+        row = MagicMock(spec=AssistantAuditRecord)
+        row.id = "audit-1"
+        row.user_id = "user-1"
+        row.conversation_id = "conv-1"
+        row.decision = "executed"
+        row.task_id = "task-1"
+        row.message = "test command"
+        row.payload = {"key": "val"}
+        row.created_at = datetime.now(UTC)
+        return row
+
+    def test_get_audit(self):
+        from src.api.routes.assistant._admin_routes import get_assistant_audit
+        from src.schemas.auth import User
+
+        user = User(id="user-1", username="test", email="test@x.com", is_active=True,
+                    auth_source="internal", created_at=datetime.now(UTC), last_login=datetime.now(UTC), roles=[])
+        db = MagicMock()
+
+        q = MagicMock()
+        q.filter.return_value = q
+        q.order_by.return_value = q
+        q.limit.return_value = q
+        q.all.return_value = [self._make_mock_row()]
+        db.query.return_value = q
+
+        result = __import__("asyncio").run(get_assistant_audit(
+            limit=50, current_user=user, db=db, _=None,
+        ))
+        assert result["total"] == 1
+        assert len(result["items"]) == 1
+        assert result["items"][0]["id"] == "audit-1"
+        assert result["memory_total"] == 0
+# #endregion Test.Assistant.AdminRoutes
diff --git a/backend/tests/api/test_assistant_command_parser.py b/backend/tests/api/test_assistant_command_parser.py
new file mode 100644
index 00000000..016bc34b
--- /dev/null
+++ b/backend/tests/api/test_assistant_command_parser.py
@@ -0,0 +1,367 @@
+# #region Test.Assistant.CommandParser [C:2] [TYPE Module] [SEMANTICS test,assistant,command,parser,nlu]
+# @BRIEF Tests for _command_parser.py — deterministic RU/EN command parser.
+# @RELATION BINDS_TO -> [AssistantCommandParser]
+# @TEST_EDGE: help_variants -> All help phrases map to show_capabilities
+# @TEST_EDGE: status_variants -> Status commands get_task_status with/without task_id
+# @TEST_EDGE: branch_create -> Branch create with dashboard_id
+# @TEST_EDGE: commit_variants -> Commit with quoted message
+# @TEST_EDGE: deploy_variants -> Deploy with env, production detection
+# @TEST_EDGE: migrate_variants -> Migration with src/tgt/dry_run/replace_db
+# @TEST_EDGE: backup_variants -> Backup with env
+# @TEST_EDGE: health_variants -> Health check with env
+# @TEST_EDGE: validation_variants -> LLM validation with provider
+# @TEST_EDGE: documentation_variants -> LLM documentation with dataset_id
+# @TEST_EDGE: fallback -> Unknown commands return clarify intent
+
+from pathlib import Path
+import sys
+
+sys.path.insert(0, str(Path(__file__).parent.parent / "src"))
+
+from unittest.mock import MagicMock
+import pytest
+
+
+def _make_config_manager(envs=None):
+    cm = MagicMock()
+    envs = envs or []
+    cm.get_environments = MagicMock(return_value=envs)
+    return cm
+
+
+def _make_env(id_, name_, stage="DEV", is_production=False, is_default=False):
+    env = MagicMock()
+    env.id = id_
+    env.name = name_
+    env.stage = stage
+    env.is_production = is_production
+    env.is_default = is_default
+    return env
+
+
+class TestParseCommandHelp:
+    """_parse_command — help/capabilities variants."""
+
+    @pytest.mark.parametrize("msg", [
+        "что ты умеешь",
+        "что умеешь",
+        "что ты можешь",
+        "help",
+        "помощь",
+        "доступные команды",
+        "какие команды",
+    ])
+    def test_help_variants(self, msg):
+        from src.api.routes.assistant._command_parser import _parse_command
+        cm = _make_config_manager()
+        result = _parse_command(msg, cm)
+        assert result["domain"] == "assistant"
+        assert result["operation"] == "show_capabilities"
+        assert result["entities"] == {}
+        assert result["confidence"] == 0.98
+        assert result["risk_level"] == "safe"
+        assert result["requires_confirmation"] is False
+
+
+class TestParseCommandStatus:
+    """_parse_command — status/task status variants."""
+
+    @pytest.mark.parametrize("msg,expected_conf", [
+        ("статус", 0.66),
+        ("status", 0.66),
+        ("state", 0.66),
+        ("проверь задачу", 0.66),
+        ("статус task-abc123", 0.92),
+        ("status task-abc-def", 0.92),
+    ])
+    def test_status_variants(self, msg, expected_conf):
+        from src.api.routes.assistant._command_parser import _parse_command
+        cm = _make_config_manager()
+        result = _parse_command(msg, cm)
+        assert result["domain"] == "status"
+        assert result["operation"] == "get_task_status"
+        assert result["confidence"] == expected_conf
+        assert result["risk_level"] == "safe"
+
+
+class TestParseCommandBranch:
+    """_parse_command — branch create variants."""
+
+    @pytest.mark.parametrize("msg,expected_branch,expected_conf", [
+        ("создай ветку feature/test", "feature/test", 0.7),
+        ("create branch feature/abc", "feature/abc", 0.7),
+        ("сделай ветку fix/123 для дашборд 42", "fix/123", 0.95),
+        ("create branch main", "main", 0.7),
+    ])
+    def test_branch_create(self, msg, expected_branch, expected_conf):
+        from src.api.routes.assistant._command_parser import _parse_command
+        cm = _make_config_manager()
+        result = _parse_command(msg, cm)
+        assert result["domain"] == "git"
+        assert result["operation"] == "create_branch"
+        assert result["risk_level"] == "guarded"
+        assert result["confidence"] == expected_conf
+
+
+class TestParseCommandCommit:
+    """_parse_command — commit variants."""
+
+    def test_commit_with_message(self):
+        from src.api.routes.assistant._command_parser import _parse_command
+        cm = _make_config_manager()
+        result = _parse_command('commit "update dashboard filters"', cm)
+        assert result["operation"] == "commit_changes"
+        assert result["entities"]["message"] == "update dashboard filters"
+
+    def test_commit_without_quotes(self):
+        from src.api.routes.assistant._command_parser import _parse_command
+        cm = _make_config_manager()
+        result = _parse_command("commit", cm)
+        assert result["operation"] == "commit_changes"
+        assert result["entities"]["message"] == "assistant: update dashboard changes"
+
+
+class TestParseCommandDeploy:
+    """_parse_command — deploy variants with production detection."""
+
+    def test_deploy_to_dev(self):
+        from src.api.routes.assistant._command_parser import _parse_command
+        envs = [_make_env("dev", "DEV", stage="DEV")]
+        cm = _make_config_manager(envs)
+        result = _parse_command("деплой дашборд 42 в dev", cm)
+        assert result["operation"] == "deploy_dashboard"
+        assert result["risk_level"] == "guarded"
+        assert result["requires_confirmation"] is False
+
+    def test_deploy_to_production(self):
+        from src.api.routes.assistant._command_parser import _parse_command
+        prod = _make_env("prod", "PROD", stage="PROD", is_production=True)
+        cm = _make_config_manager([prod])
+        result = _parse_command("deploy dashboard 42 в prod", cm)
+        assert result["operation"] == "deploy_dashboard"
+        assert result["risk_level"] == "dangerous"
+        assert result["requires_confirmation"] is True
+
+    def test_deploy_without_env(self):
+        from src.api.routes.assistant._command_parser import _parse_command
+        cm = _make_config_manager()
+        result = _parse_command("разверн дашборд 42", cm)
+        assert result["operation"] == "deploy_dashboard"
+        assert result["confidence"] == 0.7
+
+
+class TestParseCommandMigration:
+    """_parse_command — migration variants."""
+
+    def test_migration_with_src_tgt(self):
+        from src.api.routes.assistant._command_parser import _parse_command
+        envs = [_make_env("prod", "PROD", stage="PROD", is_production=True)]
+        cm = _make_config_manager(envs)
+        result = _parse_command("миграц дашборд 42 с dev на prod", cm)
+        assert result["operation"] == "execute_migration"
+        assert result["entities"]["source_env"] == "dev"
+        assert result["entities"]["target_env"] == "prod"
+        assert result["risk_level"] == "dangerous"  # because tgt=prod
+
+    def test_migration_dry_run(self):
+        from src.api.routes.assistant._command_parser import _parse_command
+        cm = _make_config_manager()
+        result = _parse_command("migrate dashboard 42 from dev to staging --dry-run", cm)
+        assert result["entities"]["dry_run"] is True
+        assert result["entities"]["replace_db_config"] is False
+        assert result["entities"]["fix_cross_filters"] is True
+
+    def test_migration_replace_db(self):
+        from src.api.routes.assistant._command_parser import _parse_command
+        cm = _make_config_manager()
+        result = _parse_command("миграция дашборд 42 с dev на qa --replace-db-config --no-fix-cross-filters", cm)
+        assert result["entities"]["replace_db_config"] is True
+        assert result["entities"]["fix_cross_filters"] is False
+
+    def test_migration_without_dashboard(self):
+        from src.api.routes.assistant._command_parser import _parse_command
+        cm = _make_config_manager()
+        result = _parse_command("migrate from dev to staging", cm)
+        assert result["confidence"] == 0.72
+
+
+class TestParseCommandBackup:
+    """_parse_command — backup variants."""
+
+    @pytest.mark.parametrize("msg,expected_env", [
+        ("бэкап в dev", "dev"),
+        ("backup for prod", "prod"),
+        ("резерв из staging", "staging"),
+        ("бэкап", None),
+    ])
+    def test_backup(self, msg, expected_env):
+        from src.api.routes.assistant._command_parser import _parse_command
+        cm = _make_config_manager()
+        result = _parse_command(msg, cm)
+        assert result["operation"] == "run_backup"
+        assert result["entities"].get("environment") == expected_env
+        assert result["risk_level"] == "guarded"
+
+
+class TestParseCommandHealth:
+    """_parse_command — health check variants."""
+
+    @pytest.mark.parametrize("msg,expected_env", [
+        ("здоровье в dev", "dev"),
+        ("health for prod", "prod"),
+        ("ошибки", None),
+        ("failing env staging", "staging"),
+        ("проблемы", None),
+    ])
+    def test_health(self, msg, expected_env):
+        from src.api.routes.assistant._command_parser import _parse_command
+        cm = _make_config_manager()
+        result = _parse_command(msg, cm)
+        assert result["domain"] == "health"
+        assert result["operation"] == "get_health_summary"
+        assert result["entities"].get("environment") == expected_env
+        assert result["risk_level"] == "safe"
+
+
+class TestParseCommandValidation:
+    """_parse_command — LLM validation variants."""
+
+    def test_validation_with_provider(self):
+        from src.api.routes.assistant._command_parser import _parse_command
+        cm = _make_config_manager()
+        result = _parse_command("валидац дашборд 42 для env dev провайдер openai", cm)
+        assert result["operation"] == "run_llm_validation"
+        assert result["entities"]["environment"] == "dev"
+        assert result["entities"]["provider"] == "openai"
+
+    def test_validation_without_env(self):
+        from src.api.routes.assistant._command_parser import _parse_command
+        cm = _make_config_manager()
+        # "проверь" alone without a dashboard id -> low confidence
+        result = _parse_command("проверь", cm)
+        assert result["operation"] == "run_llm_validation"
+        assert result["confidence"] == 0.64
+
+
+class TestParseCommandDocumentation:
+    """_parse_command — LLM documentation variants."""
+
+    def test_documentation_with_dataset(self):
+        from src.api.routes.assistant._command_parser import _parse_command
+        cm = _make_config_manager()
+        result = _parse_command("документац датасет 123 env prod провайдер anthropic", cm)
+        assert result["operation"] == "run_llm_documentation"
+        assert result["entities"]["dataset_id"] == 123
+        assert result["entities"]["environment"] == "prod"
+        assert result["entities"]["provider"] == "anthropic"
+
+    def test_documentation_without_dataset(self):
+        from src.api.routes.assistant._command_parser import _parse_command
+        cm = _make_config_manager()
+        result = _parse_command("generate docs", cm)
+        assert result["operation"] == "run_llm_documentation"
+        assert result["confidence"] == 0.64
+
+
+class TestParseCommandSearchDashboards:
+    """_parse_command — search/list dashboards."""
+
+    @pytest.mark.parametrize("msg,expected_env", [
+        ("список дашборд в dev", "dev"),
+        ("покажи дашборд в prod", "prod"),
+        ("найди дашборд test в staging", "staging"),
+        ("list dashboard в dev", "dev"),
+        ("search dashboard covid в dev", "dev"),
+        ("all dashboards", None),
+        ("все дашборд", None),
+    ])
+    def test_search_dashboards(self, msg, expected_env):
+        from src.api.routes.assistant._command_parser import _parse_command
+        cm = _make_config_manager()
+        result = _parse_command(msg, cm)
+        assert result["domain"] == "dashboards"
+        assert result["operation"] == "search_dashboards"
+        assert result["entities"].get("environment") == expected_env
+
+
+class TestParseCommandEnvironments:
+    """_parse_command — list environments."""
+
+    @pytest.mark.parametrize("msg", [
+        "окружени",
+        "environment",
+        "среды",
+        "сред",
+    ])
+    def test_list_environments(self, msg):
+        from src.api.routes.assistant._command_parser import _parse_command
+        cm = _make_config_manager()
+        result = _parse_command(msg, cm)
+        assert result["domain"] == "environments"
+        assert result["operation"] == "list_environments"
+        assert result["risk_level"] == "safe"
+
+
+class TestParseCommandMaintenance:
+    """_parse_command — maintenance/events."""
+
+    @pytest.mark.parametrize("msg", [
+        "обслуживан",
+        "maintenance",
+        "события",
+    ])
+    def test_list_maintenance(self, msg):
+        from src.api.routes.assistant._command_parser import _parse_command
+        cm = _make_config_manager()
+        result = _parse_command(msg, cm)
+        assert result["domain"] == "maintenance"
+        assert result["operation"] == "list_maintenance_events"
+
+
+class TestParseCommandLlmProviders:
+    """_parse_command — LLM provider listing/status."""
+
+    def test_list_providers(self):
+        from src.api.routes.assistant._command_parser import _parse_command
+        cm = _make_config_manager()
+        result = _parse_command("llm провайдер", cm)
+        assert result["domain"] == "llm"
+        assert result["operation"] == "list_llm_providers"
+
+    def test_llm_status_unreachable_via_parser(self):
+        """The deterministic parser cannot deliver get_llm_status because all
+        keywords matching line 99 are intercepted by earlier checks (status,
+        validation). This is a known source-code quirk; LLM status routing
+        happens through the LLM planner path instead."""
+        from src.api.routes.assistant._command_parser import _parse_command
+        cm = _make_config_manager()
+        result = _parse_command("llm провер", cm)
+        # Line 74 (провер in validation) catches it first
+        assert result["operation"] == "run_llm_validation"
+
+
+class TestParseCommandFallback:
+    """_parse_command — unknown commands return clarify intent."""
+
+    @pytest.mark.parametrize("msg", [
+        "",
+        "hello world",
+        "какой сегодня день",
+        "random gibberish xyz",
+        "12345",
+    ])
+    def test_fallback_to_clarify(self, msg):
+        from src.api.routes.assistant._command_parser import _parse_command
+        cm = _make_config_manager()
+        result = _parse_command(msg, cm)
+        assert result["domain"] == "unknown"
+        assert result["operation"] == "clarify"
+        assert result["confidence"] == 0.3
+        assert result["risk_level"] == "safe"
+        assert "domain" in result
+        assert "operation" in result
+        assert "entities" in result
+        assert "confidence" in result
+        assert "risk_level" in result
+        assert "requires_confirmation" in result
+# #endregion Test.Assistant.CommandParser
diff --git a/backend/tests/api/test_assistant_dataset_review.py b/backend/tests/api/test_assistant_dataset_review.py
new file mode 100644
index 00000000..d2d59897
--- /dev/null
+++ b/backend/tests/api/test_assistant_dataset_review.py
@@ -0,0 +1,726 @@
+# #region Test.Assistant.DatasetReview [C:2] [TYPE Module] [SEMANTICS test,assistant,dataset,review,context,intent]
+# @BRIEF Tests for _dataset_review.py and _dataset_review_dispatch.py — context loading, intent planning, dispatch.
+# @RELATION BINDS_TO -> [AssistantDatasetReview]
+# @RELATION BINDS_TO -> [AssistantDatasetReviewDispatch]
+# @TEST_EDGE: serialize_context -> Session context serialization with all fields
+# @TEST_EDGE: extract_target -> Target extraction from messages
+# @TEST_EDGE: match_field -> Field matching by id/name/verbose name
+# @TEST_EDGE: extract_quoted_segment -> Quoted segment extraction
+# @TEST_EDGE: plan_intent_approve -> Approve mappings intent
+# @TEST_EDGE: plan_intent_preview -> Generate SQL preview intent
+# @TEST_EDGE: plan_intent_semantic -> Set field semantics intent
+# @TEST_EDGE: plan_intent_context -> Answer context intent
+# @TEST_EDGE: dispatch_answer_context -> Context-only dispatch
+# @TEST_EDGE: dispatch_approve_mappings -> Mapping approval dispatch
+# @TEST_EDGE: dispatch_set_field -> Field semantic update dispatch
+# @TEST_EDGE: dispatch_generate_preview -> SQL preview generation dispatch
+
+from pathlib import Path
+import sys
+
+sys.path.insert(0, str(Path(__file__).parent.parent / "src"))
+
+from datetime import UTC, datetime
+from unittest.mock import MagicMock, AsyncMock, patch, PropertyMock
+import pytest
+from fastapi import HTTPException
+
+
+# ── Shared helpers ─────────────────────────────────────────────────────
+
+
+def _make_session_mock(**overrides):
+    """Create a mock DatasetReviewSession with realistic defaults."""
+    from src.models.dataset_review import ReadinessState, RecommendedAction, SessionStatus, SessionPhase
+
+    s = MagicMock()
+    s.session_id = overrides.get("session_id", "sess-1")
+    s.version = overrides.get("version", 5)
+    s.dataset_ref = overrides.get("dataset_ref", "dataset_123")
+    s.dataset_id = overrides.get("dataset_id", 123)
+    s.environment_id = overrides.get("environment_id", "env-1")
+    s.user_id = overrides.get("user_id", "user-1")
+
+    rs = overrides.get("readiness_state", ReadinessState.REVIEW_READY)
+    s.readiness_state = rs if isinstance(rs, MagicMock) else MagicMock(value=rs.value if hasattr(rs, 'value') else rs)
+    s.readiness_state.value = rs.value if hasattr(rs, 'value') else rs
+
+    ra = overrides.get("recommended_action", RecommendedAction.REVIEW_DOCUMENTATION)
+    s.recommended_action = ra if isinstance(ra, MagicMock) else MagicMock(value=ra.value if hasattr(ra, 'value') else ra)
+    s.recommended_action.value = ra.value if hasattr(ra, 'value') else ra
+
+    st = overrides.get("status", SessionStatus.ACTIVE)
+    s.status = st if isinstance(st, MagicMock) else MagicMock(value=st.value if hasattr(st, 'value') else st)
+    s.status.value = st.value if hasattr(st, 'value') else st
+
+    cp = overrides.get("current_phase", SessionPhase.REVIEW)
+    s.current_phase = cp if isinstance(cp, MagicMock) else MagicMock(value=cp.value if hasattr(cp, 'value') else cp)
+    s.current_phase.value = cp.value if hasattr(cp, 'value') else cp
+
+    s.findings = overrides.get("findings", [])
+    s.imported_filters = overrides.get("imported_filters", [])
+    s.execution_mappings = overrides.get("execution_mappings", [])
+    s.semantic_fields = overrides.get("semantic_fields", [])
+    s.previews = overrides.get("previews", [])
+    s.last_activity_at = overrides.get("last_activity_at", datetime.now(UTC))
+    return s
+
+
+def _make_finding_mock(finding_id="f-1", code="TEST", severity="WARNING", message="Test finding", resolution_state="OPEN"):
+    f = MagicMock()
+    f.finding_id = finding_id
+    f.code = code
+    f.severity = MagicMock(value=severity)
+    f.message = message
+    f.resolution_state = MagicMock(value=resolution_state)
+    return f
+
+
+def _make_filter_mock(filter_id="fl-1", filter_name="region", display_name="Region", raw_value="US", normalized_value="USA", source="SUPERSET_URL", confidence_state="HIGH", requires_confirmation=False, recovery_status="RECOVERED", notes=None):
+    fl = MagicMock()
+    fl.filter_id = filter_id
+    fl.filter_name = filter_name
+    fl.display_name = display_name
+    fl.raw_value = raw_value
+    fl.raw_value_masked = False
+    fl.normalized_value = normalized_value
+    fl.source = MagicMock(value=source)
+    fl.confidence_state = MagicMock(value=confidence_state)
+    fl.requires_confirmation = requires_confirmation
+    fl.recovery_status = MagicMock(value=recovery_status)
+    fl.notes = notes
+    return fl
+
+
+def _make_mapping_mock(mapping_id="m-1", filter_id="fl-1", variable_id="v-1", mapping_method="DIRECT_MATCH", effective_value="US", approval_state="PENDING", requires_explicit_approval=True):
+    m = MagicMock()
+    m.mapping_id = mapping_id
+    m.filter_id = filter_id
+    m.variable_id = variable_id
+    m.mapping_method = MagicMock(value=mapping_method)
+    m.effective_value = effective_value
+    m.approval_state = MagicMock(value=approval_state)
+    m.requires_explicit_approval = requires_explicit_approval
+    return m
+
+
+def _make_field_mock(field_id="fld-1", field_name="region", verbose_name="Region", description="The region field", display_format="text", provenance="DICTIONARY_EXACT", is_locked=False, needs_review=False, has_conflict=False):
+    f = MagicMock()
+    f.field_id = field_id
+    f.field_name = field_name
+    f.verbose_name = verbose_name
+    f.description = description
+    f.display_format = display_format
+    f.provenance = MagicMock(value=provenance)
+    f.is_locked = is_locked
+    f.needs_review = needs_review
+    f.has_conflict = has_conflict
+    return f
+
+
+class TestSerializeDatasetReviewContext:
+    """_serialize_dataset_review_context — context serialization."""
+
+    def test_basic_context(self):
+        from src.api.routes.assistant._dataset_review import _serialize_dataset_review_context
+
+        session = _make_session_mock()
+        ctx = _serialize_dataset_review_context(session)
+        assert ctx["session_id"] == "sess-1"
+        assert ctx["version"] == 5
+        assert ctx["dataset_ref"] == "dataset_123"
+        assert ctx["status"] == "active"
+
+    def test_context_with_findings_and_filters(self):
+        from src.api.routes.assistant._dataset_review import _serialize_dataset_review_context
+
+        session = _make_session_mock(findings=[_make_finding_mock()], imported_filters=[_make_filter_mock()], execution_mappings=[_make_mapping_mock()])
+        ctx = _serialize_dataset_review_context(session)
+        assert len(ctx["findings"]) == 1
+        assert ctx["findings"][0]["finding_id"] == "f-1"
+        assert len(ctx["imported_filters"]) == 1
+        assert ctx["imported_filters"][0]["filter_name"] == "region"
+        assert len(ctx["mappings"]) == 1
+        assert ctx["mappings"][0]["mapping_id"] == "m-1"
+
+    def test_context_with_semantic_fields(self):
+        from src.api.routes.assistant._dataset_review import _serialize_dataset_review_context
+
+        session = _make_session_mock(semantic_fields=[_make_field_mock()])
+        ctx = _serialize_dataset_review_context(session)
+        assert len(ctx["semantic_fields"]) == 1
+        assert ctx["semantic_fields"][0]["field_name"] == "region"
+
+    def test_context_with_preview(self):
+        from src.api.routes.assistant._dataset_review import _serialize_dataset_review_context
+
+        preview_mock = MagicMock()
+        preview_mock.preview_id = "pv-1"
+        preview_mock.preview_status = MagicMock(value="READY")
+        preview_mock.preview_fingerprint = "fp-abc"
+        session = _make_session_mock(previews=[preview_mock])
+        ctx = _serialize_dataset_review_context(session)
+        assert ctx["preview"]["preview_id"] == "pv-1"
+
+    def test_context_no_previews(self):
+        from src.api.routes.assistant._dataset_review import _serialize_dataset_review_context
+        session = _make_session_mock(previews=[])
+        ctx = _serialize_dataset_review_context(session)
+        assert ctx["preview"] is None
+
+    def test_context_without_some_attrs(self):
+        """Handle missing attributes gracefully."""
+        from src.api.routes.assistant._dataset_review import _serialize_dataset_review_context
+        s = MagicMock()
+        s.session_id = "sess-1"
+        s.version = 0
+        s.dataset_ref = "ds"
+        s.dataset_id = None
+        s.environment_id = "env-1"
+        # Some attributes may be missing
+        del s.readiness_state
+        s.recommended_action = MagicMock(value="NONE")
+        s.status = MagicMock(value="ACTIVE")
+        s.current_phase = MagicMock(value="REVIEW")
+        s.findings = []
+        s.imported_filters = []
+        s.execution_mappings = []
+        s.semantic_fields = []
+        s.previews = []
+        ctx = _serialize_dataset_review_context(s)
+        # Should handle gracefully or raise AttributeError
+        assert ctx["session_id"] == "sess-1"
+
+
+class TestLoadDatasetReviewContext:
+    """_load_dataset_review_context — context loading from repository."""
+
+    def test_load_with_valid_session(self):
+        from src.api.routes.assistant._dataset_review import _load_dataset_review_context
+        from src.schemas.auth import User
+
+        db = MagicMock()
+        user = User(id="user-1", username="test", email="test@x.com", is_active=True, auth_source="internal", created_at=datetime.now(UTC), last_login=datetime.now(UTC), roles=[])
+
+        session = _make_session_mock(user_id="user-1")
+        with patch("src.api.routes.assistant._dataset_review.DatasetReviewSessionRepository") as mock_repo:
+            mock_repo.return_value.load_session_detail.return_value = session
+            result = _load_dataset_review_context("sess-1", user, db)
+            assert result is not None
+            assert result["session_id"] == "sess-1"
+
+    def test_load_with_none_session_id(self):
+        from src.api.routes.assistant._dataset_review import _load_dataset_review_context
+        from src.schemas.auth import User
+
+        db = MagicMock()
+        user = User(id="user-1", username="test", email="test@x.com", is_active=True, auth_source="internal", created_at=datetime.now(UTC), last_login=datetime.now(UTC), roles=[])
+        result = _load_dataset_review_context(None, user, db)
+        assert result is None
+
+    def test_load_with_wrong_user_raises_404(self):
+        from src.api.routes.assistant._dataset_review import _load_dataset_review_context
+        from src.schemas.auth import User
+
+        db = MagicMock()
+        user = User(id="user-1", username="test", email="test@x.com", is_active=True, auth_source="internal", created_at=datetime.now(UTC), last_login=datetime.now(UTC), roles=[])
+
+        # Session belongs to different user
+        session = _make_session_mock(user_id="user-2")
+        with patch("src.api.routes.assistant._dataset_review.DatasetReviewSessionRepository") as mock_repo:
+            mock_repo.return_value.load_session_detail.return_value = session
+            with pytest.raises(HTTPException) as exc:
+                _load_dataset_review_context("sess-1", user, db)
+            assert exc.value.status_code == 404
+
+    def test_load_with_missing_session_raises_404(self):
+        from src.api.routes.assistant._dataset_review import _load_dataset_review_context
+        from src.schemas.auth import User
+
+        db = MagicMock()
+        user = User(id="user-1", username="test", email="test@x.com", is_active=True, auth_source="internal", created_at=datetime.now(UTC), last_login=datetime.now(UTC), roles=[])
+
+        with patch("src.api.routes.assistant._dataset_review.DatasetReviewSessionRepository") as mock_repo:
+            mock_repo.return_value.load_session_detail.return_value = None
+            with pytest.raises(HTTPException) as exc:
+                _load_dataset_review_context("sess-1", user, db)
+            assert exc.value.status_code == 404
+
+
+class TestExtractDatasetReviewTarget:
+    """_extract_dataset_review_target — target extraction."""
+
+    @pytest.mark.parametrize("msg,expected_kind,expected_id", [
+        ("target:field=abc123", "field", "abc123"),
+        ("focus:field=xyz789", "field", "xyz789"),
+        ("target:mapping=map_1", "mapping", "map_1"),
+        ("target:finding=fnd_1", "finding", "fnd_1"),
+        ("target:filter=flt_1", "filter", "flt_1"),
+        ("focus = finding=abc", "finding", "abc"),
+        ("target = field = def", "field", "def"),
+        ("no target here", None, None),
+        ("", None, None),
+    ])
+    def test_extract(self, msg, expected_kind, expected_id):
+        from src.api.routes.assistant._dataset_review import _extract_dataset_review_target
+        kind, tid = _extract_dataset_review_target(msg)
+        assert kind == expected_kind
+        assert tid == expected_id
+
+
+class TestMatchDatasetReviewField:
+    """_match_dataset_review_field — field matching."""
+
+    def test_match_by_id(self):
+        from src.api.routes.assistant._dataset_review import _match_dataset_review_field
+
+        ctx = {"semantic_fields": [{"field_id": "fld-1", "field_name": "region", "verbose_name": "Region"}]}
+        result = _match_dataset_review_field(ctx, "target:field=fld-1")
+        assert result is not None
+        assert result["field_id"] == "fld-1"
+
+    def test_match_by_field_name(self):
+        from src.api.routes.assistant._dataset_review import _match_dataset_review_field
+
+        ctx = {"semantic_fields": [{"field_id": "fld-1", "field_name": "region", "verbose_name": "Region"}]}
+        result = _match_dataset_review_field(ctx, "please update region field")
+        assert result is not None
+        assert result["field_name"] == "region"
+
+    def test_match_by_verbose_name(self):
+        from src.api.routes.assistant._dataset_review import _match_dataset_review_field
+
+        ctx = {"semantic_fields": [{"field_id": "fld-1", "field_name": "region", "verbose_name": "Region Name"}]}
+        result = _match_dataset_review_field(ctx, "change Region Name")
+        assert result is not None
+        assert result["verbose_name"] == "Region Name"
+
+    def test_no_match(self):
+        from src.api.routes.assistant._dataset_review import _match_dataset_review_field
+
+        ctx = {"semantic_fields": [{"field_id": "fld-1", "field_name": "region", "verbose_name": "Region"}]}
+        result = _match_dataset_review_field(ctx, "something completely different")
+        assert result is None
+
+    def test_empty_fields(self):
+        from src.api.routes.assistant._dataset_review import _match_dataset_review_field
+
+        ctx = {"semantic_fields": []}
+        result = _match_dataset_review_field(ctx, "region")
+        assert result is None
+
+
+class TestExtractQuotedSegment:
+    """_extract_quoted_segment — quoted command segment extraction."""
+
+    @pytest.mark.parametrize("msg,label,expected", [
+        ('verbose_name="My Label"', "verbose_name", "My Label"),
+        ("verbose_name='My Label'", "verbose_name", "My Label"),
+        ('description="Some desc"', "description", "Some desc"),
+        ('format="text"', "display_format", "text"),
+        ('verbose_name: "Label"', "verbose_name", "Label"),
+        ('verbose_name = "Label"', "verbose_name", "Label"),
+        ("no label here", "verbose_name", None),
+    ])
+    def test_extract(self, msg, label, expected):
+        from src.api.routes.assistant._dataset_review import _extract_quoted_segment
+        result = _extract_quoted_segment(msg, label)
+        assert result == expected
+
+
+class TestPlanDatasetReviewIntent:
+    """_plan_dataset_review_intent — intent planning from message."""
+
+    @pytest.fixture
+    def context(self):
+        return {
+            "session_id": "sess-1",
+            "version": 5,
+            "readiness_state": "REVIEW_READY",
+            "recommended_action": "REVIEW_DOCUMENTATION",
+            "mappings": [
+                {"mapping_id": "m-1", "requires_explicit_approval": True, "approval_state": "PENDING"},
+                {"mapping_id": "m-2", "requires_explicit_approval": True, "approval_state": "PENDING"},
+            ],
+            "findings": [{"code": "TEST"}],
+            "imported_filters": [{"filter_name": "region"}],
+            "preview": {"preview_status": "missing"},
+            "semantic_fields": [],
+        }
+
+    def test_approve_mappings(self, context):
+        from src.api.routes.assistant._dataset_review import _plan_dataset_review_intent
+        result = _plan_dataset_review_intent("approve mappings", context)
+        assert result is not None
+        assert result["operation"] == "dataset_review_approve_mappings"
+        assert result["entities"]["mapping_ids"] == ["m-1", "m-2"]
+        assert result["requires_confirmation"] is True
+
+    def test_approve_mappings_russian(self, context):
+        from src.api.routes.assistant._dataset_review import _plan_dataset_review_intent
+        result = _plan_dataset_review_intent("подтверди маппинг", context)
+        assert result is not None
+        assert result["operation"] == "dataset_review_approve_mappings"
+
+    def test_generate_sql_preview(self, context):
+        from src.api.routes.assistant._dataset_review import _plan_dataset_review_intent
+        result = _plan_dataset_review_intent("generate sql preview", context)
+        assert result is not None
+        assert result["operation"] == "dataset_review_generate_sql_preview"
+
+    def test_generate_preview_russian(self, context):
+        from src.api.routes.assistant._dataset_review import _plan_dataset_review_intent
+        result = _plan_dataset_review_intent("сгенерируй превью", context)
+        assert result is not None
+        assert result["operation"] == "dataset_review_generate_sql_preview"
+
+    def test_set_field_semantics_with_target(self, context):
+        from src.api.routes.assistant._dataset_review import _plan_dataset_review_intent
+        ctx_with_field = dict(context)
+        ctx_with_field["semantic_fields"] = [
+            {"field_id": "fld-1", "field_name": "region", "verbose_name": "Region", "candidates": []}
+        ]
+        result = _plan_dataset_review_intent('set field semantics target:field=fld-1 verbose_name="Country"', ctx_with_field)
+        assert result is not None
+        assert result["operation"] == "dataset_review_set_field_semantics"
+        assert result["entities"]["verbose_name"] == "Country"
+
+    def test_set_field_semantics_no_field_returns_none(self, context):
+        from src.api.routes.assistant._dataset_review import _plan_dataset_review_intent
+        # No matching field in message and no target
+        result = _plan_dataset_review_intent("set field semantics", context)
+        assert result is None
+
+    def test_set_field_semantics_with_lock(self, context):
+        from src.api.routes.assistant._dataset_review import _plan_dataset_review_intent
+        ctx_with_field = dict(context)
+        ctx_with_field["semantic_fields"] = [
+            {"field_id": "fld-1", "field_name": "region", "verbose_name": "Region", "candidates": []}
+        ]
+        result = _plan_dataset_review_intent('apply field semantics target:field=fld-1 lock it verbose_name="Country"', ctx_with_field)
+        assert result is not None
+        assert result["entities"]["lock_field"] is True
+
+    def test_answer_context(self, context):
+        from src.api.routes.assistant._dataset_review import _plan_dataset_review_intent
+        result = _plan_dataset_review_intent("what are the filters?", context)
+        assert result is not None
+        assert result["operation"] == "dataset_review_answer_context"
+        assert "readiness_state" in result["entities"]["summary"]
+        assert result["requires_confirmation"] is False
+
+    def test_answer_context_russian(self, context):
+        from src.api.routes.assistant._dataset_review import _plan_dataset_review_intent
+        result = _plan_dataset_review_intent("фильтр", context)
+        assert result is not None
+        assert result["operation"] == "dataset_review_answer_context"
+
+    def test_unrelated_message_returns_none(self, context):
+        from src.api.routes.assistant._dataset_review import _plan_dataset_review_intent
+        result = _plan_dataset_review_intent("hello world", context)
+        assert result is None
+
+
+class TestDispatchDatasetReviewIntent:
+    """_dispatch_dataset_review_intent — intent dispatch."""
+
+    @pytest.fixture
+    def user(self):
+        from src.schemas.auth import User
+        return User(id="user-1", username="test", email="test@x.com", is_active=True, auth_source="internal", created_at=datetime.now(UTC), last_login=datetime.now(UTC), roles=[])
+
+    @pytest.fixture
+    def config_manager(self):
+        cm = MagicMock()
+        cm.get_environments = MagicMock(return_value=[])
+        return cm
+
+    def test_answer_context(self, user, config_manager):
+        from src.api.routes.assistant._dataset_review_dispatch import _dispatch_dataset_review_intent
+
+        db = MagicMock()
+        intent = {
+            "operation": "dataset_review_answer_context",
+            "entities": {
+                "dataset_review_session_id": "sess-1",
+                "session_version": 5,
+                "summary": "Session sess-1: readiness=READY",
+            },
+        }
+        text, task_id, actions = (
+            __import__("asyncio").run(_dispatch_dataset_review_intent(intent, user, config_manager, db))
+        )
+        assert "READY" in text
+        assert task_id is None
+
+    def test_missing_session_id_raises_422(self, user, config_manager):
+        from src.api.routes.assistant._dataset_review_dispatch import _dispatch_dataset_review_intent
+
+        db = MagicMock()
+        intent = {"operation": "test", "entities": {}}
+        with pytest.raises(HTTPException) as exc:
+            __import__("asyncio").run(_dispatch_dataset_review_intent(intent, user, config_manager, db))
+        assert exc.value.status_code == 422
+
+    def test_unsupported_operation_raises_400(self, user, config_manager):
+        from src.api.routes.assistant._dataset_review_dispatch import _dispatch_dataset_review_intent
+
+        db = MagicMock()
+        intent = {
+            "operation": "dataset_review_unknown",
+            "entities": {"dataset_review_session_id": "sess-1", "session_version": 5},
+        }
+
+        session = _make_session_mock()
+        repository_cls = "src.api.routes.assistant._dataset_review_dispatch.DatasetReviewSessionRepository"
+        with patch(repository_cls) as mock_repo_cls:
+            mock_repo = MagicMock()
+            mock_repo_cls.return_value = mock_repo
+            mock_repo.load_session_detail.return_value = session
+            mock_repo.require_session_version.return_value = None
+            mock_repo.bump_session_version.return_value = None
+            mock_repo.db = MagicMock()
+
+            with pytest.raises(HTTPException) as exc:
+                __import__("asyncio").run(_dispatch_dataset_review_intent(intent, user, config_manager, db))
+            assert exc.value.status_code == 400
+
+    def test_approve_mappings_success(self, user, config_manager):
+        from src.api.routes.assistant._dataset_review_dispatch import _dispatch_dataset_review_intent
+        from src.models.dataset_review import ApprovalState, ReadinessState
+
+        db = MagicMock()
+        mapping = _make_mapping_mock(mapping_id="m-1", filter_id="fl-1", variable_id="v-1",
+                                      approval_state=ApprovalState.PENDING.value,
+                                      requires_explicit_approval=True)
+
+        session = _make_session_mock(execution_mappings=[mapping],
+                                     readiness_state=ReadinessState.MAPPING_REVIEW_NEEDED,
+                                     user_id="user-1")
+
+        intent = {
+            "operation": "dataset_review_approve_mappings",
+            "entities": {
+                "dataset_review_session_id": "sess-1",
+                "session_version": 5,
+                "mapping_ids": ["m-1"],
+            },
+        }
+
+        repository_cls = "src.api.routes.assistant._dataset_review_dispatch.DatasetReviewSessionRepository"
+        with patch(repository_cls) as mock_repo_cls:
+            mock_repo = MagicMock()
+            mock_repo_cls.return_value = mock_repo
+            mock_repo.load_session_detail.return_value = session
+            mock_repo.require_session_version.return_value = None
+            mock_repo.bump_session_version.return_value = None
+            mock_repo.db = MagicMock()
+            mock_repo.event_logger = MagicMock()
+
+            text, task_id, actions = (
+                __import__("asyncio").run(_dispatch_dataset_review_intent(intent, user, config_manager, db))
+            )
+            assert "Approved" in text
+            assert "1 mapping" in text
+
+    def test_approve_mappings_no_pending_raises_409(self, user, config_manager):
+        from src.api.routes.assistant._dataset_review_dispatch import _dispatch_dataset_review_intent
+
+        db = MagicMock()
+        intent = {
+            "operation": "dataset_review_approve_mappings",
+            "entities": {
+                "dataset_review_session_id": "sess-1",
+                "session_version": 5,
+                "mapping_ids": [],
+            },
+        }
+
+        with pytest.raises(HTTPException) as exc:
+            __import__("asyncio").run(_dispatch_dataset_review_intent(intent, user, config_manager, db))
+        assert exc.value.status_code == 409
+        assert "No pending mappings" in exc.value.detail
+
+    def test_approve_mappings_no_match_raises_409(self, user, config_manager):
+        from src.api.routes.assistant._dataset_review_dispatch import _dispatch_dataset_review_intent
+        from src.models.dataset_review import ApprovalState
+
+        db = MagicMock()
+        session = _make_session_mock(execution_mappings=[], user_id="user-1")
+
+        intent = {
+            "operation": "dataset_review_approve_mappings",
+            "entities": {
+                "dataset_review_session_id": "sess-1",
+                "session_version": 5,
+                "mapping_ids": ["m-nonexistent"],
+            },
+        }
+
+        repository_cls = "src.api.routes.assistant._dataset_review_dispatch.DatasetReviewSessionRepository"
+        with patch(repository_cls) as mock_repo_cls:
+            mock_repo = MagicMock()
+            mock_repo_cls.return_value = mock_repo
+            mock_repo.load_session_detail.return_value = session
+            mock_repo.require_session_version.return_value = None
+
+            with pytest.raises(HTTPException) as exc:
+                __import__("asyncio").run(_dispatch_dataset_review_intent(intent, user, config_manager, db))
+            assert exc.value.status_code == 409
+
+    def test_set_field_semantics_success(self, user, config_manager):
+        from src.api.routes.assistant._dataset_review_dispatch import _dispatch_dataset_review_intent
+
+        field = _make_field_mock()
+        session = _make_session_mock(semantic_fields=[field], user_id="user-1")
+
+        intent = {
+            "operation": "dataset_review_set_field_semantics",
+            "entities": {
+                "dataset_review_session_id": "sess-1",
+                "session_version": 5,
+                "field_id": "fld-1",
+                "verbose_name": "Country",
+            },
+        }
+
+        db = MagicMock()
+        repository_cls = "src.api.routes.assistant._dataset_review_dispatch.DatasetReviewSessionRepository"
+        with patch(repository_cls) as mock_repo_cls:
+            mock_repo = MagicMock()
+            mock_repo_cls.return_value = mock_repo
+            mock_repo.load_session_detail.return_value = session
+            mock_repo.require_session_version.return_value = None
+            mock_repo.bump_session_version.return_value = None
+            mock_repo.db = MagicMock()
+            mock_repo.event_logger = MagicMock()
+
+            with patch("src.api.routes.assistant._dataset_review_dispatch._update_semantic_field_state") as mock_update:
+                mock_update.return_value = None
+                text, task_id, actions = (
+                    __import__("asyncio").run(_dispatch_dataset_review_intent(intent, user, config_manager, db))
+                )
+                assert "Updated semantic field" in text
+
+    def test_set_field_semantics_missing_field_raises_422(self, user, config_manager):
+        from src.api.routes.assistant._dataset_review_dispatch import _dispatch_dataset_review_intent
+
+        session = _make_session_mock(semantic_fields=[], user_id="user-1")
+        intent = {
+            "operation": "dataset_review_set_field_semantics",
+            "entities": {
+                "dataset_review_session_id": "sess-1",
+                "session_version": 5,
+                "field_id": "",
+            },
+        }
+
+        db = MagicMock()
+        repository_cls = "src.api.routes.assistant._dataset_review_dispatch.DatasetReviewSessionRepository"
+        with patch(repository_cls) as mock_repo_cls:
+            mock_repo = MagicMock()
+            mock_repo_cls.return_value = mock_repo
+            mock_repo.load_session_detail.return_value = session
+            mock_repo.require_session_version.return_value = None
+
+            with pytest.raises(HTTPException) as exc:
+                __import__("asyncio").run(_dispatch_dataset_review_intent(intent, user, config_manager, db))
+            assert exc.value.status_code == 422
+
+    def test_session_not_found_raises_404(self, user, config_manager):
+        from src.api.routes.assistant._dataset_review_dispatch import _dispatch_dataset_review_intent
+
+        db = MagicMock()
+        intent = {
+            "operation": "dataset_review_approve_mappings",
+            "entities": {"dataset_review_session_id": "sess-1", "session_version": 5, "mapping_ids": ["m-1"]},
+        }
+
+        repository_cls = "src.api.routes.assistant._dataset_review_dispatch.DatasetReviewSessionRepository"
+        with patch(repository_cls) as mock_repo_cls:
+            mock_repo = MagicMock()
+            mock_repo_cls.return_value = mock_repo
+            mock_repo.load_session_detail.return_value = None
+
+            with pytest.raises(HTTPException) as exc:
+                __import__("asyncio").run(_dispatch_dataset_review_intent(intent, user, config_manager, db))
+            assert exc.value.status_code == 404
+
+    def test_version_conflict_raises_409(self, user, config_manager):
+        from src.api.routes.assistant._dataset_review_dispatch import _dispatch_dataset_review_intent, DatasetReviewSessionVersionConflictError
+
+        db = MagicMock()
+        intent = {
+            "operation": "dataset_review_approve_mappings",
+            "entities": {"dataset_review_session_id": "sess-1", "session_version": 4, "mapping_ids": ["m-1"]},
+        }
+
+        session = _make_session_mock(user_id="user-1")
+        repository_cls = "src.api.routes.assistant._dataset_review_dispatch.DatasetReviewSessionRepository"
+        with patch(repository_cls) as mock_repo_cls:
+            mock_repo = MagicMock()
+            mock_repo_cls.return_value = mock_repo
+            mock_repo.load_session_detail.return_value = session
+            mock_repo.require_session_version.side_effect = DatasetReviewSessionVersionConflictError(
+                session_id="sess-1", expected_version=4, actual_version=5
+            )
+
+            with pytest.raises(HTTPException) as exc:
+                __import__("asyncio").run(_dispatch_dataset_review_intent(intent, user, config_manager, db))
+            assert exc.value.status_code == 409
+            assert "session_version_conflict" in exc.value.detail["error_code"]
+
+    def test_generate_sql_preview_success(self, user, config_manager):
+        from src.api.routes.assistant._dataset_review_dispatch import _dispatch_dataset_review_intent
+
+        session = _make_session_mock(user_id="user-1")
+        db = MagicMock()
+
+        intent = {
+            "operation": "dataset_review_generate_sql_preview",
+            "entities": {
+                "dataset_review_session_id": "sess-1",
+                "session_version": 5,
+            },
+        }
+
+        repository_cls = "src.api.routes.assistant._dataset_review_dispatch.DatasetReviewSessionRepository"
+        orchestrator_cls = "src.api.routes.assistant._dataset_review_dispatch.DatasetReviewOrchestrator"
+        with patch(repository_cls) as mock_repo_cls, patch(orchestrator_cls) as mock_orch_cls:
+            mock_repo = MagicMock()
+            mock_repo_cls.return_value = mock_repo
+            mock_repo.load_session_detail.return_value = session
+            mock_repo.require_session_version.return_value = None
+            mock_repo.bump_session_version.return_value = None
+            mock_repo.db = MagicMock()
+            mock_repo.event_logger = MagicMock()
+
+            mock_orch = MagicMock()
+            mock_orch_cls.return_value = mock_orch
+            preview_result = MagicMock()
+            preview_result.preview = MagicMock()
+            preview_result.preview.preview_status = "READY"
+            mock_orch.prepare_launch_preview.return_value = preview_result
+
+            text, task_id, actions = (
+                __import__("asyncio").run(_dispatch_dataset_review_intent(intent, user, config_manager, db))
+            )
+            assert "READY" in text
+
+
+class TestDatasetReviewConflictHttpException:
+    """_dataset_review_conflict_http_exception — version conflict error mapping."""
+
+    def test_conflict_exception(self):
+        from src.api.routes.assistant._dataset_review_dispatch import (
+            _dataset_review_conflict_http_exception,
+            DatasetReviewSessionVersionConflictError,
+        )
+        exc = DatasetReviewSessionVersionConflictError(
+            session_id="sess-1", expected_version=4, actual_version=5
+        )
+        http_exc = _dataset_review_conflict_http_exception(exc)
+        assert http_exc.status_code == 409
+        assert http_exc.detail["error_code"] == "session_version_conflict"
+        assert http_exc.detail["session_id"] == "sess-1"
+# #endregion Test.Assistant.DatasetReview
diff --git a/backend/tests/api/test_assistant_dispatch.py b/backend/tests/api/test_assistant_dispatch.py
new file mode 100644
index 00000000..cbb5f853
--- /dev/null
+++ b/backend/tests/api/test_assistant_dispatch.py
@@ -0,0 +1,140 @@
+# #region Test.Assistant.Dispatch [C:2] [TYPE Module] [SEMANTICS test,assistant,dispatch,confirmation,clarification]
+# @BRIEF Tests for _dispatch.py — clarification text, confirmation summary, and dispatch wrapper.
+# @RELATION BINDS_TO -> [AssistantDispatch]
+# @TEST_EDGE: clarification_text -> Missing parameter guidance per operation
+# @TEST_EDGE: confirmation_summary -> Confirmation prompts for supported ops
+# @TEST_EDGE: confirmation_summary_unknown -> Fallback for unknown ops
+# @TEST_EDGE: confirmation_summary_migration -> Migration summary with flags and dry-run
+# @TEST_EDGE: dispatch_intent -> Backward-compat wrapper calls registry
+
+from pathlib import Path
+import sys
+
+sys.path.insert(0, str(Path(__file__).parent.parent / "src"))
+
+from unittest.mock import AsyncMock, MagicMock, patch
+import pytest
+
+
+class TestClarificationTextForIntent:
+    """_clarification_text_for_intent — per-operation guidance texts."""
+
+    @pytest.mark.parametrize("operation,expected_keyword", [
+        ("run_llm_validation", "LLM-валидации"),
+        ("run_llm_documentation", "документации"),
+        ("create_branch", "ветки"),
+        ("commit_changes", "коммита"),
+        ("deploy_dashboard", "дашборд"),
+        ("execute_migration", "source_env"),
+        ("run_backup", "окружение"),
+        ("unknown_op", "default"),
+    ])
+    def test_clarification_text(self, operation, expected_keyword):
+        from src.api.routes.assistant._dispatch import _clarification_text_for_intent
+
+        intent = {"operation": operation}
+        detail = "default clarification needed"
+        text = _clarification_text_for_intent(intent, detail)
+        if expected_keyword == "default":
+            assert text == detail
+        else:
+            assert expected_keyword in text
+
+
+class TestAsyncConfirmationSummary:
+    """_async_confirmation_summary — confirmation prompt generation."""
+
+    @pytest.fixture
+    def config_manager(self):
+        cm = MagicMock()
+        cm.get_environments = MagicMock(return_value=[])
+        return cm
+
+    def _make_intent(self, operation, entities=None):
+        return {"operation": operation, "entities": entities or {}}
+
+    @pytest.mark.parametrize("operation,expected_title_word", [
+        ("create_branch", "ветки"),
+        ("commit_changes", "Коммит"),
+        ("deploy_dashboard", "Деплой"),
+        ("execute_migration", "Миграция"),
+        ("run_backup", "Бэкап"),
+        ("run_llm_validation", "LLM-валидация"),
+        ("run_llm_documentation", "документации"),
+        ("start_maintenance", "Обслуживание"),
+        ("end_maintenance", "Завершение"),
+    ])
+    def test_known_operations(self, operation, expected_title_word, config_manager):
+        from src.api.routes.assistant._dispatch import _async_confirmation_summary
+
+        db = MagicMock()
+        intent = self._make_intent(operation, {
+            "dashboard_id": "42",
+            "branch_name": "feature/test",
+            "environment": "dev",
+            "source_env": "dev",
+            "target_env": "prod",
+            "tables": ["users", "orders"],
+            "dataset_id": "123",
+        })
+        text = __import__("asyncio").run(_async_confirmation_summary(intent, config_manager, db))
+        assert expected_title_word in text
+
+    def test_unknown_operation(self, config_manager):
+        from src.api.routes.assistant._dispatch import _async_confirmation_summary
+
+        db = MagicMock()
+        intent = self._make_intent("unknown_operation")
+        text = __import__("asyncio").run(_async_confirmation_summary(intent, config_manager, db))
+        assert "Подтвердите" in text
+
+    def test_migration_with_flags(self, config_manager):
+        from src.api.routes.assistant._dispatch import _async_confirmation_summary
+
+        db = MagicMock()
+        intent = self._make_intent("execute_migration", {
+            "dashboard_id": "42",
+            "source_env": "dev",
+            "target_env": "staging",
+            "replace_db_config": True,
+            "fix_cross_filters": True,
+            "dry_run": False,
+        })
+        text = __import__("asyncio").run(_async_confirmation_summary(intent, config_manager, db))
+        assert "маппинг" in text  # should mention db mapping
+        assert "cross" in text or "кросс" in text
+
+    def test_migration_with_tables_list(self, config_manager):
+        from src.api.routes.assistant._dispatch import _async_confirmation_summary
+
+        db = MagicMock()
+        intent = self._make_intent("start_maintenance", {
+            "tables": ["users", "orders", "products", "reviews", "logs", "extra"],
+        })
+        text = __import__("asyncio").run(_async_confirmation_summary(intent, config_manager, db))
+        assert "users" in text
+
+
+class TestDispatchIntent:
+    """_dispatch_intent — backward-compat wrapper."""
+
+    @pytest.mark.asyncio
+    async def test_dispatches_to_registry(self):
+        from src.api.routes.assistant._dispatch import _dispatch_intent
+        from src.schemas.auth import User
+        from datetime import UTC, datetime
+
+        user = User(id="user-1", username="test", email="test@x.com", is_active=True, auth_source="internal", created_at=datetime.now(UTC), last_login=datetime.now(UTC), roles=[])
+        task_manager = MagicMock()
+        config_manager = MagicMock()
+        db = MagicMock()
+
+        intent = {"operation": "run_backup", "entities": {"environment": "dev"}}
+
+        with patch("src.api.routes.assistant._dispatch._registry_dispatch",
+                   new=AsyncMock(return_value=("Backup started", "task-42", []))) as mock_dispatch:
+            text, task_id, actions = await _dispatch_intent(intent, user, task_manager, config_manager, db)
+            assert text == "Backup started"
+            assert task_id == "task-42"
+            mock_dispatch.assert_called_once_with("run_backup", intent, user, task_manager, config_manager, db)
+# #endregion Test.Assistant.Dispatch
diff --git a/backend/tests/api/test_assistant_history.py b/backend/tests/api/test_assistant_history.py
new file mode 100644
index 00000000..0abe28cd
--- /dev/null
+++ b/backend/tests/api/test_assistant_history.py
@@ -0,0 +1,444 @@
+# #region Test.Assistant.History [C:2] [TYPE Module] [SEMANTICS test,assistant,history,audit,confirmation]
+# @BRIEF Tests for _history.py — conversation history, audit trail, confirmation persistence helpers.
+# @RELATION BINDS_TO -> [AssistantHistory]
+# @TEST_EDGE: append_history -> In-memory history appends correctly
+# @TEST_EDGE: persist_message -> DB persistence with rollback on failure
+# @TEST_EDGE: audit -> In-memory audit recording
+# @TEST_EDGE: persist_audit -> DB audit persistence
+# @TEST_EDGE: persist_confirmation -> Confirmation token persistence
+# @TEST_EDGE: update_confirmation_state -> State transitions
+# @TEST_EDGE: load_confirmation_from_db -> Confirmation loading
+# @TEST_EDGE: ensure_conversation -> Conversation ID resolution
+# @TEST_EDGE: resolve_or_create_conversation -> DB-backed conversation resolution
+# @TEST_EDGE: cleanup_history_ttl -> TTL-based cleanup
+# @TEST_EDGE: is_conversation_archived -> Archive detection
+# @TEST_EDGE: coerce_query_bool -> Boolean coercion
+
+from pathlib import Path
+import sys
+
+sys.path.insert(0, str(Path(__file__).parent.parent / "src"))
+
+from datetime import UTC, datetime, timedelta
+from unittest.mock import MagicMock, patch
+import uuid
+
+import pytest
+
+
+class TestAppendHistory:
+    """_append_history — in-memory conversation buffer."""
+
+    def test_append_new_conversation(self):
+        from datetime import UTC, datetime
+        from src.api.routes.assistant._schemas import CONVERSATIONS
+        from src.api.routes.assistant._history import _append_history
+
+        CONVERSATIONS.clear()
+        _append_history("user-1", "conv-1", "user", "Hello")
+        key = ("user-1", "conv-1")
+        assert key in CONVERSATIONS
+        assert len(CONVERSATIONS[key]) == 1
+        assert CONVERSATIONS[key][0]["role"] == "user"
+        assert CONVERSATIONS[key][0]["text"] == "Hello"
+
+    def test_append_existing_conversation(self):
+        from src.api.routes.assistant._schemas import CONVERSATIONS
+        from src.api.routes.assistant._history import _append_history
+
+        CONVERSATIONS.clear()
+        _append_history("user-1", "conv-1", "user", "First")
+        _append_history("user-1", "conv-1", "assistant", "Response")
+        key = ("user-1", "conv-1")
+        assert len(CONVERSATIONS[key]) == 2
+        assert CONVERSATIONS[key][0]["role"] == "user"
+        assert CONVERSATIONS[key][1]["role"] == "assistant"
+
+    def test_append_with_state_and_task(self):
+        from src.api.routes.assistant._schemas import CONVERSATIONS
+        from src.api.routes.assistant._history import _append_history
+
+        CONVERSATIONS.clear()
+        _append_history("user-1", "conv-1", "assistant", "Done", state="success", task_id="task-42")
+        key = ("user-1", "conv-1")
+        assert CONVERSATIONS[key][0]["state"] == "success"
+        assert CONVERSATIONS[key][0]["task_id"] == "task-42"
+
+    def test_message_id_is_uuid(self):
+        from src.api.routes.assistant._schemas import CONVERSATIONS
+        from src.api.routes.assistant._history import _append_history
+
+        CONVERSATIONS.clear()
+        _append_history("user-1", "conv-1", "user", "test")
+        mid = CONVERSATIONS[("user-1", "conv-1")][0]["message_id"]
+        assert uuid.UUID(mid)
+
+
+class TestPersistMessage:
+    """_persist_message — DB persistence."""
+
+    def test_persist_success(self):
+        from src.api.routes.assistant._history import _persist_message
+
+        db = MagicMock()
+        _persist_message(db, "user-1", "conv-1", "user", "Hello", state="active")
+        db.add.assert_called_once()
+        db.commit.assert_called_once()
+
+    def test_persist_with_metadata(self):
+        from src.api.routes.assistant._history import _persist_message
+
+        db = MagicMock()
+        _persist_message(db, "user-1", "conv-1", "user", "test", metadata={"key": "val"})
+        # Verify the payload field was set
+        added = db.add.call_args[0][0]
+        assert added.payload == {"key": "val"}
+
+    def test_persist_failure_rollback(self):
+        from src.api.routes.assistant._history import _persist_message
+
+        db = MagicMock()
+        db.commit.side_effect = Exception("DB error")
+        # Should not raise — just log warning
+        _persist_message(db, "user-1", "conv-1", "user", "Hello")
+        db.rollback.assert_called_once()
+
+
+class TestAudit:
+    """_audit — in-memory audit recording."""
+
+    def test_audit_new_user(self):
+        from src.api.routes.assistant._schemas import ASSISTANT_AUDIT
+        from src.api.routes.assistant._history import _audit
+
+        ASSISTANT_AUDIT.clear()
+        _audit("user-1", {"decision": "executed", "task_id": "task-42"})
+        assert "user-1" in ASSISTANT_AUDIT
+        assert len(ASSISTANT_AUDIT["user-1"]) == 1
+        entry = ASSISTANT_AUDIT["user-1"][0]
+        assert entry["decision"] == "executed"
+        assert "created_at" in entry
+
+    def test_audit_append(self):
+        from src.api.routes.assistant._schemas import ASSISTANT_AUDIT
+        from src.api.routes.assistant._history import _audit
+
+        ASSISTANT_AUDIT.clear()
+        _audit("user-1", {"decision": "first"})
+        _audit("user-1", {"decision": "second"})
+        assert len(ASSISTANT_AUDIT["user-1"]) == 2
+
+
+class TestPersistAudit:
+    """_persist_audit — DB audit persistence."""
+
+    def test_persist_audit_success(self):
+        from src.api.routes.assistant._history import _persist_audit
+
+        db = MagicMock()
+        _persist_audit(db, "user-1", {"decision": "executed"}, "conv-1")
+        db.add.assert_called_once()
+        db.commit.assert_called_once()
+
+    def test_persist_audit_failure(self):
+        from src.api.routes.assistant._history import _persist_audit
+
+        db = MagicMock()
+        db.commit.side_effect = Exception("DB error")
+        _persist_audit(db, "user-1", {"decision": "executed"}, "conv-1")
+        db.rollback.assert_called_once()
+
+
+class TestPersistConfirmation:
+    """_persist_confirmation — Confirmation token DB persistence."""
+
+    def test_persist_confirmation_success(self):
+        from src.api.routes.assistant._history import _persist_confirmation
+        from src.api.routes.assistant._schemas import ConfirmationRecord
+
+        db = MagicMock()
+        record = ConfirmationRecord(
+            id="conf-1",
+            user_id="user-1",
+            conversation_id="conv-1",
+            state="pending",
+            intent={"op": "test"},
+            dispatch={},
+            expires_at=datetime.now(UTC) + timedelta(minutes=5),
+            created_at=datetime.now(UTC),
+        )
+        _persist_confirmation(db, record)
+        db.merge.assert_called_once()
+        db.commit.assert_called_once()
+
+    def test_persist_confirmation_failure(self):
+        from src.api.routes.assistant._history import _persist_confirmation
+        from src.api.routes.assistant._schemas import ConfirmationRecord
+
+        db = MagicMock()
+        db.commit.side_effect = Exception("DB error")
+        record = ConfirmationRecord(
+            id="conf-1",
+            user_id="user-1",
+            conversation_id="conv-1",
+            state="pending",
+            intent={"op": "test"},
+            dispatch={},
+            expires_at=datetime.now(UTC) + timedelta(minutes=5),
+            created_at=datetime.now(UTC),
+        )
+        _persist_confirmation(db, record)
+        db.rollback.assert_called_once()
+
+
+class TestUpdateConfirmationState:
+    """_update_confirmation_state — confirmation lifecycle."""
+
+    def test_update_to_consumed(self):
+        from src.api.routes.assistant._history import _update_confirmation_state
+        from src.models.assistant import AssistantConfirmationRecord
+
+        db = MagicMock()
+        row = MagicMock(spec=AssistantConfirmationRecord)
+        row.state = "pending"
+        row.consumed_at = None
+        db.query.return_value.filter.return_value.first.return_value = row
+
+        _update_confirmation_state(db, "conf-1", "consumed")
+        assert row.state == "consumed"
+        assert row.consumed_at is not None
+        db.commit.assert_called_once()
+
+    def test_update_not_found(self):
+        from src.api.routes.assistant._history import _update_confirmation_state
+
+        db = MagicMock()
+        db.query.return_value.filter.return_value.first.return_value = None
+        # Should not raise
+        _update_confirmation_state(db, "conf-nonexistent", "consumed")
+        db.commit.assert_not_called()
+
+    def test_update_failure_rollback(self):
+        from src.api.routes.assistant._history import _update_confirmation_state
+        from src.models.assistant import AssistantConfirmationRecord
+
+        db = MagicMock()
+        row = MagicMock(spec=AssistantConfirmationRecord)
+        row.state = "pending"
+        db.query.return_value.filter.return_value.first.return_value = row
+        db.commit.side_effect = Exception("DB error")
+
+        _update_confirmation_state(db, "conf-1", "consumed")
+        db.rollback.assert_called_once()
+
+
+class TestLoadConfirmationFromDb:
+    """_load_confirmation_from_db — confirmation loading."""
+
+    def test_load_found(self):
+        from src.api.routes.assistant._history import _load_confirmation_from_db
+        from src.models.assistant import AssistantConfirmationRecord
+
+        db = MagicMock()
+        mock_row = MagicMock(spec=AssistantConfirmationRecord)
+        mock_row.id = "conf-1"
+        mock_row.user_id = "user-1"
+        mock_row.conversation_id = "conv-1"
+        mock_row.intent = {"op": "test"}
+        mock_row.dispatch = {}
+        mock_row.state = "pending"
+        mock_row.expires_at = datetime.now(UTC) + timedelta(hours=1)
+        mock_row.created_at = datetime.now(UTC)
+        db.query.return_value.filter.return_value.first.return_value = mock_row
+
+        record = _load_confirmation_from_db(db, "conf-1")
+        assert record is not None
+        assert record.id == "conf-1"
+        assert record.user_id == "user-1"
+        assert record.intent == {"op": "test"}
+
+    def test_load_not_found(self):
+        from src.api.routes.assistant._history import _load_confirmation_from_db
+
+        db = MagicMock()
+        db.query.return_value.filter.return_value.first.return_value = None
+        record = _load_confirmation_from_db(db, "conf-nonexistent")
+        assert record is None
+
+
+class TestEnsureConversation:
+    """_ensure_conversation — conversation ID resolution."""
+
+    def test_with_explicit_id(self):
+        from src.api.routes.assistant._schemas import USER_ACTIVE_CONVERSATION
+        from src.api.routes.assistant._history import _ensure_conversation
+
+        USER_ACTIVE_CONVERSATION.clear()
+        result = _ensure_conversation("user-1", "conv-1")
+        assert result == "conv-1"
+        assert USER_ACTIVE_CONVERSATION.get("user-1") == "conv-1"
+
+    def test_with_active_conversation(self):
+        from src.api.routes.assistant._schemas import USER_ACTIVE_CONVERSATION
+        from src.api.routes.assistant._history import _ensure_conversation
+
+        USER_ACTIVE_CONVERSATION.clear()
+        USER_ACTIVE_CONVERSATION["user-1"] = "existing-conv"
+        result = _ensure_conversation("user-1", None)
+        assert result == "existing-conv"
+
+    def test_creates_new(self):
+        from src.api.routes.assistant._schemas import USER_ACTIVE_CONVERSATION
+        from src.api.routes.assistant._history import _ensure_conversation
+
+        USER_ACTIVE_CONVERSATION.clear()
+        result = _ensure_conversation("user-1", None)
+        assert result is not None
+        assert uuid.UUID(result)
+        assert USER_ACTIVE_CONVERSATION.get("user-1") == result
+
+
+class TestResolveOrCreateConversation:
+    """_resolve_or_create_conversation — DB-backed conversation resolution."""
+
+    def test_with_explicit_id(self):
+        from src.api.routes.assistant._schemas import USER_ACTIVE_CONVERSATION
+        from src.api.routes.assistant._history import _resolve_or_create_conversation
+
+        USER_ACTIVE_CONVERSATION.clear()
+        db = MagicMock()
+        result = _resolve_or_create_conversation("user-1", "conv-1", db)
+        assert result == "conv-1"
+
+    def test_with_active_memory(self):
+        from src.api.routes.assistant._schemas import USER_ACTIVE_CONVERSATION
+        from src.api.routes.assistant._history import _resolve_or_create_conversation
+
+        USER_ACTIVE_CONVERSATION.clear()
+        USER_ACTIVE_CONVERSATION["user-1"] = "mem-conv"
+        db = MagicMock()
+        result = _resolve_or_create_conversation("user-1", None, db)
+        assert result == "mem-conv"
+
+    def test_creates_new_without_db_hit(self):
+        from src.api.routes.assistant._schemas import USER_ACTIVE_CONVERSATION
+        from src.api.routes.assistant._history import _resolve_or_create_conversation
+
+        USER_ACTIVE_CONVERSATION.clear()
+        db = MagicMock()
+        db.query.return_value.filter.return_value.order_by.return_value.first.return_value = None
+        result = _resolve_or_create_conversation("user-1", None, db)
+        assert result is not None
+        assert uuid.UUID(result)
+
+
+class TestCleanupHistoryTtl:
+    """_cleanup_history_ttl — TTL-based history cleanup."""
+
+    def test_cleanup_deletes_expired_db_rows(self):
+        from src.api.routes.assistant._history import _cleanup_history_ttl
+
+        db = MagicMock()
+        db.query.return_value.filter.return_value = MagicMock()
+        db.query.return_value.filter.return_value.filter.return_value.delete.return_value = 5
+
+        _cleanup_history_ttl(db, "user-1")
+        db.commit.assert_called_once()
+
+    def test_cleanup_failure_rollback(self):
+        from src.api.routes.assistant._history import _cleanup_history_ttl
+
+        db = MagicMock()
+        # The second filter (for TTL) raises
+        q = MagicMock()
+        q.filter.return_value = q  # chain .filter().filter()
+        q.delete.side_effect = Exception("DB error")
+        db.query.return_value = q
+
+        _cleanup_history_ttl(db, "user-1")
+        db.rollback.assert_called_once()
+
+    def test_cleanup_clears_stale_in_memory(self):
+        from datetime import UTC, datetime, timedelta
+        from src.api.routes.assistant._schemas import CONVERSATIONS
+        from src.api.routes.assistant._history import _cleanup_history_ttl
+
+        CONVERSATIONS.clear()
+        old_time = datetime.now() - timedelta(days=400)
+        # Add a stale entry
+        CONVERSATIONS[("user-1", "old-conv")] = [
+            {"created_at": old_time, "role": "user", "text": "old"}
+        ]
+        # Add a recent entry
+        CONVERSATIONS[("user-1", "new-conv")] = [
+            {"created_at": datetime.now(UTC), "role": "user", "text": "new"}
+        ]
+
+        db = MagicMock()
+        db.query.return_value.filter.return_value = MagicMock()
+        db.query.return_value.filter.return_value.filter.return_value.delete.return_value = 0
+        db.query.return_value.filter.return_value.filter.return_value = db.query.return_value.filter.return_value
+
+        _cleanup_history_ttl(db, "user-1")
+
+        # Old conversation should be removed
+        assert ("user-1", "old-conv") not in CONVERSATIONS
+        # New conversation should be kept
+        assert ("user-1", "new-conv") in CONVERSATIONS
+
+
+class TestIsConversationArchived:
+    """_is_conversation_archived — archive detection."""
+
+    def test_updated_at_none(self):
+        from src.api.routes.assistant._history import _is_conversation_archived
+        assert _is_conversation_archived(None) is False
+
+    def test_recent_conversation(self):
+        from datetime import datetime
+        from src.api.routes.assistant._history import _is_conversation_archived
+        recent = datetime.now()
+        assert _is_conversation_archived(recent) is False
+
+    def test_old_conversation(self):
+        from datetime import datetime, timedelta
+        from src.api.routes.assistant._history import _is_conversation_archived
+        old = datetime.now() - timedelta(days=100)
+        assert _is_conversation_archived(old) is True
+
+    def test_aware_datetime(self):
+        from datetime import UTC, datetime, timedelta
+        from src.api.routes.assistant._history import _is_conversation_archived
+        old = datetime.now(UTC) - timedelta(days=100)
+        assert _is_conversation_archived(old) is True
+
+
+class TestCoerceQueryBool:
+    """_coerce_query_bool — boolean normalization."""
+
+    def test_bool_values(self):
+        from src.api.routes.assistant._history import _coerce_query_bool
+        assert _coerce_query_bool(True) is True
+        assert _coerce_query_bool(False) is False
+
+    @pytest.mark.parametrize("val,expected", [
+        ("1", True),
+        ("true", True),
+        ("yes", True),
+        ("on", True),
+        ("0", False),
+        ("false", False),
+        ("no", False),
+        ("off", False),
+        ("", False),
+    ])
+    def test_string_values(self, val, expected):
+        from src.api.routes.assistant._history import _coerce_query_bool
+        assert _coerce_query_bool(val) is expected
+
+    def test_non_bool_non_string(self):
+        from src.api.routes.assistant._history import _coerce_query_bool
+        assert _coerce_query_bool(None) is False
+        assert _coerce_query_bool(42) is False
+        assert _coerce_query_bool([]) is False
+# #endregion Test.Assistant.History
diff --git a/backend/tests/api/test_assistant_llm_planner_intent.py b/backend/tests/api/test_assistant_llm_planner_intent.py
new file mode 100644
index 00000000..869bd8de
--- /dev/null
+++ b/backend/tests/api/test_assistant_llm_planner_intent.py
@@ -0,0 +1,356 @@
+# #region Test.Assistant.LlmPlannerIntent [C:2] [TYPE Module] [SEMANTICS test,assistant,llm,planner,intent,authorization]
+# @BRIEF Tests for _llm_planner_intent.py — LLM-based intent planning and authorization.
+# @RELATION BINDS_TO -> [AssistantLlmPlannerIntent]
+# @TEST_EDGE: plan_intent_empty_tools -> Returns None when tools list empty
+# @TEST_EDGE: plan_intent_no_provider -> Returns None when no LLM provider
+# @TEST_EDGE: plan_intent_llm_failure -> Returns None on LLM error
+# @TEST_EDGE: plan_intent_clarify -> Returns clarify intent for ambiguous input
+# @TEST_EDGE: plan_intent_valid_operation -> Valid operation mapped correctly
+# @TEST_EDGE: plan_intent_production_env -> Production env sets dangerous risk
+# @TEST_EDGE: authorize_intent -> Permission checks pass or raise 403
+
+from pathlib import Path
+import sys
+
+sys.path.insert(0, str(Path(__file__).parent.parent / "src"))
+
+from unittest.mock import AsyncMock, MagicMock, patch
+import pytest
+from fastapi import HTTPException
+
+
+class TestPlanIntentWithLlm:
+    """_plan_intent_with_llm — LLM-based intent planning."""
+
+    @pytest.fixture
+    def db(self):
+        return MagicMock()
+
+    @pytest.fixture
+    def config_manager(self):
+        cm = MagicMock()
+        cm.get_config.return_value.settings.llm = {
+            "assistant_planner_provider": "openai",
+            "assistant_planner_model": "gpt-4",
+        }
+        cm.get_environments.return_value = []
+        return cm
+
+    def test_empty_tools_returns_none(self, db, config_manager):
+        from src.api.routes.assistant._llm_planner_intent import _plan_intent_with_llm
+
+        result = __import__("asyncio").run(_plan_intent_with_llm("test", [], db, config_manager))
+        assert result is None
+
+    def test_no_provider_returns_none(self, db, config_manager):
+        from src.api.routes.assistant._llm_planner_intent import _plan_intent_with_llm
+
+        tools = [{"operation": "run_backup", "domain": "backup"}]
+        with patch("src.api.routes.assistant._llm_planner_intent.LLMProviderService") as mock_svc:
+            mock_svc.return_value.get_all_providers.return_value = []
+            result = __import__("asyncio").run(
+                _plan_intent_with_llm("test", tools, db, config_manager)
+            )
+            assert result is None
+
+    def test_no_api_key_returns_none(self, db, config_manager):
+        from src.api.routes.assistant._llm_planner_intent import _plan_intent_with_llm
+
+        tools = [{"operation": "run_backup", "domain": "backup"}]
+        provider_mock = MagicMock()
+        provider_mock.id = "p-1"
+        provider_mock.provider_type = "openai"
+        provider_mock.base_url = None
+        provider_mock.default_model = "gpt-4"
+
+        with patch("src.api.routes.assistant._llm_planner_intent.LLMProviderService") as mock_svc_cls:
+            mock_svc = MagicMock()
+            mock_svc_cls.return_value = mock_svc
+            mock_svc.get_all_providers.return_value = [provider_mock]
+            mock_svc.get_decrypted_api_key.return_value = None
+
+            with patch("src.api.routes.assistant._llm_planner_intent._resolve_provider_id") as mock_resolve:
+                mock_resolve.return_value = "p-1"
+                result = __import__("asyncio").run(
+                    _plan_intent_with_llm("test", tools, db, config_manager)
+                )
+                assert result is None
+
+    def test_llm_failure_returns_none(self, db, config_manager):
+        from src.api.routes.assistant._llm_planner_intent import _plan_intent_with_llm
+
+        tools = [{"operation": "run_backup", "domain": "backup"}]
+        provider_mock = MagicMock()
+        provider_mock.id = "p-1"
+        provider_mock.provider_type = "openai"
+        provider_mock.base_url = None
+        provider_mock.default_model = "gpt-4"
+
+        with patch("src.api.routes.assistant._llm_planner_intent.LLMProviderService") as mock_svc_cls, \
+             patch("src.api.routes.assistant._llm_planner_intent.LLMClient") as mock_client_cls:
+            mock_svc = MagicMock()
+            mock_svc_cls.return_value = mock_svc
+            mock_svc.get_all_providers.return_value = [provider_mock]
+            mock_svc.get_decrypted_api_key.return_value = "sk-test"
+
+            mock_client = AsyncMock()
+            mock_client_cls.return_value = mock_client
+            mock_client.get_json_completion.side_effect = Exception("LLM unavailable")
+
+            with patch("src.api.routes.assistant._llm_planner_intent._resolve_provider_id") as mock_resolve:
+                mock_resolve.return_value = "p-1"
+                result = __import__("asyncio").run(
+                    _plan_intent_with_llm("test", tools, db, config_manager)
+                )
+                assert result is None
+
+    def test_clarify_response(self, db, config_manager):
+        from src.api.routes.assistant._llm_planner_intent import _plan_intent_with_llm
+
+        tools = [{"operation": "run_backup", "domain": "backup"}]
+        provider_mock = MagicMock()
+        provider_mock.id = "p-1"
+        provider_mock.provider_type = "openai"
+        provider_mock.base_url = None
+        provider_mock.default_model = "gpt-4"
+
+        with patch("src.api.routes.assistant._llm_planner_intent.LLMProviderService") as mock_svc_cls, \
+             patch("src.api.routes.assistant._llm_planner_intent.LLMClient") as mock_client_cls:
+            mock_svc = MagicMock()
+            mock_svc_cls.return_value = mock_svc
+            mock_svc.get_all_providers.return_value = [provider_mock]
+            mock_svc.get_decrypted_api_key.return_value = "sk-test"
+
+            mock_client = AsyncMock()
+            mock_client_cls.return_value = mock_client
+            mock_client.get_json_completion.return_value = {
+                "operation": "clarify",
+                "domain": "unknown",
+                "confidence": 0.3,
+                "clarification": "Что именно вы хотите сделать?",
+                "entities": {},
+            }
+
+            with patch("src.api.routes.assistant._llm_planner_intent._resolve_provider_id") as mock_resolve:
+                mock_resolve.return_value = "p-1"
+                result = __import__("asyncio").run(
+                    _plan_intent_with_llm("test", tools, db, config_manager)
+                )
+                assert result is not None
+                assert result["operation"] == "clarify"
+                assert "Что именно" in result.get("clarification", "")
+
+    def test_invalid_operation_returns_none(self, db, config_manager):
+        from src.api.routes.assistant._llm_planner_intent import _plan_intent_with_llm
+
+        tools = [{"operation": "run_backup", "domain": "backup"}]
+        provider_mock = MagicMock()
+        provider_mock.id = "p-1"
+        provider_mock.provider_type = "openai"
+        provider_mock.base_url = None
+        provider_mock.default_model = "gpt-4"
+
+        with patch("src.api.routes.assistant._llm_planner_intent.LLMProviderService") as mock_svc_cls, \
+             patch("src.api.routes.assistant._llm_planner_intent.LLMClient") as mock_client_cls:
+            mock_svc = MagicMock()
+            mock_svc_cls.return_value = mock_svc
+            mock_svc.get_all_providers.return_value = [provider_mock]
+            mock_svc.get_decrypted_api_key.return_value = "sk-test"
+
+            mock_client = AsyncMock()
+            mock_client_cls.return_value = mock_client
+            mock_client.get_json_completion.return_value = {
+                "operation": "nonexistent_op",
+                "domain": "backup",
+                "confidence": 0.9,
+                "entities": {},
+            }
+
+            with patch("src.api.routes.assistant._llm_planner_intent._resolve_provider_id") as mock_resolve:
+                mock_resolve.return_value = "p-1"
+                result = __import__("asyncio").run(
+                    _plan_intent_with_llm("test", tools, db, config_manager)
+                )
+                assert result is None
+
+    def test_non_dict_response_returns_none(self, db, config_manager):
+        from src.api.routes.assistant._llm_planner_intent import _plan_intent_with_llm
+
+        tools = [{"operation": "run_backup", "domain": "backup"}]
+        provider_mock = MagicMock()
+        provider_mock.id = "p-1"
+        provider_mock.provider_type = "openai"
+        provider_mock.base_url = None
+        provider_mock.default_model = "gpt-4"
+
+        with patch("src.api.routes.assistant._llm_planner_intent.LLMProviderService") as mock_svc_cls, \
+             patch("src.api.routes.assistant._llm_planner_intent.LLMClient") as mock_client_cls:
+            mock_svc = MagicMock()
+            mock_svc_cls.return_value = mock_svc
+            mock_svc.get_all_providers.return_value = [provider_mock]
+            mock_svc.get_decrypted_api_key.return_value = "sk-test"
+
+            mock_client = AsyncMock()
+            mock_client_cls.return_value = mock_client
+            mock_client.get_json_completion.return_value = "not a dict"
+
+            with patch("src.api.routes.assistant._llm_planner_intent._resolve_provider_id") as mock_resolve:
+                mock_resolve.return_value = "p-1"
+                result = __import__("asyncio").run(
+                    _plan_intent_with_llm("test", tools, db, config_manager)
+                )
+                assert result is None
+
+    def test_valid_operation_with_defaults(self, db, config_manager):
+        from src.api.routes.assistant._llm_planner_intent import _plan_intent_with_llm
+
+        tools = [{
+            "operation": "run_backup",
+            "domain": "backup",
+            "risk_level": "guarded",
+            "requires_confirmation": False,
+            "defaults": {"environment": "dev"},
+        }]
+        provider_mock = MagicMock()
+        provider_mock.id = "p-1"
+        provider_mock.provider_type = "openai"
+        provider_mock.base_url = None
+        provider_mock.default_model = "gpt-4"
+
+        with patch("src.api.routes.assistant._llm_planner_intent.LLMProviderService") as mock_svc_cls, \
+             patch("src.api.routes.assistant._llm_planner_intent.LLMClient") as mock_client_cls:
+            mock_svc = MagicMock()
+            mock_svc_cls.return_value = mock_svc
+            mock_svc.get_all_providers.return_value = [provider_mock]
+            mock_svc.get_decrypted_api_key.return_value = "sk-test"
+
+            mock_client = AsyncMock()
+            mock_client_cls.return_value = mock_client
+            mock_client.get_json_completion.return_value = {
+                "operation": "run_backup",
+                "domain": "backup",
+                "confidence": 0.9,
+                "entities": {},
+                "risk_level": "guarded",
+                "requires_confirmation": False,
+            }
+
+            with patch("src.api.routes.assistant._llm_planner_intent._resolve_provider_id") as mock_resolve, \
+                 patch("src.api.routes.assistant._llm_planner_intent._coerce_intent_entities") as mock_coerce, \
+                 patch("src.api.routes.assistant._llm_planner_intent.normalize_llm_settings") as mock_norm:
+                mock_resolve.return_value = "p-1"
+                mock_coerce.return_value = {
+                    "domain": "backup",
+                    "operation": "run_backup",
+                    "entities": {},
+                    "confidence": 0.9,
+                    "risk_level": "guarded",
+                    "requires_confirmation": False,
+                }
+                mock_norm.return_value = {
+                    "assistant_planner_provider": "openai",
+                    "assistant_planner_model": "gpt-4",
+                }
+                result = __import__("asyncio").run(
+                    _plan_intent_with_llm("test", tools, db, config_manager)
+                )
+                assert result is not None
+                assert result["operation"] == "run_backup"
+                # Default should be applied
+                assert result["entities"].get("environment") == "dev"
+
+    def test_production_deploy_sets_dangerous(self, db, config_manager):
+        from src.api.routes.assistant._llm_planner_intent import _plan_intent_with_llm
+
+        # Environments must be JSON-serializable dicts
+        config_manager.get_environments.return_value = [
+            type('Env', (), {'id': 'prod', 'name': 'PROD'})(),
+        ]
+        tools = [{
+            "operation": "deploy_dashboard",
+            "domain": "git",
+            "risk_level": "guarded",
+            "requires_confirmation": False,
+        }]
+        provider_mock = MagicMock()
+        provider_mock.id = "p-1"
+        provider_mock.provider_type = "openai"
+        provider_mock.base_url = None
+        provider_mock.default_model = "gpt-4"
+
+        with patch("src.api.routes.assistant._llm_planner_intent.LLMProviderService") as mock_svc_cls, \
+             patch("src.api.routes.assistant._llm_planner_intent.LLMClient") as mock_client_cls:
+            mock_svc = MagicMock()
+            mock_svc_cls.return_value = mock_svc
+            mock_svc.get_all_providers.return_value = [provider_mock]
+            mock_svc.get_decrypted_api_key.return_value = "sk-test"
+
+            mock_client = AsyncMock()
+            mock_client_cls.return_value = mock_client
+            mock_client.get_json_completion.return_value = {
+                "operation": "deploy_dashboard",
+                "domain": "git",
+                "confidence": 0.9,
+                "entities": {"environment": "prod"},
+                "risk_level": "guarded",
+                "requires_confirmation": False,
+            }
+
+            with patch("src.api.routes.assistant._llm_planner_intent._resolve_provider_id") as mock_resolve, \
+                 patch("src.api.routes.assistant._llm_planner_intent._coerce_intent_entities") as mock_coerce, \
+                 patch("src.api.routes.assistant._llm_planner_intent._is_production_env") as mock_is_prod, \
+                 patch("src.api.routes.assistant._llm_planner_intent.normalize_llm_settings") as mock_norm:
+                mock_resolve.return_value = "p-1"
+                mock_coerce.return_value = {
+                    "domain": "git",
+                    "operation": "deploy_dashboard",
+                    "entities": {"environment": "prod"},
+                    "confidence": 0.9,
+                    "risk_level": "guarded",
+                    "requires_confirmation": False,
+                }
+                mock_is_prod.return_value = True
+                mock_norm.return_value = {
+                    "assistant_planner_provider": "openai",
+                    "assistant_planner_model": "gpt-4",
+                }
+                result = __import__("asyncio").run(
+                    _plan_intent_with_llm("test", tools, db, config_manager)
+                )
+                assert result is not None
+                assert result["risk_level"] == "dangerous"
+                assert result["requires_confirmation"] is True
+
+
+class TestAuthorizeIntent:
+    """_authorize_intent — permission checks."""
+
+    def test_authorized(self):
+        from src.api.routes.assistant._llm_planner_intent import _authorize_intent
+        from src.schemas.auth import User
+        from datetime import UTC, datetime
+
+        user = User(id="user-1", username="test", email="test@x.com", is_active=True, auth_source="internal",
+                    created_at=datetime.now(UTC), last_login=datetime.now(UTC), roles=[])
+        intent = {"operation": "run_backup"}
+
+        with patch("src.api.routes.assistant._llm_planner_intent.get_permission_checks") as mock_checks, \
+             patch("src.api.routes.assistant._llm_planner_intent._check_any_permission") as mock_check:
+            mock_checks.return_value = {"run_backup": [("backup", "WRITE")]}
+            _authorize_intent(intent, user)
+            mock_check.assert_called_once_with(user, [("backup", "WRITE")])
+
+    def test_no_checks_skips(self):
+        from src.api.routes.assistant._llm_planner_intent import _authorize_intent
+        from src.schemas.auth import User
+        from datetime import UTC, datetime
+
+        user = User(id="user-1", username="test", email="test@x.com", is_active=True, auth_source="internal",
+                    created_at=datetime.now(UTC), last_login=datetime.now(UTC), roles=[])
+        intent = {"operation": "show_capabilities"}
+
+        with patch("src.api.routes.assistant._llm_planner_intent.get_permission_checks") as mock_checks:
+            mock_checks.return_value = {}  # no checks for this op
+            # Should not raise
+            _authorize_intent(intent, user)
+# #endregion Test.Assistant.LlmPlannerIntent
diff --git a/backend/tests/api/test_assistant_tool_maintenance.py b/backend/tests/api/test_assistant_tool_maintenance.py
new file mode 100644
index 00000000..0116b741
--- /dev/null
+++ b/backend/tests/api/test_assistant_tool_maintenance.py
@@ -0,0 +1,384 @@
+# #region Test.Assistant.ToolMaintenance [C:2] [TYPE Module] [SEMANTICS test,assistant,maintenance,events]
+# @BRIEF Tests for _tool_maintenance.py — maintenance handlers.
+# @RELATION BINDS_TO -> [AssistantToolMaintenance]
+# @TEST_EDGE: list_maintenance_events -> Lists active and completed events
+# @TEST_EDGE: list_maintenance_no_events -> Empty maintenance response
+# @TEST_EDGE: start_maintenance -> Start with tables and start_time
+# @TEST_EDGE: start_maintenance_missing_params -> 422 on missing params
+# @TEST_EDGE: start_maintenance_invalid_time -> 422 on bad datetime
+# @TEST_EDGE: start_maintenance_too_many_tables -> 422 on >100 tables
+# @TEST_EDGE: end_maintenance_by_id -> End specific event
+# @TEST_EDGE: end_maintenance_not_found -> 404 on missing event
+# @TEST_EDGE: end_maintenance_all_active -> End all active events
+
+from pathlib import Path
+import sys
+
+sys.path.insert(0, str(Path(__file__).parent.parent / "src"))
+
+from datetime import UTC, datetime, timedelta
+from unittest.mock import AsyncMock, MagicMock, patch
+import pytest
+from fastapi import HTTPException
+
+
+def _make_maintenance_event(event_id="evt-1", status="ACTIVE", tables=None, environment_id="env-1"):
+    ev = MagicMock()
+    ev.id = event_id
+    ev.status = status
+    ev.tables = tables or ["users"]
+    ev.environment_id = environment_id
+    ev.created_at = datetime.now(UTC)
+    ev.end_time = None
+    return ev
+
+
+class TestHandleListMaintenanceEvents:
+    """handle_list_maintenance_events — listing events."""
+
+    @pytest.fixture
+    def intent(self):
+        return {"entities": {}}
+
+    @pytest.fixture
+    def current_user(self):
+        from src.schemas.auth import User
+        return User(id="user-1", username="test", email="test@x.com", is_active=True,
+                    auth_source="internal", created_at=datetime.now(UTC), last_login=datetime.now(UTC), roles=[])
+
+    @pytest.mark.asyncio
+    async def test_lists_active_and_completed(self, intent, current_user):
+        from src.api.routes.assistant._tool_maintenance import handle_list_maintenance_events
+        from src.models.maintenance import MaintenanceEventStatus
+
+        task_manager = MagicMock()
+        task_manager.create_task = AsyncMock(return_value=MagicMock(id="task-99"))
+        config_manager = MagicMock()
+        db = MagicMock()
+
+        active_events = [_make_maintenance_event("evt-1", MaintenanceEventStatus.ACTIVE)]
+        completed_events = [_make_maintenance_event("evt-2", MaintenanceEventStatus.COMPLETED)]
+
+        # Mock the queries
+        db.query.return_value.filter.return_value = MagicMock()
+        db.query.return_value.filter.return_value.filter.return_value.all.return_value = []
+
+        # Separate mocks for each query
+        def query_side_effect(model):
+            q = MagicMock()
+            if model.__name__ == "MaintenanceEvent":
+                # Active events
+                q.filter.return_value = MagicMock()
+                q.filter.return_value.filter.return_value.all.side_effect = [
+                    [],  # expired
+                    active_events,  # active
+                ]
+                q.filter.return_value.filter.return_value.order_by.return_value.limit.return_value.all.return_value = completed_events
+                q.filter.return_value.filter.return_value.order_by.return_value.all.return_value = active_events
+                q.filter.return_value.in_.return_value = q.filter.return_value
+                q.filter.return_value.isnot.return_value = q.filter.return_value
+            return q
+
+        db.query.side_effect = query_side_effect
+
+        # Mock dashboard state count
+        db.query.return_value.filter.return_value.filter.return_value = MagicMock()
+        db.query.return_value.filter.return_value.filter.return_value.count.return_value = 3
+
+        text, task_id, actions = await handle_list_maintenance_events(
+            intent, current_user, task_manager, config_manager, db
+        )
+        assert "Активные" in text or "Завершённые" in text
+        assert actions is not None
+
+    @pytest.mark.asyncio
+    async def test_no_events(self, intent, current_user):
+        from src.api.routes.assistant._tool_maintenance import handle_list_maintenance_events
+
+        task_manager = MagicMock()
+        task_manager.create_task = AsyncMock(return_value=MagicMock(id="task-99"))
+        config_manager = MagicMock()
+        db = MagicMock()
+
+        def query_side_effect(model):
+            q = MagicMock()
+            if model.__name__ == "MaintenanceEvent":
+                q.filter.return_value = MagicMock()
+                q.filter.return_value.filter.return_value.all.return_value = []
+                q.filter.return_value.filter.return_value.order_by.return_value.all.return_value = []
+                q.filter.return_value.filter.return_value.order_by.return_value.limit.return_value.all.return_value = []
+            else:
+                q.filter.return_value.filter.return_value.count.return_value = 0
+            return q
+
+        db.query.side_effect = query_side_effect
+
+        text, task_id, actions = await handle_list_maintenance_events(
+            intent, current_user, task_manager, config_manager, db
+        )
+        assert "Нет событий" in text
+
+
+class TestHandleStartMaintenance:
+    """handle_start_maintenance — starting maintenance events."""
+
+    @pytest.fixture
+    def intent(self):
+        return {"entities": {}}
+
+    @pytest.fixture
+    def current_user(self):
+        from src.schemas.auth import User
+        return User(id="user-1", username="test", email="test@x.com", is_active=True,
+                    auth_source="internal", created_at=datetime.now(UTC), last_login=datetime.now(UTC), roles=[])
+
+    @pytest.mark.asyncio
+    async def test_start_success(self, intent, current_user):
+        from src.api.routes.assistant._tool_maintenance import handle_start_maintenance
+
+        task_manager = MagicMock()
+        task_manager.create_task = AsyncMock(return_value=MagicMock(id="task-42"))
+        config_manager = MagicMock()
+        db = MagicMock()
+
+        intent["entities"] = {
+            "tables": ["users", "orders"],
+            "start_time": "2025-06-15T10:00",
+            "end_time": "2025-06-15T12:00",
+            "message": "Scheduled maintenance",
+        }
+
+        text, task_id, actions = await handle_start_maintenance(
+            intent, current_user, task_manager, config_manager, db
+        )
+        assert "Обслуживание запущено" in text
+        assert task_id == "task-42"
+
+    @pytest.mark.asyncio
+    async def test_start_without_end_time(self, intent, current_user):
+        from src.api.routes.assistant._tool_maintenance import handle_start_maintenance
+
+        task_manager = MagicMock()
+        task_manager.create_task = AsyncMock(return_value=MagicMock(id="task-43"))
+        config_manager = MagicMock()
+        db = MagicMock()
+
+        intent["entities"] = {
+            "tables": ["users"],
+            "start_time": "2025-06-15T10:00",
+        }
+
+        text, task_id, actions = await handle_start_maintenance(
+            intent, current_user, task_manager, config_manager, db
+        )
+        assert "Обслуживание запущено" in text
+
+    @pytest.mark.asyncio
+    async def test_start_missing_params_raises_422(self, intent, current_user):
+        from src.api.routes.assistant._tool_maintenance import handle_start_maintenance
+
+        task_manager = MagicMock()
+        config_manager = MagicMock()
+        db = MagicMock()
+
+        intent["entities"] = {}
+
+        with pytest.raises(HTTPException) as exc:
+            await handle_start_maintenance(
+                intent, current_user, task_manager, config_manager, db
+            )
+        assert exc.value.status_code == 422
+
+    @pytest.mark.asyncio
+    async def test_start_missing_tables_raises_422(self, intent, current_user):
+        from src.api.routes.assistant._tool_maintenance import handle_start_maintenance
+
+        task_manager = MagicMock()
+        config_manager = MagicMock()
+        db = MagicMock()
+
+        intent["entities"] = {"start_time": "2025-06-15T10:00"}
+
+        with pytest.raises(HTTPException) as exc:
+            await handle_start_maintenance(
+                intent, current_user, task_manager, config_manager, db
+            )
+        assert exc.value.status_code == 422
+
+    @pytest.mark.asyncio
+    async def test_start_invalid_time_raises_422(self, intent, current_user):
+        from src.api.routes.assistant._tool_maintenance import handle_start_maintenance
+
+        task_manager = MagicMock()
+        config_manager = MagicMock()
+        db = MagicMock()
+
+        intent["entities"] = {
+            "tables": ["users"],
+            "start_time": "not-a-date",
+        }
+
+        with pytest.raises(HTTPException) as exc:
+            await handle_start_maintenance(
+                intent, current_user, task_manager, config_manager, db
+            )
+        assert exc.value.status_code == 422
+        assert "start_time" in exc.value.detail.lower()
+
+    @pytest.mark.asyncio
+    async def test_start_invalid_end_time_raises_422(self, intent, current_user):
+        from src.api.routes.assistant._tool_maintenance import handle_start_maintenance
+
+        task_manager = MagicMock()
+        config_manager = MagicMock()
+        db = MagicMock()
+
+        intent["entities"] = {
+            "tables": ["users"],
+            "start_time": "2025-06-15T10:00",
+            "end_time": "not-a-date",
+        }
+
+        with pytest.raises(HTTPException) as exc:
+            await handle_start_maintenance(
+                intent, current_user, task_manager, config_manager, db
+            )
+        assert exc.value.status_code == 422
+        assert "end_time" in exc.value.detail.lower()
+
+    @pytest.mark.asyncio
+    async def test_start_too_many_tables_raises_422(self, intent, current_user):
+        from src.api.routes.assistant._tool_maintenance import handle_start_maintenance
+
+        task_manager = MagicMock()
+        config_manager = MagicMock()
+        db = MagicMock()
+
+        intent["entities"] = {
+            "tables": [f"table_{i}" for i in range(101)],
+            "start_time": "2025-06-15T10:00",
+        }
+
+        with pytest.raises(HTTPException) as exc:
+            await handle_start_maintenance(
+                intent, current_user, task_manager, config_manager, db
+            )
+        assert exc.value.status_code == 422
+        assert "100" in exc.value.detail
+
+    @pytest.mark.asyncio
+    async def test_start_with_tables_as_string(self, intent, current_user):
+        from src.api.routes.assistant._tool_maintenance import handle_start_maintenance
+
+        task_manager = MagicMock()
+        task_manager.create_task = AsyncMock(return_value=MagicMock(id="task-44"))
+        config_manager = MagicMock()
+        db = MagicMock()
+
+        intent["entities"] = {
+            "tables": "users,orders,products",
+            "start_time": "2025-06-15T10:00",
+        }
+
+        text, task_id, actions = await handle_start_maintenance(
+            intent, current_user, task_manager, config_manager, db
+        )
+        assert "Обслуживание запущено" in text
+        # Should have split the string
+        task_params = task_manager.create_task.call_args[1]["params"]
+        assert task_params["tables"] == ["users", "orders", "products"]
+
+
+class TestHandleEndMaintenance:
+    """handle_end_maintenance — ending maintenance events."""
+
+    @pytest.fixture
+    def intent(self):
+        return {"entities": {}}
+
+    @pytest.fixture
+    def current_user(self):
+        from src.schemas.auth import User
+        return User(id="user-1", username="test", email="test@x.com", is_active=True,
+                    auth_source="internal", created_at=datetime.now(UTC), last_login=datetime.now(UTC), roles=[])
+
+    @pytest.mark.asyncio
+    async def test_end_by_id(self, intent, current_user):
+        from src.api.routes.assistant._tool_maintenance import handle_end_maintenance
+
+        task_manager = MagicMock()
+        task_manager.create_task = AsyncMock(return_value=MagicMock(id="task-45"))
+        config_manager = MagicMock()
+        db = MagicMock()
+
+        event = _make_maintenance_event("evt-1")
+        db.query.return_value.filter.return_value.first.return_value = event
+
+        intent["entities"] = {"event_id": "evt-1"}
+
+        text, task_id, actions = await handle_end_maintenance(
+            intent, current_user, task_manager, config_manager, db
+        )
+        assert "Завершение" in text
+        assert task_id == "task-45"
+
+    @pytest.mark.asyncio
+    async def test_end_not_found_raises_404(self, intent, current_user):
+        from src.api.routes.assistant._tool_maintenance import handle_end_maintenance
+
+        task_manager = MagicMock()
+        config_manager = MagicMock()
+        db = MagicMock()
+
+        db.query.return_value.filter.return_value.first.return_value = None
+
+        intent["entities"] = {"event_id": "evt-nonexistent"}
+
+        with pytest.raises(HTTPException) as exc:
+            await handle_end_maintenance(
+                intent, current_user, task_manager, config_manager, db
+            )
+        assert exc.value.status_code == 404
+
+    @pytest.mark.asyncio
+    async def test_end_all_active_success(self, intent, current_user):
+        from src.api.routes.assistant._tool_maintenance import handle_end_maintenance
+        from src.models.maintenance import MaintenanceEventStatus
+
+        task_manager = MagicMock()
+        task_manager.create_task = AsyncMock(return_value=MagicMock(id="task-46"))
+        config_manager = MagicMock()
+        db = MagicMock()
+
+        active_events = [
+            _make_maintenance_event("evt-1", MaintenanceEventStatus.ACTIVE),
+            _make_maintenance_event("evt-2", MaintenanceEventStatus.PARTIAL),
+        ]
+        db.query.return_value.filter.return_value.filter.return_value.all.return_value = active_events
+
+        intent["entities"] = {}  # No event_id -> end all
+
+        text, task_id, actions = await handle_end_maintenance(
+            intent, current_user, task_manager, config_manager, db
+        )
+        assert "Завершение" in text
+        assert "2" in text  # Should mention 2 events
+
+    @pytest.mark.asyncio
+    async def test_end_all_no_active(self, intent, current_user):
+        from src.api.routes.assistant._tool_maintenance import handle_end_maintenance
+
+        task_manager = MagicMock()
+        config_manager = MagicMock()
+        db = MagicMock()
+
+        db.query.return_value.filter.return_value.filter.return_value.all.return_value = []
+
+        intent["entities"] = {}  # No event_id
+
+        text, task_id, actions = await handle_end_maintenance(
+            intent, current_user, task_manager, config_manager, db
+        )
+        assert "Нет активных" in text
+        assert task_id is None
+# #endregion Test.Assistant.ToolMaintenance
diff --git a/backend/tests/services/dataset_review/test_event_logger.py b/backend/tests/services/dataset_review/test_event_logger.py
new file mode 100644
index 00000000..58aa3cb3
--- /dev/null
+++ b/backend/tests/services/dataset_review/test_event_logger.py
@@ -0,0 +1,218 @@
+# #region Test.SessionEventLogger [C:2] [TYPE Module] [SEMANTICS test,dataset,event,logger,session,audit]
+# @BRIEF Tests for event_logger.py — session event logging and persistence.
+# @RELATION BINDS_TO -> [SessionEventLoggerModule]
+# @TEST_EDGE: log_event -> Persists event payload correctly
+# @TEST_EDGE: log_event_validation -> Empty fields raise ValueError
+# @TEST_EDGE: log_event_with_details -> Event details stored
+# @TEST_EDGE: log_for_session -> Convenience wrapper from session object
+
+from pathlib import Path
+import sys
+
+sys.path.insert(0, str(Path(__file__).parent.parent.parent / "src"))
+
+from unittest.mock import MagicMock, patch
+import pytest
+
+
+class TestSessionEventPayload:
+    """SessionEventPayload — dataclass contract."""
+
+    def test_payload_creation(self):
+        from src.services.dataset_review.event_logger import SessionEventPayload
+
+        payload = SessionEventPayload(
+            session_id="sess-1",
+            actor_user_id="user-1",
+            event_type="session_started",
+            event_summary="Session created",
+            current_phase="RECOVERY",
+            readiness_state="IMPORTING",
+            event_details={"key": "val"},
+        )
+        assert payload.session_id == "sess-1"
+        assert payload.event_type == "session_started"
+        assert payload.event_details == {"key": "val"}
+
+
+class TestSessionEventLoggerLogEvent:
+    """SessionEventLogger.log_event — persist event."""
+
+    @pytest.fixture
+    def db(self):
+        return MagicMock()
+
+    @pytest.fixture
+    def logger(self, db):
+        from src.services.dataset_review.event_logger import SessionEventLogger
+        return SessionEventLogger(db)
+
+    def test_log_event_success(self, logger, db):
+        from src.services.dataset_review.event_logger import SessionEventPayload
+
+        # Mock SessionEvent
+        mock_event = MagicMock()
+        mock_event.session_event_id = "evt-1"
+        mock_event.event_type = "session_started"
+        db.add.return_value = None
+        db.refresh.return_value = None
+
+        payload = SessionEventPayload(
+            session_id="sess-1",
+            actor_user_id="user-1",
+            event_type="session_started",
+            event_summary="Session created",
+            current_phase="RECOVERY",
+            readiness_state="IMPORTING",
+            event_details={"source": "superset"},
+        )
+
+        result = logger.log_event(payload)
+        db.add.assert_called_once()
+        db.commit.assert_called_once()
+        db.refresh.assert_called_once()
+
+    def test_log_event_minimal(self, logger, db):
+        from src.services.dataset_review.event_logger import SessionEventPayload
+
+        payload = SessionEventPayload(
+            session_id="sess-1",
+            actor_user_id="user-1",
+            event_type="update",
+            event_summary="Updated",
+        )
+        result = logger.log_event(payload)
+        db.add.assert_called_once()
+        db.commit.assert_called_once()
+
+    def test_empty_session_id_raises(self, logger, db):
+        from src.services.dataset_review.event_logger import SessionEventPayload
+
+        payload = SessionEventPayload(
+            session_id="",
+            actor_user_id="user-1",
+            event_type="test",
+            event_summary="test",
+        )
+        with pytest.raises(ValueError, match="session_id"):
+            logger.log_event(payload)
+        db.add.assert_not_called()
+
+    def test_empty_actor_user_id_raises(self, logger, db):
+        from src.services.dataset_review.event_logger import SessionEventPayload
+
+        payload = SessionEventPayload(
+            session_id="sess-1",
+            actor_user_id="",
+            event_type="test",
+            event_summary="test",
+        )
+        with pytest.raises(ValueError, match="actor_user_id"):
+            logger.log_event(payload)
+        db.add.assert_not_called()
+
+    def test_empty_event_type_raises(self, logger, db):
+        from src.services.dataset_review.event_logger import SessionEventPayload
+
+        payload = SessionEventPayload(
+            session_id="sess-1",
+            actor_user_id="user-1",
+            event_type="",
+            event_summary="test",
+        )
+        with pytest.raises(ValueError, match="event_type"):
+            logger.log_event(payload)
+        db.add.assert_not_called()
+
+    def test_empty_event_summary_raises(self, logger, db):
+        from src.services.dataset_review.event_logger import SessionEventPayload
+
+        payload = SessionEventPayload(
+            session_id="sess-1",
+            actor_user_id="user-1",
+            event_type="test",
+            event_summary="",
+        )
+        with pytest.raises(ValueError, match="event_summary"):
+            logger.log_event(payload)
+        db.add.assert_not_called()
+
+    def test_event_with_complex_details(self, logger, db):
+        from src.services.dataset_review.event_logger import SessionEventPayload
+
+        payload = SessionEventPayload(
+            session_id="sess-1",
+            actor_user_id="user-1",
+            event_type="mapping_approved",
+            event_summary="Mappings approved by assistant",
+            event_details={
+                "mapping_ids": ["m-1", "m-2"],
+                "count": 2,
+                "nested": {"key": "value"},
+            },
+        )
+        result = logger.log_event(payload)
+        db.add.assert_called_once()
+        # Verify the event_details dict
+        added = db.add.call_args[0][0]
+        assert added.event_details["mapping_ids"] == ["m-1", "m-2"]
+
+
+class TestSessionEventLoggerLogForSession:
+    """SessionEventLogger.log_for_session — convenience wrapper."""
+
+    def test_log_for_session(self):
+        from src.services.dataset_review.event_logger import SessionEventLogger, SessionEventPayload
+        from src.models.dataset_review import DatasetReviewSession, SessionPhase, ReadinessState
+
+        db = MagicMock()
+
+        session = MagicMock(spec=DatasetReviewSession)
+        session.session_id = "sess-1"
+        session.current_phase = MagicMock(value="review")
+        session.readiness_state = MagicMock(value="review_ready")
+
+        logger = SessionEventLogger(db)
+
+        with patch.object(logger, "log_event") as mock_log:
+            mock_log.return_value = MagicMock(session_event_id="evt-1")
+
+            result = logger.log_for_session(
+                session,
+                actor_user_id="user-1",
+                event_type="preview_generated",
+                event_summary="Preview ready",
+                event_details={"preview_id": "pv-1"},
+            )
+            mock_log.assert_called_once()
+            payload = mock_log.call_args[0][0]
+            assert payload.session_id == "sess-1"
+            assert payload.event_type == "preview_generated"
+            assert payload.current_phase == "review"
+            assert payload.readiness_state == "review_ready"
+
+    def test_log_for_session_no_details(self):
+        from src.services.dataset_review.event_logger import SessionEventLogger
+        from src.models.dataset_review import DatasetReviewSession
+
+        db = MagicMock()
+        session = MagicMock(spec=DatasetReviewSession)
+        session.session_id = "sess-1"
+        session.current_phase = None
+        session.readiness_state = None
+
+        logger = SessionEventLogger(db)
+
+        with patch.object(logger, "log_event") as mock_log:
+            mock_log.return_value = MagicMock(session_event_id="evt-1")
+
+            logger.log_for_session(
+                session,
+                actor_user_id="user-1",
+                event_type="test",
+                event_summary="test",
+            )
+            payload = mock_log.call_args[0][0]
+            assert payload.current_phase is None
+            assert payload.readiness_state is None
+# #endregion Test.SessionEventLogger
diff --git a/backend/tests/services/dataset_review/test_semantic_resolver.py b/backend/tests/services/dataset_review/test_semantic_resolver.py
new file mode 100644
index 00000000..150fc342
--- /dev/null
+++ b/backend/tests/services/dataset_review/test_semantic_resolver.py
@@ -0,0 +1,376 @@
+# #region Test.SemanticSourceResolver [C:2] [TYPE Module] [SEMANTICS test,dataset,semantic,resolver,mapping]
+# @BRIEF Tests for semantic_resolver.py — semantic candidate resolution, ranking, and conflict detection.
+# @RELATION BINDS_TO -> [SemanticSourceResolver]
+# @TEST_EDGE: resolve_from_file -> Returns basic resolution result
+# @TEST_EDGE: resolve_from_dictionary -> Exact, fuzzy, and unresolved field matching
+# @TEST_EDGE: resolve_from_dictionary_validation -> Missing source_ref or rows raises ValueError
+# @TEST_EDGE: resolve_from_reference_dataset -> Returns basic result
+# @TEST_EDGE: rank_candidates -> Confidence ordering
+# @TEST_EDGE: detect_conflicts -> Multi-candidate detection
+# @TEST_EDGE: apply_field_decision -> Field merging
+# @TEST_EDGE: propagate_source_version_update -> Version propagation with locks
+
+from pathlib import Path
+import sys
+
+sys.path.insert(0, str(Path(__file__).parent.parent.parent / "src"))
+
+from unittest.mock import MagicMock, patch
+import pytest
+
+
+class TestSemanticSourceResolverResolveFromFile:
+    """SemanticSourceResolver.resolve_from_file — basic file resolution."""
+
+    def test_resolve_from_file(self):
+        from src.services.dataset_review.semantic_resolver import SemanticSourceResolver
+
+        resolver = SemanticSourceResolver()
+        result = resolver.resolve_from_file(
+            {"source_ref": "upload.csv"},
+            [{"field_name": "region"}],
+        )
+        assert result.source_ref == "uploaded_file"
+
+
+class TestSemanticSourceResolverResolveFromDictionary:
+    """SemanticSourceResolver.resolve_from_dictionary — dictionary resolution."""
+
+    def test_exact_match(self):
+        from src.services.dataset_review.semantic_resolver import SemanticSourceResolver
+
+        resolver = SemanticSourceResolver()
+        source_payload = {
+            "source_ref": "dict_1",
+            "rows": [
+                {"column_name": "region", "verbose_name": "Region", "description": "Geographic region"},
+                {"column_name": "product", "verbose_name": "Product", "description": "Product name"},
+            ],
+        }
+        fields = [{"field_name": "region"}]
+        result = resolver.resolve_from_dictionary(source_payload, fields)
+        assert len(result.resolved_fields) == 1
+        assert result.resolved_fields[0]["field_name"] == "region"
+        assert result.resolved_fields[0]["provenance"] == "dictionary_exact"
+        assert result.resolved_fields[0]["needs_review"] is False
+
+    def test_fuzzy_match(self):
+        from src.services.dataset_review.semantic_resolver import SemanticSourceResolver
+
+        resolver = SemanticSourceResolver()
+        source_payload = {
+            "source_ref": "dict_1",
+            "rows": [
+                {"column_name": "region_code", "verbose_name": "Region Code", "description": "Region code"},
+            ],
+        }
+        fields = [{"field_name": "region"}]
+        result = resolver.resolve_from_dictionary(source_payload, fields)
+        assert len(result.resolved_fields) == 1
+        assert result.resolved_fields[0]["provenance"] == "fuzzy_inferred"
+        assert result.resolved_fields[0]["needs_review"] is True
+
+    def test_unresolved_field(self):
+        from src.services.dataset_review.semantic_resolver import SemanticSourceResolver
+
+        resolver = SemanticSourceResolver()
+        source_payload = {
+            "source_ref": "dict_1",
+            "rows": [
+                {"column_name": "product", "verbose_name": "Product"},
+            ],
+        }
+        fields = [{"field_name": "zzz_nonexistent"}]
+        result = resolver.resolve_from_dictionary(source_payload, fields)
+        assert len(result.resolved_fields) == 1
+        assert result.resolved_fields[0]["status"] == "unresolved"
+        assert result.resolved_fields[0]["provenance"] == "unresolved"
+        assert result.unresolved_fields == ["zzz_nonexistent"]
+        assert result.partial_recovery is True
+
+    def test_missing_source_ref_raises_value_error(self):
+        from src.services.dataset_review.semantic_resolver import SemanticSourceResolver
+
+        resolver = SemanticSourceResolver()
+        with pytest.raises(ValueError, match="source_ref"):
+            resolver.resolve_from_dictionary({"rows": [{"column_name": "x"}]}, [{"field_name": "x"}])
+
+    def test_empty_rows_raises_value_error(self):
+        from src.services.dataset_review.semantic_resolver import SemanticSourceResolver
+
+        resolver = SemanticSourceResolver()
+        with pytest.raises(ValueError, match="rows"):
+            resolver.resolve_from_dictionary({"source_ref": "dict_1", "rows": []}, [{"field_name": "x"}])
+
+    def test_non_list_rows_raises_value_error(self):
+        from src.services.dataset_review.semantic_resolver import SemanticSourceResolver
+
+        resolver = SemanticSourceResolver()
+        with pytest.raises(ValueError, match="rows"):
+            resolver.resolve_from_dictionary({"source_ref": "dict_1", "rows": "not_a_list"}, [{"field_name": "x"}])
+
+    def test_locked_field_preserved(self):
+        from src.services.dataset_review.semantic_resolver import SemanticSourceResolver
+
+        resolver = SemanticSourceResolver()
+        source_payload = {
+            "source_ref": "dict_1",
+            "rows": [{"column_name": "region", "verbose_name": "Region"}],
+        }
+        fields = [{"field_name": "region", "is_locked": True}]
+        result = resolver.resolve_from_dictionary(source_payload, fields)
+        assert len(result.resolved_fields) == 1
+        assert result.resolved_fields[0]["status"] == "preserved_manual"
+        assert result.resolved_fields[0]["is_locked"] is True
+
+    def test_skip_empty_field_name(self):
+        from src.services.dataset_review.semantic_resolver import SemanticSourceResolver
+
+        resolver = SemanticSourceResolver()
+        source_payload = {
+            "source_ref": "dict_1",
+            "rows": [{"column_name": "region", "verbose_name": "Region"}],
+        }
+        fields = [{"field_name": ""}, {"field_name": "region"}]
+        result = resolver.resolve_from_dictionary(source_payload, fields)
+        assert len(result.resolved_fields) == 1
+        assert result.resolved_fields[0]["field_name"] == "region"
+
+    def test_mixed_resolved_and_unresolved(self):
+        from src.services.dataset_review.semantic_resolver import SemanticSourceResolver
+
+        resolver = SemanticSourceResolver()
+        source_payload = {
+            "source_ref": "dict_1",
+            "rows": [{"column_name": "region", "verbose_name": "Region"}],
+        }
+        fields = [
+            {"field_name": "region"},
+            {"field_name": "product"},
+            {"field_name": "sales_amount"},
+        ]
+        result = resolver.resolve_from_dictionary(source_payload, fields)
+        assert len(result.resolved_fields) == 3
+        region_field = next(f for f in result.resolved_fields if f["field_name"] == "region")
+        assert region_field["provenance"] == "dictionary_exact"
+        product_field = next(f for f in result.resolved_fields if f["field_name"] == "product")
+        assert product_field["status"] == "unresolved"
+        assert len(result.unresolved_fields) == 2
+
+
+class TestSemanticSourceResolverResolveFromReference:
+    """SemanticSourceResolver.resolve_from_reference_dataset — reference dataset."""
+
+    def test_resolve_from_reference(self):
+        from src.services.dataset_review.semantic_resolver import SemanticSourceResolver
+
+        resolver = SemanticSourceResolver()
+        result = resolver.resolve_from_reference_dataset(
+            {"source_ref": "ref_1"},
+            [{"field_name": "region"}],
+        )
+        assert result.source_ref == "reference_dataset"
+
+
+class TestSemanticSourceResolverRankCandidates:
+    """SemanticSourceResolver.rank_candidates — confidence ordering."""
+
+    def test_exact_before_fuzzy(self):
+        from src.services.dataset_review.semantic_resolver import SemanticSourceResolver
+        from src.models.dataset_review import CandidateMatchType
+
+        resolver = SemanticSourceResolver()
+        candidates = [
+            {"match_type": CandidateMatchType.FUZZY.value, "confidence_score": 0.95, "candidate_rank": 999},
+            {"match_type": CandidateMatchType.EXACT.value, "confidence_score": 1.0, "candidate_rank": 999},
+        ]
+        ranked = resolver.rank_candidates(candidates)
+        assert ranked[0]["match_type"] == CandidateMatchType.EXACT.value
+        assert ranked[1]["match_type"] == CandidateMatchType.FUZZY.value
+
+    def test_higher_confidence_first(self):
+        from src.services.dataset_review.semantic_resolver import SemanticSourceResolver
+        from src.models.dataset_review import CandidateMatchType
+
+        resolver = SemanticSourceResolver()
+        candidates = [
+            {"match_type": CandidateMatchType.FUZZY.value, "confidence_score": 0.8, "candidate_rank": 999},
+            {"match_type": CandidateMatchType.FUZZY.value, "confidence_score": 0.9, "candidate_rank": 999},
+        ]
+        ranked = resolver.rank_candidates(candidates)
+        assert ranked[0]["confidence_score"] == 0.9
+        assert ranked[1]["confidence_score"] == 0.8
+
+    def test_ranks_assigned(self):
+        from src.services.dataset_review.semantic_resolver import SemanticSourceResolver
+        from src.models.dataset_review import CandidateMatchType
+
+        resolver = SemanticSourceResolver()
+        candidates = [
+            {"match_type": CandidateMatchType.EXACT.value, "confidence_score": 1.0, "candidate_rank": 999},
+            {"match_type": CandidateMatchType.FUZZY.value, "confidence_score": 0.9, "candidate_rank": 999},
+            {"match_type": CandidateMatchType.FUZZY.value, "confidence_score": 0.8, "candidate_rank": 999},
+        ]
+        ranked = resolver.rank_candidates(candidates)
+        assert ranked[0]["candidate_rank"] == 1
+        assert ranked[1]["candidate_rank"] == 2
+        assert ranked[2]["candidate_rank"] == 3
+
+
+class TestSemanticSourceResolverDetectConflicts:
+    """SemanticSourceResolver.detect_conflicts — multi-candidate detection."""
+
+    def test_single_candidate_no_conflict(self):
+        from src.services.dataset_review.semantic_resolver import SemanticSourceResolver
+
+        resolver = SemanticSourceResolver()
+        assert resolver.detect_conflicts([{"rank": 1}]) is False
+
+    def test_multiple_candidates_conflict(self):
+        from src.services.dataset_review.semantic_resolver import SemanticSourceResolver
+
+        resolver = SemanticSourceResolver()
+        assert resolver.detect_conflicts([{"rank": 1}, {"rank": 2}]) is True
+
+
+class TestSemanticSourceResolverApplyFieldDecision:
+    """SemanticSourceResolver.apply_field_decision — field merging."""
+
+    def test_merge_fields(self):
+        from src.services.dataset_review.semantic_resolver import SemanticSourceResolver
+
+        resolver = SemanticSourceResolver()
+        field_state = {"field_name": "region", "status": "unresolved"}
+        decision = {"status": "resolved", "verbose_name": "Region"}
+        merged = resolver.apply_field_decision(field_state, decision)
+        assert merged["field_name"] == "region"
+        assert merged["status"] == "resolved"
+        assert merged["verbose_name"] == "Region"
+
+
+class TestSemanticSourceResolverPropagateSourceVersion:
+    """SemanticSourceResolver.propagate_source_version_update — version propagation."""
+
+    def test_propagate_success(self):
+        from src.services.dataset_review.semantic_resolver import SemanticSourceResolver
+        from src.models.dataset_review import SemanticSource
+
+        resolver = SemanticSourceResolver()
+        source = SemanticSource(source_id="src-1", source_version="v2")
+
+        field_mock = MagicMock()
+        field_mock.source_id = "src-1"
+        field_mock.is_locked = False
+        field_mock.provenance = "dictionary_exact"
+        field_mock.source_version = "v1"
+        field_mock.needs_review = False
+        field_mock.has_conflict = False
+
+        result = resolver.propagate_source_version_update(source, [field_mock])
+        assert result["propagated"] == 1
+        assert result["preserved_locked"] == 0
+        assert result["untouched"] == 0
+        assert field_mock.source_version == "v2"
+        assert field_mock.needs_review is True
+
+    def test_preserve_locked(self):
+        from src.services.dataset_review.semantic_resolver import SemanticSourceResolver
+        from src.models.dataset_review import SemanticSource
+
+        resolver = SemanticSourceResolver()
+        source = SemanticSource(source_id="src-1", source_version="v2")
+
+        field_mock = MagicMock()
+        field_mock.source_id = "src-1"
+        field_mock.is_locked = True
+        field_mock.provenance = "manual_override"
+        field_mock.source_version = "v1"
+
+        result = resolver.propagate_source_version_update(source, [field_mock])
+        assert result["propagated"] == 0
+        assert result["preserved_locked"] == 1
+
+    def test_untouched_different_source(self):
+        from src.services.dataset_review.semantic_resolver import SemanticSourceResolver
+        from src.models.dataset_review import SemanticSource
+
+        resolver = SemanticSourceResolver()
+        source = SemanticSource(source_id="src-1", source_version="v2")
+
+        field_mock = MagicMock()
+        field_mock.source_id = "src-2"
+        field_mock.is_locked = False
+        field_mock.provenance = "dictionary_exact"
+
+        result = resolver.propagate_source_version_update(source, [field_mock])
+        assert result["propagated"] == 0
+        assert result["untouched"] == 1
+
+    def test_missing_source_metadata_raises(self):
+        from src.services.dataset_review.semantic_resolver import SemanticSourceResolver
+        from src.models.dataset_review import SemanticSource
+
+        resolver = SemanticSourceResolver()
+        source = SemanticSource(source_id="src-1", source_version="")
+        with pytest.raises(ValueError, match="source_version"):
+            resolver.propagate_source_version_update(source, [])
+
+
+class TestNormalizeDictionaryRow:
+    """SemanticSourceResolver._normalize_dictionary_row — row normalization."""
+
+    def test_field_name_variants(self):
+        from src.services.dataset_review.semantic_resolver import SemanticSourceResolver
+
+        resolver = SemanticSourceResolver()
+        r1 = resolver._normalize_dictionary_row({"column_name": "col1", "verbose_name": "Col 1"})
+        assert r1["field_name"] == "col1"
+        r2 = resolver._normalize_dictionary_row({"name": "named_field"})
+        assert r2["field_name"] == "named_field"
+        r3 = resolver._normalize_dictionary_row({"field": "field_val"})
+        assert r3["field_name"] == "field_val"
+        r4 = resolver._normalize_dictionary_row({"field_name": "direct"})
+        assert r4["field_name"] == "direct"
+
+    def test_verbose_name_fallback(self):
+        from src.services.dataset_review.semantic_resolver import SemanticSourceResolver
+
+        resolver = SemanticSourceResolver()
+        r1 = resolver._normalize_dictionary_row({"field_name": "a", "label": "Label A"})
+        assert r1["verbose_name"] == "Label A"
+        r2 = resolver._normalize_dictionary_row({"field_name": "b"})
+        assert r2.get("verbose_name") is None
+
+
+class TestMatchPriority:
+    """SemanticSourceResolver._match_priority — priority encoding."""
+
+    def test_priority_ordering(self):
+        from src.services.dataset_review.semantic_resolver import SemanticSourceResolver
+        from src.models.dataset_review import CandidateMatchType
+
+        resolver = SemanticSourceResolver()
+        assert resolver._match_priority(CandidateMatchType.EXACT.value) == 0
+        assert resolver._match_priority(CandidateMatchType.FUZZY.value) == 2
+        assert resolver._match_priority(CandidateMatchType.GENERATED.value) == 3
+        assert resolver._match_priority(None) == 99
+        assert resolver._match_priority("UNKNOWN") == 99
+
+
+class TestNormalizeKey:
+    """SemanticSourceResolver._normalize_key — key normalization."""
+
+    @pytest.mark.parametrize("input_val,expected", [
+        ("Hello World", "helloworld"),
+        ("region_code", "region_code"),
+        ("UPPER_CASE", "uppercase"),
+        (" with spaces ", "withspaces"),
+        ("special!@#chars", "specialchars"),
+        ("", ""),
+    ])
+    def test_normalize_key(self, input_val, expected):
+        from src.services.dataset_review.semantic_resolver import SemanticSourceResolver
+
+        resolver = SemanticSourceResolver()
+        assert resolver._normalize_key(input_val) == expected
+# #endregion Test.SemanticSourceResolver