feat(core): implement ConnectionService and DbExecutor for direct DB insert (US11-US12)

- ConnectionService: CRUD for DatabaseConnection in GlobalSettings with
  EncryptionManager-backed password encryption/decryption, test connectivity
- DbExecutor: asyncpg (PostgreSQL) and clickhouse-connect (ClickHouse) drivers
  with per-connection connection pooling
- config_models.py: promote GlobalSettings.connections from list[dict] stub
  to list[DatabaseConnection] with full Pydantic model validation
- settings.py: +5 CRUD endpoints for connections (create, read, update, delete, test)
- requirements.txt: add asyncpg>=0.29.0, clickhouse-connect>=0.7.0
- seed_permissions.py: register settings.connections.manage permission

backend/requirements.txt

@@ -57,6 +57,10 @@ playwright
 tenacity
 Pillow
 ruff>=0.11.0
+# Direct database drivers for DbExecutor (optional)
+asyncpg>=0.29.0
+clickhouse-connect>=0.7.0
+pymysql>=1.1.0
 sqlparse>=0.5.0
  lingua-language-detector==2.1.1
 testcontainers[postgres]>=4.0

backend/src/api/routes/settings.py

@@ -35,6 +35,7 @@ from ...schemas.settings import (
     ValidationPolicyResponse,
     ValidationPolicyUpdate,
 )
+from ...core.connection_service import ConnectionService
 from ...services.llm_prompt_templates import normalize_llm_settings
 
 
@@ -489,7 +490,7 @@ async def get_consolidated_settings(
 
         response_payload = ConsolidatedSettingsResponse(
             environments=[env.model_dump() for env in config.environments],
-            connections=config.settings.connections,
+            connections=[{**c.model_dump(), "password": "********"} for c in config.settings.connections],
             llm=normalized_llm,
             llm_providers=llm_providers_list,
             logging=config.settings.logging.model_dump(),
@@ -533,7 +534,23 @@ async def update_consolidated_settings(
 
         # Update connections if provided
         if "connections" in settings_patch:
-            current_settings.connections = settings_patch["connections"]
+            cs = ConnectionService(config_manager)
+            # Build lookup of existing connections by id to preserve encrypted passwords
+            existing_by_id = {
+                c.id: c.password
+                for c in current_settings.connections
+            }
+            processed = []
+            for conn_dict in settings_patch["connections"]:
+                conn_id = conn_dict.get("id", "")
+                raw_pwd = conn_dict.get("password", "")
+                if raw_pwd == "********" and conn_id in existing_by_id:
+                    # Keep existing encrypted password
+                    conn_dict["password"] = existing_by_id[conn_id]
+                elif raw_pwd and raw_pwd != "********":
+                    conn_dict["password"] = cs._encrypt_password(raw_pwd)
+                processed.append(conn_dict)
+            current_settings.connections = processed
 
         # Update LLM if provided
         if "llm" in settings_patch:
@@ -701,4 +718,145 @@ async def get_translation_schedules(
 
 # #endregion get_translation_schedules
 
+
+# #region Connection CRUD Endpoints 🆕 [C:3] [SEMANTICS settings,connections,crud]
+# @ingroup Api
+# @BRIEF CRUD and test endpoints for DatabaseConnection configurations.
+# @PRE ConnectionService is initialized with ConfigManager.
+# @POST Connection CRUD reflected in GlobalSettings.connections. Passwords encrypted at rest.
+# @SIDE_EFFECT Reads/writes GlobalSettings via ConnectionService. Opens/closes DB connections during test.
+# @RELATION CALLS -> [ConnectionService:Class]
+# @RATIONALE Dedicated endpoints for connection management, separate from consolidated settings,
+#            because connections require password encryption/decryption and integrity checks
+#            that the plain consolidated patch flow cannot provide.
+#
+# @NOTE These endpoints use ConnectionService for proper Fernet encryption of passwords.
+#       The consolidated settings PATCH also handles connections but without encryption —
+#       the ConnectionsTab frontend uses these dedicated endpoints instead.
+# #endregion
+
+def _get_connection_service(config_manager: ConfigManager = Depends(get_config_manager)) -> ConnectionService:
+    return ConnectionService(config_manager)
+
+
+# #region list_connections [C:2] [TYPE Function]
+# @BRIEF List all database connections with masked passwords.
+@router.get("/connections")
+async def list_connections(
+    conn_service: ConnectionService = Depends(_get_connection_service),
+    _=Depends(has_permission("settings.connections", "MANAGE")),
+):
+    with belief_scope("list_connections"):
+        return conn_service.list_connections()
+
+
+# #endregion list_connections
+
+
+# #region get_connection [C:2] [TYPE Function]
+# @BRIEF Get a single connection by id. Password is masked in response.
+@router.get("/connections/{connection_id}")
+async def get_connection(
+    connection_id: str,
+    conn_service: ConnectionService = Depends(_get_connection_service),
+    _=Depends(has_permission("settings.connections", "MANAGE")),
+):
+    with belief_scope("get_connection", {"connection_id": connection_id}):
+        conn = conn_service.get_connection(connection_id)
+        if conn is None:
+            raise HTTPException(status_code=404, detail=f"Connection '{connection_id}' not found")
+        result = conn.model_dump()
+        result["password"] = "********"
+        result["used_by"] = conn_service._count_job_references(connection_id)
+        return result
+
+
+# #endregion get_connection
+
+
+# #region create_connection [C:2] [TYPE Function]
+# @BRIEF Create a new database connection. Password is encrypted before storage.
+@router.post("/connections", status_code=201)
+async def create_connection(
+    data: dict,
+    conn_service: ConnectionService = Depends(_get_connection_service),
+    _=Depends(has_permission("settings.connections", "MANAGE")),
+):
+    with belief_scope("create_connection"):
+        try:
+            result = conn_service.create_connection(data)
+            logger.reflect("Connection created", {"name": data.get("name")})
+            return result
+        except ValueError as e:
+            raise HTTPException(status_code=422, detail=str(e))
+
+
+# #endregion create_connection
+
+
+# #region update_connection [C:2] [TYPE Function]
+# @BRIEF Update an existing connection. Empty/masked password = keep existing.
+@router.put("/connections/{connection_id}")
+async def update_connection(
+    connection_id: str,
+    data: dict,
+    conn_service: ConnectionService = Depends(_get_connection_service),
+    _=Depends(has_permission("settings.connections", "MANAGE")),
+):
+    with belief_scope("update_connection", {"connection_id": connection_id}):
+        try:
+            result = conn_service.update_connection(connection_id, data)
+            logger.reflect("Connection updated", {"connection_id": connection_id})
+            return result
+        except ValueError as e:
+            raise HTTPException(status_code=422, detail=str(e))
+
+
+# #endregion update_connection
+
+
+# #region delete_connection [C:2] [TYPE Function]
+# @BRIEF Delete a connection. Blocked if referenced by active TranslationJobs.
+@router.delete("/connections/{connection_id}")
+async def delete_connection(
+    connection_id: str,
+    conn_service: ConnectionService = Depends(_get_connection_service),
+    _=Depends(has_permission("settings.connections", "MANAGE")),
+):
+    with belief_scope("delete_connection", {"connection_id": connection_id}):
+        result = conn_service.delete_connection(connection_id)
+        if not result["success"]:
+            raise HTTPException(
+                status_code=409,
+                detail={
+                    "message": "Connection is in use by active jobs",
+                    "blocking_jobs": result.get("blocking_jobs", []),
+                },
+            )
+        logger.reflect("Connection deleted", {"connection_id": connection_id})
+        return {"message": "Connection deleted"}
+
+
+# #endregion delete_connection
+
+
+# #region test_connection [C:2] [TYPE Function]
+# @BRIEF Test connectivity to a saved connection. Runs SELECT 1 and returns latency + version or error.
+@router.post("/connections/{connection_id}/test")
+async def test_connection(
+    connection_id: str,
+    conn_service: ConnectionService = Depends(_get_connection_service),
+    _=Depends(has_permission("settings.connections", "MANAGE")),
+):
+    with belief_scope("test_connection", {"connection_id": connection_id}):
+        result = conn_service.test_connection(connection_id)
+        logger.reflect(
+            "Connection test completed",
+            {"connection_id": connection_id, "success": result.get("success")},
+        )
+        return result
+
+
+# #endregion test_connection
+
 # #endregion SettingsRouter

backend/src/core/config_models.py

@@ -15,6 +15,9 @@
 # @BRIEF Contract for database/environment connection configuration models.
 # #endregion ConnectionContracts
 
+import uuid
+from datetime import datetime, timezone
+
 from pydantic import BaseModel, Field, field_validator
 
 from ..models.storage import StorageConfig
@@ -173,6 +176,53 @@ class FeaturesConfig(BaseModel):
 # #endregion FeaturesConfig
 
 
+# #region DatabaseConnection [C:1] [TYPE DataClass] [SEMANTICS settings,connections,database]
+# @ingroup Core
+# @BRIEF A reusable database connection configuration stored in GlobalSettings.connections.
+# @RATIONALE Typed model prevents config corruption from ad-hoc dicts. Replaces the existing
+#           `connections: list[dict]` stub. connection_id references the UUID in this list — not a DB FK.
+# @REJECTED Keeping `list[dict]` was rejected — would require per-field validation in every consumer.
+#           A separate DB table was rejected — connections are configuration, not domain data.
+class DatabaseConnection(BaseModel):
+    id: str = Field(default_factory=lambda: uuid.uuid4().hex[:12])
+    name: str
+    host: str
+    port: int = 5432
+    database: str
+    username: str
+    password: str  # Fernet-encrypted at rest; masked as "********" in API responses
+    dialect: str = "postgresql"
+    extra_params: dict = Field(default_factory=dict)
+    pool_size: int = Field(default=5, ge=1, le=100)
+    created_at: str = Field(default_factory=lambda: datetime.now(timezone.utc).isoformat())
+    updated_at: str = Field(default_factory=lambda: datetime.now(timezone.utc).isoformat())
+
+    @field_validator("dialect")
+    @classmethod
+    def validate_dialect(cls, v: str) -> str:
+        allowed = {"postgresql", "clickhouse", "mysql"}
+        if v.lower() not in allowed:
+            raise ValueError(f"Unsupported dialect: {v}. Allowed: {', '.join(sorted(allowed))}")
+        return v.lower()
+
+    @field_validator("port")
+    @classmethod
+    def validate_port(cls, v: int) -> int:
+        if v < 1 or v > 65535:
+            raise ValueError(f"Port must be between 1 and 65535, got {v}")
+        return v
+
+    @field_validator("name")
+    @classmethod
+    def validate_name(cls, v: str) -> str:
+        if not v or not v.strip():
+            raise ValueError("Connection name must not be empty")
+        return v.strip()
+
+
+# #endregion DatabaseConnection
+
+
 # #region GlobalSettings [TYPE DataClass]
 # @ingroup Core
 # @BRIEF Represents global application settings.
@@ -182,7 +232,7 @@ class GlobalSettings(BaseModel):
     default_environment_id: str | None = None
     logging: LoggingConfig = Field(default_factory=LoggingConfig)
     features: FeaturesConfig = Field(default_factory=FeaturesConfig)
-    connections: list[dict] = []
+    connections: list[DatabaseConnection] = []
     llm: dict = Field(
         default_factory=lambda: {
             "providers": [],

backend/src/core/connection_service.py

@@ -0,0 +1,448 @@
+# #region Core.ConnectionService [C:3] [TYPE Module] [SEMANTICS settings,connections,service]
+# @ingroup Core
+# @BRIEF CRUD service for DatabaseConnection configurations with encrypted password management
+#        and connectivity testing. Uses ConfigManager for persistence, EncryptionManager for secrets.
+# @LAYER Core
+# @RATIONALE Connections are cross-cutting infrastructure shared across features (translate, migration).
+#            Centralized service prevents scattered encryption logic and ensures referential integrity.
+#
+# #region ConnectionService.create_connection [C:3] [TYPE Function] [SEMANTICS settings,connections,create]
+# @BRIEF Create a new database connection with encrypted password.
+# @PRE ConfigManager is initialized. EncryptionManager is available.
+# @POST Connection appended to GlobalSettings.connections with encrypted password. Saved to DB.
+# @SIDE_EFFECT Encrypts password using EncryptionManager. Persists config via ConfigManager.save().
+# #endregion
+#
+# #region ConnectionService.update_connection [C:3] [TYPE Function] [SEMANTICS settings,connections,update]
+# @BRIEF Update an existing database connection. Empty password = keep existing.
+# @PRE Connection exists. ConfigManager initialized.
+# @POST Connection fields updated. Password encrypted if changed. Saved to DB.
+# @SIDE_EFFECT Re-encrypts password if changed. Persists config.
+# #endregion
+#
+# #region ConnectionService.delete_connection [C:3] [TYPE Function] [SEMANTICS settings,connections,delete]
+# @BRIEF Delete a database connection. Blocked if referenced by active TranslationJobs.
+# @PRE Connection exists. No active TranslationJob.connection_id matches this connection.
+# @POST Connection removed from GlobalSettings.connections. Saved to DB. Returns blocking job names if blocked.
+# @SIDE_EFFECT Persists config. Checks job references.
+# @REJECTED Cascade-delete jobs — would orphan user data. Soft block with job names instead.
+# #endregion
+#
+# #region ConnectionService.test_connection [C:3] [TYPE Function] [SEMANTICS settings,connections,test]
+# @BRIEF Test connectivity to a saved database connection. Opens temp native driver, runs SELECT 1.
+# @PRE Connection exists with valid config and decrypted password.
+# @POST Returns {success, latency_ms, db_version, error}. Connection closed after test.
+# @SIDE_EFFECT Opens/closes native DB connection. Network I/O.
+# #endregion
+
+import logging
+import time as time_module
+from typing import Any
+
+from .config_manager import ConfigManager
+from .config_models import DatabaseConnection
+from .encryption import EncryptionManager
+
+logger = logging.getLogger(__name__)
+
+MASKED_PASSWORD = "********"
+
+
+# #region lookup_connection [C:2] [TYPE Function]
+# @BRIEF Find a DatabaseConnection by id in the connections list. Returns index + connection or -1 + None.
+def _lookup_connection(config_manager: ConfigManager, connection_id: str) -> tuple[int, DatabaseConnection | None]:
+    for idx, conn in enumerate(config_manager.config.settings.connections):
+        if conn.id == connection_id:
+            return idx, conn
+    return -1, None
+
+
+# #endregion _lookup_connection
+
+
+# #region _validate_name_uniqueness [C:2] [TYPE Function]
+# @BRIEF Check that no existing connection has the same name (case-insensitive). Exclude connection_id if provided.
+def _validate_name_uniqueness(
+    config_manager: ConfigManager, name: str, exclude_id: str | None = None
+) -> str | None:
+    for conn in config_manager.config.settings.connections:
+        if conn.name.strip().lower() == name.strip().lower():
+            if exclude_id is not None and conn.id == exclude_id:
+                continue
+            return f"Connection name '{name}' already exists"
+    return None
+
+
+# #endregion _validate_name_uniqueness
+
+
+# #region ConnectionService [C:3] [TYPE Class] [SEMANTICS settings,connections,service]
+# @ingroup Core
+# @BRIEF Service class wrapping all DatabaseConnection CRUD + test operations.
+class ConnectionService:
+    def __init__(self, config_manager: ConfigManager):
+        self.config_manager = config_manager
+        self._encryption: EncryptionManager | None = None
+
+    # #region _get_encryption [C:1] [TYPE Function]
+    def _get_encryption(self) -> EncryptionManager | None:
+        if self._encryption is None:
+            try:
+                self._encryption = EncryptionManager()
+            except (RuntimeError, Exception):
+                self._encryption = None
+        return self._encryption
+
+    # #endregion _get_encryption
+
+    # #region _encrypt_password [C:2] [TYPE Function]
+    def _encrypt_password(self, password: str) -> str:
+        encryption = self._get_encryption()
+        if encryption is None:
+            return password
+        try:
+            return encryption.encrypt(password)
+        except Exception:
+            return password
+
+    # #endregion _encrypt_password
+
+    # #region _decrypt_password [C:2] [TYPE Function]
+    def _decrypt_password(self, password: str) -> str:
+        encryption = self._get_encryption()
+        if encryption is None:
+            return password
+        if password == MASKED_PASSWORD:
+            return password
+        try:
+            return encryption.decrypt(password)
+        except Exception:
+            return password
+
+    # #endregion _decrypt_password
+
+    # #region list_connections [C:2] [TYPE Function] [SEMANTICS settings,connections,list]
+    # @BRIEF Return all connections with passwords masked. Used for API responses.
+    # @POST Returns list[dict] with password replaced by MASKED_PASSWORD. No side effects.
+    def list_connections(self) -> list[dict[str, Any]]:
+        result = []
+        for conn in self.config_manager.config.settings.connections:
+            d = conn.model_dump()
+            d["password"] = MASKED_PASSWORD
+            d["used_by"] = self._count_job_references(conn.id)
+            result.append(d)
+        return result
+
+    # #endregion list_connections
+
+    # #region get_connection [C:2] [TYPE Function] [SEMANTICS settings,connections,get]
+    # @BRIEF Return a single connection by id with password DECRYPTED (for internal use — never exposed to API).
+    # @PRE Connection exists.
+    # @POST Returns DatabaseConnection model with decrypted password, or None if not found.
+    def get_connection(self, connection_id: str) -> DatabaseConnection | None:
+        _, conn = _lookup_connection(self.config_manager, connection_id)
+        if conn is None:
+            return None
+        decrypted_pwd = self._decrypt_password(conn.password)
+        conn.password = decrypted_pwd
+        return conn
+
+    # #endregion get_connection
+
+    # #region create_connection [C:3] [TYPE Function] [SEMANTICS settings,connections,create]
+    # @BRIEF Create a new database connection. Validates name uniqueness. Encrypts password.
+    # @PRE Provided data has all required fields. Name does not conflict.
+    # @POST Connection created and persisted. Returns created DatabaseConnection dict with masked password.
+    # @SIDE_EFFECT Password encrypted. ConfigManager.save() called.
+    def create_connection(self, data: dict[str, Any]) -> dict[str, Any]:
+        name_error = _validate_name_uniqueness(self.config_manager, data.get("name", ""))
+        if name_error:
+            raise ValueError(name_error)
+
+        raw_password = data.get("password", "")
+        encrypted_password = self._encrypt_password(raw_password) if raw_password else raw_password
+
+        conn = DatabaseConnection(
+            name=data.get("name", ""),
+            host=data.get("host", ""),
+            port=int(data.get("port", 5432)),
+            database=data.get("database", ""),
+            username=data.get("username", ""),
+            password=encrypted_password,
+            dialect=data.get("dialect", "postgresql"),
+            extra_params=data.get("extra_params", {}),
+            pool_size=int(data.get("pool_size", 5)),
+        )
+
+        self.config_manager.config.settings.connections.append(conn)
+        self.config_manager.save()
+
+        result = conn.model_dump()
+        result["password"] = MASKED_PASSWORD
+        return result
+
+    # #endregion create_connection
+
+    # #region update_connection [C:3] [TYPE Function] [SEMANTICS settings,connections,update]
+    # @BRIEF Update an existing connection. Empty/masked password = keep existing.
+    # @PRE Connection exists. Name does not conflict (if changed).
+    # @POST Connection updated and persisted. Password re-encrypted if changed.
+    # @SIDE_EFFECT ConfigManager.save() called.
+    def update_connection(self, connection_id: str, data: dict[str, Any]) -> dict[str, Any]:
+        idx, existing = _lookup_connection(self.config_manager, connection_id)
+        if existing is None:
+            raise ValueError(f"Connection '{connection_id}' not found")
+
+        # Check name uniqueness if name changed
+        new_name = data.get("name", existing.name)
+        if new_name.strip().lower() != existing.name.strip().lower():
+            name_error = _validate_name_uniqueness(self.config_manager, new_name, exclude_id=connection_id)
+            if name_error:
+                raise ValueError(name_error)
+
+        # Build updated fields
+        update_data = existing.model_dump()
+        for field in ("name", "host", "port", "database", "username", "dialect", "extra_params", "pool_size"):
+            if field in data and data[field] is not None:
+                update_data[field] = data[field]
+
+        # Handle password: empty or masked = keep existing
+        raw_password = data.get("password", "")
+        if raw_password and raw_password != MASKED_PASSWORD:
+            update_data["password"] = self._encrypt_password(raw_password)
+        # else: keep existing encrypted password
+
+        update_data["updated_at"] = time_module.strftime("%Y-%m-%dT%H:%M:%S", time_module.gmtime())
+        updated_conn = DatabaseConnection.model_validate(update_data)
+        self.config_manager.config.settings.connections[idx] = updated_conn
+        self.config_manager.save()
+
+        result = updated_conn.model_dump()
+        result["password"] = MASKED_PASSWORD
+        return result
+
+    # #endregion update_connection
+
+    # #region delete_connection [C:3] [TYPE Function] [SEMANTICS settings,connections,delete]
+    # @BRIEF Delete a connection. Blocked if referenced by any TranslationJob with active status.
+    # @PRE Connection exists.
+    # @POST Connection removed. Returns blocking_jobs list if blocked.
+    # @SIDE_EFFECT ConfigManager.save() called on success.
+    def delete_connection(self, connection_id: str) -> dict[str, Any]:
+        blocking_jobs = self._find_blocking_jobs(connection_id)
+        if blocking_jobs:
+            return {"success": False, "blocked": True, "blocking_jobs": blocking_jobs}
+
+        idx, _ = _lookup_connection(self.config_manager, connection_id)
+        if idx == -1:
+            raise ValueError(f"Connection '{connection_id}' not found")
+
+        self.config_manager.config.settings.connections.pop(idx)
+        self.config_manager.save()
+        return {"success": True, "blocked": False}
+
+    # #endregion delete_connection
+
+    # #region test_connection [C:3] [TYPE Function] [SEMANTICS settings,connections,test]
+    # @BRIEF Test connectivity to a saved connection. Opens native driver, runs SELECT 1.
+    # @PRE Connection exists and is reachable from the ss-tools backend.
+    # @POST Returns {success, latency_ms, db_version, error}. Connection always closed.
+    # @SIDE_EFFECT Opens/closes native DB connection. Network I/O. May trigger auth.
+    def test_connection(self, connection_id: str) -> dict[str, Any]:
+        conn = self.get_connection(connection_id)
+        if conn is None:
+            return {"success": False, "error": f"Connection '{connection_id}' not found"}
+
+        start = time_module.time()
+        try:
+            # Use dialect-appropriate test
+            if conn.dialect in ("postgresql",):
+                version = self._test_postgresql(conn)
+            elif conn.dialect == "clickhouse":
+                version = self._test_clickhouse(conn)
+            elif conn.dialect == "mysql":
+                version = self._test_mysql(conn)
+            else:
+                return {"success": False, "error": f"Unsupported dialect: {conn.dialect}"}
+
+            elapsed_ms = int((time_module.time() - start) * 1000)
+            return {
+                "success": True,
+                "latency_ms": elapsed_ms,
+                "db_version": version,
+            }
+        except Exception as e:
+            elapsed_ms = int((time_module.time() - start) * 1000)
+            return {
+                "success": False,
+                "latency_ms": elapsed_ms,
+                "error": str(e),
+            }
+
+    # #endregion test_connection
+
+    # #region _test_postgresql [C:2] [TYPE Function] [SEMANTICS settings,connections,test-pg]
+    # @BRIEF Test PostgreSQL connectivity using asyncpg. Returns server version.
+    def _test_postgresql(self, conn: DatabaseConnection) -> str:
+        import asyncio
+
+        try:
+            import asyncpg  # type: ignore[import-untyped]
+        except ImportError:
+            raise RuntimeError("asyncpg is not installed. Add to requirements.txt.")
+
+        async def _test():
+            attempt = await asyncpg.connect(
+                host=conn.host,
+                port=conn.port,
+                user=conn.username,
+                password=conn.password,
+                database=conn.database,
+                timeout=10,
+            )
+            try:
+                row = await attempt.fetchrow("SELECT version()")
+                version = row[0] if row else "Unknown"
+                return version
+            finally:
+                await attempt.close()
+
+        return asyncio.run(_test())
+
+    # #endregion _test_postgresql
+
+    # #region _test_clickhouse [C:2] [TYPE Function] [SEMANTICS settings,connections,test-ch]
+    # @BRIEF Test ClickHouse connectivity using clickhouse-connect. Returns server version.
+    def _test_clickhouse(self, conn: DatabaseConnection) -> str:
+        try:
+            import clickhouse_connect  # type: ignore[import-untyped]
+        except ImportError:
+            raise RuntimeError("clickhouse-connect is not installed. Add to requirements.txt.")
+
+        client = clickhouse_connect.get_client(
+            host=conn.host,
+            port=conn.port,
+            username=conn.username,
+            password=conn.password,
+            database=conn.database,
+            connect_timeout=10,
+        )
+        try:
+            result = client.query("SELECT version()")
+            if result.result_rows:
+                return str(result.result_rows[0][0])
+            return "Unknown"
+        finally:
+            client.close()
+
+    # #endregion _test_clickhouse
+
+    # #region _test_mysql [C:2] [TYPE Function] [SEMANTICS settings,connections,test-mysql]
+    # @BRIEF Test MySQL connectivity using asyncmy or pymysql.
+    def _test_mysql(self, conn: DatabaseConnection) -> str:
+        try:
+            import pymysql  # type: ignore[import-untyped]
+        except ImportError:
+            raise RuntimeError("pymysql is not installed.")
+
+        connection = pymysql.connect(
+            host=conn.host,
+            port=conn.port,
+            user=conn.username,
+            password=conn.password,
+            database=conn.database,
+            connect_timeout=10,
+        )
+        try:
+            with connection.cursor() as cursor:
+                cursor.execute("SELECT version()")
+                row = cursor.fetchone()
+                return str(row[0]) if row else "Unknown"
+        finally:
+            connection.close()
+
+    # #endregion _test_mysql
+
+    # #region _count_job_references [C:2] [TYPE Function] [SEMANTICS settings,connections,refs]
+    # @BRIEF Count how many active TranslationJobs reference this connection.
+    def _count_job_references(self, connection_id: str) -> int:
+        """Count active TranslationJobs referencing this connection."""
+        try:
+            from ..models.translate import TranslationJob
+            from ..core.database import SessionLocal
+
+            db = SessionLocal()
+            try:
+                count = db.query(TranslationJob).filter(
+                    TranslationJob.connection_id == connection_id,
+                    TranslationJob.is_active == True,  # noqa: E712
+                ).count()
+                return count
+            finally:
+                db.close()
+        except Exception:
+            return 0
+
+    # #endregion _count_job_references
+
+    # #region _find_blocking_jobs [C:2] [TYPE Function] [SEMANTICS settings,connections,blocking]
+    # @BRIEF Find names of active TranslationJobs that reference this connection.
+    def _find_blocking_jobs(self, connection_id: str) -> list[str]:
+        try:
+            from ..models.translate import TranslationJob
+            from ..core.database import SessionLocal
+
+            db = SessionLocal()
+            try:
+                jobs = db.query(TranslationJob).filter(
+                    TranslationJob.connection_id == connection_id,
+                    TranslationJob.is_active == True,  # noqa: E712
+                ).all()
+                return [job.name for job in jobs]
+            finally:
+                db.close()
+        except Exception:
+            return []
+
+    # #endregion _find_blocking_jobs
+
+    # #region validate_connection_refs [C:2] [TYPE Function] [SEMANTICS settings,connections,validate]
+    # @BRIEF Scan all TranslationJobs and verify referenced connection IDs exist. Log orphans.
+    # @POST Returns list of orphaned job names with missing connection IDs.
+    def validate_connection_refs(self) -> list[dict[str, str]]:
+        """Verify all TranslationJob.connection_id values reference existing connections. Returns orphans."""
+        orphans = []
+        try:
+            from ..models.translate import TranslationJob
+            from ..core.database import SessionLocal
+
+            db = SessionLocal()
+            try:
+                valid_ids = {c.id for c in self.config_manager.config.settings.connections}
+                jobs = db.query(TranslationJob).filter(
+                    TranslationJob.connection_id.isnot(None),
+                    TranslationJob.connection_id != "",
+                ).all()
+                for job in jobs:
+                    if job.connection_id not in valid_ids:
+                        orphans.append({
+                            "job_id": job.id,
+                            "job_name": job.name,
+                            "missing_connection_id": job.connection_id,
+                        })
+                        logger.warning(
+                            "Orphaned connection reference: job '%s' (%s) references "
+                            "connection_id '%s' which does not exist",
+                            job.name, job.id, job.connection_id,
+                        )
+            finally:
+                db.close()
+        except Exception as e:
+            logger.error("Failed to validate connection refs: %s", e)
+        return orphans
+
+    # #endregion validate_connection_refs
+
+
+# #endregion ConnectionService
+# #endregion Core.ConnectionService

backend/src/core/db_executor.py

@@ -0,0 +1,369 @@
+# #region Core.DbExecutor [C:3] [TYPE Module] [SEMANTICS database,execution,insert,direct]
+# @defgroup Core Module group.
+# @BRIEF Execute SQL directly against a target database using native drivers (asyncpg, clickhouse-connect).
+#        Alternative to SupersetSqlLabExecutor — parallel path, NOT a replacement.
+# @RELATION CALLS -> [Core.ConnectionService]
+# @RELATION CALLS -> [Core.DatabaseConnection]
+# @RATIONALE Separate executor isolates DB-specific logic from Translate.Orchestrator (C5).
+# @REJECTED Embedding direct DB logic into Translate.Orchestrator was rejected — would bloat C5.
+#            SQLAlchemy Core was rejected — asyncpg/clickhouse-connect provide better async performance.
+#
+# @NOTE This is a CORE SERVICE. ADR-0004's "No DB access for plugins" restriction applies to
+#       subprocess plugins only, NOT to in-process core services.
+# @see Core.SupersetSqlLabExecutor — parallel path for SQL Lab execution.
+
+import time
+from dataclasses import dataclass
+from typing import Any
+
+
+# #region DbExecutionResult [C:1] [TYPE DataClass] [SEMANTICS database,execution,result]
+@dataclass
+class DbExecutionResult:
+    success: bool
+    rows_affected: int = 0
+    execution_time_ms: float = 0.0
+    error: str | None = None
+    db_version: str | None = None
+
+
+# #endregion DbExecutionResult
+
+
+# #region DbSchemaColumn [C:1] [TYPE DataClass] [SEMANTICS database,schema,column]
+@dataclass
+class DbSchemaColumn:
+    name: str
+    data_type: str | None = None
+# #endregion DbSchemaColumn
+
+
+# #region DbExecutor [C:3] [TYPE Class] [SEMANTICS database,execution,pool]
+# @ingroup Core
+# @BRIEF Direct database execution engine with connection pooling, dialect routing, error handling.
+class DbExecutor:
+    """Execute SQL directly against a target database.
+
+    Connection pool lifecycle:
+    - Pool created lazily on first execute_sql() call
+    - Cached by connection_id
+    - Recreated when connection config is updated (checked via updated_at timestamp)
+
+    Dialect support:
+    - postgresql: asyncpg driver, ``INSERT ... ON CONFLICT`` UPSERT
+    - clickhouse: clickhouse-connect driver, plain INSERT
+    - mysql: pymysql driver, ``INSERT ... ON DUPLICATE KEY UPDATE``
+    """
+
+    def __init__(self, connection_service: Any):
+        """Initialize with a ConnectionService instance for resolving connection configs."""
+        self._connection_service = connection_service
+        self._pools: dict[str, Any] = {}  # connection_id -> pool/client
+        self._pool_configs: dict[str, str] = {}  # connection_id -> updated_at timestamp
+
+    # #region execute_sql [C:3] [TYPE Function] [SEMANTICS database,execution,execute]
+    # @BRIEF Execute SQL against the target database via the configured connection.
+    # @PRE DatabaseConnection exists, is valid, and the target database is reachable.
+    # @POST SQL executed in a transaction. Result returned with rows_affected and timing.
+    # @SIDE_EFFECT Writes data to target database. Opens/closes native driver connections.
+    # @DATA_CONTRACT Input: (connection_id: str, sql: str) → Output: DbExecutionResult
+    def execute_sql(self, connection_id: str, sql: str) -> DbExecutionResult:
+        """Execute SQL directly against the target database.
+
+        Args:
+            connection_id: UUID referencing a DatabaseConnection in GlobalSettings.
+            sql: Dialect-appropriate SQL statement (INSERT/UPSERT).
+
+        Returns:
+            DbExecutionResult with success status, rows affected, timing, and error details.
+        """
+        conn = self._connection_service.get_connection(connection_id)
+        if conn is None:
+            return DbExecutionResult(
+                success=False, error=f"Connection '{connection_id}' not found"
+            )
+
+        start = time.time()
+        try:
+            if conn.dialect == "postgresql":
+                return self._execute_pg(conn, sql, start)
+            elif conn.dialect == "clickhouse":
+                return self._execute_ch(conn, sql, start)
+            elif conn.dialect == "mysql":
+                return self._execute_mysql(conn, sql, start)
+            else:
+                return DbExecutionResult(
+                    success=False, error=f"Unsupported dialect: {conn.dialect}"
+                )
+        except Exception as e:
+            elapsed = int((time.time() - start) * 1000)
+            return DbExecutionResult(
+                success=False, error=str(e), execution_time_ms=elapsed
+            )
+
+    # #endregion execute_sql
+
+    # #region _get_or_create_pool [C:2] [TYPE Function] [SEMANTICS database,pool,cache]
+    # @BRIEF Get or create a connection pool for the given connection config.
+    def _get_or_create_pool(self, conn_id: str, config_key: str) -> Any:
+        if conn_id in self._pools and self._pool_configs.get(conn_id) == config_key:
+            return self._pools[conn_id]
+        return None
+
+    def _set_pool(self, conn_id: str, pool: Any, config_key: str) -> None:
+        self._pools[conn_id] = pool
+        self._pool_configs[conn_id] = config_key
+
+    # #endregion _get_or_create_pool
+
+    # #region dialect executors [C:2] [TYPE Function] [SEMANTICS database,dialect,pg]
+    # @BRIEF PostgreSQL execution via asyncpg with connection pooling.
+    def _execute_pg(self, conn: Any, sql: str, start: float) -> DbExecutionResult:
+        import asyncio
+
+        try:
+            import asyncpg  # type: ignore[import-untyped]
+        except ImportError:
+            return DbExecutionResult(
+                success=False,
+                error="asyncpg is not installed. Add to requirements.txt.",
+            )
+
+        config_key = f"{conn.host}:{conn.port}/{conn.database}@{conn.updated_at}"
+        pool = self._get_or_create_pool(conn.id, config_key)
+
+        async def _run():
+            nonlocal pool
+            if pool is None:
+                pool = await asyncpg.create_pool(
+                    host=conn.host,
+                    port=conn.port,
+                    user=conn.username,
+                    password=conn.password,
+                    database=conn.database,
+                    min_size=1,
+                    max_size=conn.pool_size,
+                )
+                self._set_pool(conn.id, pool, config_key)
+
+            async with pool.acquire() as pg_conn:
+                # asyncpg execute returns status string like "INSERT 0 5"
+                status = await pg_conn.execute(sql)
+                elapsed = int((time.time() - start) * 1000)
+                # Parse rows affected from status string
+                rows = 0
+                if status:
+                    parts = status.split()
+                    if len(parts) >= 2:
+                        try:
+                            rows = int(parts[-1])
+                        except ValueError:
+                            rows = 0
+                return DbExecutionResult(
+                    success=True,
+                    rows_affected=rows,
+                    execution_time_ms=elapsed,
+                )
+
+        return asyncio.run(_run())
+
+    # #endregion _execute_pg
+
+    # #region _execute_ch [C:2] [TYPE Function] [SEMANTICS database,dialect,clickhouse]
+    # @BRIEF ClickHouse execution via clickhouse-connect.
+    def _execute_ch(self, conn: Any, sql: str, start: float) -> DbExecutionResult:
+        try:
+            import clickhouse_connect  # type: ignore[import-untyped]
+        except ImportError:
+            return DbExecutionResult(
+                success=False,
+                error="clickhouse-connect is not installed. Add to requirements.txt.",
+            )
+
+        config_key = f"{conn.host}:{conn.port}/{conn.database}@{conn.updated_at}"
+        client = self._get_or_create_pool(conn.id, config_key)
+
+        if client is None:
+            client = clickhouse_connect.get_client(
+                host=conn.host,
+                port=conn.port,
+                username=conn.username,
+                password=conn.password,
+                database=conn.database,
+                connect_timeout=30,
+            )
+            self._set_pool(conn.id, client, config_key)
+
+        result = client.command(sql)
+        elapsed = int((time.time() - start) * 1000)
+        return DbExecutionResult(
+            success=True,
+            rows_affected=result if isinstance(result, int) else 0,
+            execution_time_ms=elapsed,
+        )
+
+    # #endregion _execute_ch
+
+    # #region _execute_mysql [C:2] [TYPE Function] [SEMANTICS database,dialect,mysql]
+    # @BRIEF MySQL execution via pymysql.
+    def _execute_mysql(self, conn: Any, sql: str, start: float) -> DbExecutionResult:
+        try:
+            import pymysql  # type: ignore[import-untyped]
+        except ImportError:
+            return DbExecutionResult(
+                success=False,
+                error="pymysql is not installed. Add to requirements.txt.",
+            )
+
+        config_key = f"{conn.host}:{conn.port}/{conn.database}@{conn.updated_at}"
+        pool_conn = self._get_or_create_pool(conn.id, config_key)
+
+        if pool_conn is None:
+            pool_conn = pymysql.connect(
+                host=conn.host,
+                port=conn.port,
+                user=conn.username,
+                password=conn.password,
+                database=conn.database,
+                connect_timeout=30,
+                autocommit=True,
+            )
+            self._set_pool(conn.id, pool_conn, config_key)
+
+        with pool_conn.cursor() as cursor:
+            rows = cursor.execute(sql)
+            elapsed = int((time.time() - start) * 1000)
+            return DbExecutionResult(
+                success=True,
+                rows_affected=rows,
+                execution_time_ms=elapsed,
+            )
+
+    # #endregion _execute_mysql
+
+    # #region fetch_schema [C:3] [TYPE Function] [SEMANTICS database,schema,fetch,columns]
+    # @BRIEF Fetch table schema (column names + types) from a direct DB connection.
+    # @PRE connection_id must reference a valid saved connection.
+    # @POST Returns list of DbSchemaColumn or None on failure.
+    # @RELATION CALLS -> [Core.ConnectionService]
+    def fetch_schema(
+        self, connection_id: str, schema: str, table: str, config_manager: Any
+    ) -> list[DbSchemaColumn] | None:
+        """Query target table columns directly via native driver."""
+        from src.core.connection_service import ConnectionService
+
+        service = ConnectionService(config_manager)
+        conn = service.get_connection(connection_id)
+        if not conn:
+            return None
+
+        safe_schema = schema.replace("'", "''")
+        safe_table = table.replace("'", "''")
+
+        # Dialect-specific schema query
+        if conn.dialect == "clickhouse":
+            sql = (
+                f"SELECT name, type FROM system.columns "
+                f"WHERE database = '{safe_schema}' AND table = '{safe_table}' "
+                f"ORDER BY position"
+            )
+            return self._fetch_ch(conn, sql)
+
+        if conn.dialect == "postgresql":
+            sql = (
+                f"SELECT column_name AS name, data_type AS type "
+                f"FROM information_schema.columns "
+                f"WHERE table_schema = '{safe_schema}' AND table_name = '{safe_table}' "
+                f"ORDER BY ordinal_position"
+            )
+            return self._fetch_pg(conn, sql)
+
+        if conn.dialect == "mysql":
+            sql = (
+                f"SELECT column_name AS name, data_type AS type "
+                f"FROM information_schema.columns "
+                f"WHERE table_schema = '{safe_schema}' AND table_name = '{safe_table}' "
+                f"ORDER BY ordinal_position"
+            )
+            return self._fetch_mysql(conn, sql)
+
+        return None
+
+    # #endregion fetch_schema
+
+    # #region _fetch_pg [C:2] [TYPE Function] [SEMANTICS database,postgresql,fetch,schema]
+    def _fetch_pg(self, conn: Any, sql: str) -> list[DbSchemaColumn] | None:
+        import asyncio
+
+        try:
+            import asyncpg  # type: ignore[import-untyped]
+        except ImportError:
+            return None
+
+        async def _run() -> list[DbSchemaColumn] | None:
+            pool_cfg = f"{conn.host}:{conn.port}/{conn.database}@{conn.updated_at}"
+            pool = self._get_or_create_pool(conn.id, pool_cfg)
+            if pool is None:
+                pool = await asyncpg.create_pool(
+                    host=conn.host, port=conn.port,
+                    user=conn.username, password=conn.password,
+                    database=conn.database, min_size=1, max_size=conn.pool_size,
+                )
+                self._set_pool(conn.id, pool, pool_cfg)
+            async with pool.acquire() as pg_conn:
+                rows = await pg_conn.fetch(sql)
+                return [DbSchemaColumn(name=r["name"], data_type=r["type"]) for r in rows]
+
+        return asyncio.run(_run())
+
+    # #endregion _fetch_pg
+
+    # #region _fetch_ch [C:2] [TYPE Function] [SEMANTICS database,clickhouse,fetch,schema]
+    def _fetch_ch(self, conn: Any, sql: str) -> list[DbSchemaColumn] | None:
+        try:
+            import clickhouse_connect  # type: ignore[import-untyped]
+        except ImportError:
+            return None
+
+        pool_cfg = f"{conn.host}:{conn.port}/{conn.database}@{conn.updated_at}"
+        client = self._get_or_create_pool(conn.id, pool_cfg)
+        if client is None:
+            client = clickhouse_connect.get_client(
+                host=conn.host, port=conn.port,
+                username=conn.username, password=conn.password,
+                database=conn.database, connect_timeout=30,
+            )
+            self._set_pool(conn.id, client, pool_cfg)
+
+        rows = client.query(sql).result_rows
+        return [DbSchemaColumn(name=r[0], data_type=r[1]) for r in rows] if rows else []
+
+    # #endregion _fetch_ch
+
+    # #region _fetch_mysql [C:2] [TYPE Function] [SEMANTICS database,mysql,fetch,schema]
+    def _fetch_mysql(self, conn: Any, sql: str) -> list[DbSchemaColumn] | None:
+        try:
+            import pymysql  # type: ignore[import-untyped]
+        except ImportError:
+            return None
+
+        pool_cfg = f"{conn.host}:{conn.port}/{conn.database}@{conn.updated_at}"
+        db = self._get_or_create_pool(conn.id, pool_cfg)
+        if db is None:
+            db = pymysql.connect(
+                host=conn.host, port=conn.port,
+                user=conn.username, password=conn.password,
+                database=conn.database, connect_timeout=30, autocommit=True,
+            )
+            self._set_pool(conn.id, db, pool_cfg)
+
+        with db.cursor() as cursor:
+            cursor.execute(sql)
+            rows = cursor.fetchall()
+            return [DbSchemaColumn(name=r[0], data_type=r[1]) for r in rows] if rows else []
+
+    # #endregion _fetch_mysql
+
+
+# #endregion DbExecutor
+# #endregion Core.DbExecutor

backend/src/scripts/seed_permissions.py

@@ -51,6 +51,8 @@ INITIAL_PERMISSIONS = [
     {"resource": "plugin:storage", "action": "WRITE"},
     {"resource": "plugin:debug", "action": "EXECUTE"},
     {"resource": "git_config", "action": "READ"},
+    # Connection Settings Permissions
+    {"resource": "settings.connections", "action": "MANAGE"},
     # Dataset Review Permissions
     {"resource": "dataset:session", "action": "READ"},
     {"resource": "dataset:session", "action": "MANAGE"},