feat: connect enterprise catalog ingestion

app/api/data_development/routes.py

@@ -27,6 +27,69 @@ def get_ingestion_service():
     )
 
 
+def get_database_source_registration_service():
+    from app.core.data_research.repository import (
+        SqlAlchemyIngestionSourceRepository,
+    )
+    from app.core.data_research.sources import (
+        DatabaseSourceRegistrationService,
+    )
+    from app.core.data_source.runtime import get_data_source_manager
+
+    manager = get_data_source_manager()
+    return DatabaseSourceRegistrationService(
+        SqlAlchemyIngestionSourceRepository(db.session),
+        definition_resolver=manager.definitions.get,
+        commit=db.session.commit,
+        rollback=db.session.rollback,
+    )
+
+
+def get_catalog_snapshot_repository():
+    from app.core.data_research.repository import (
+        SqlAlchemyCatalogSnapshotRepository,
+    )
+
+    return SqlAlchemyCatalogSnapshotRepository(db.session)
+
+
+def get_catalog_ingestion_executor():
+    from app.core.data_research.catalog.execution import (
+        CatalogIngestionExecutor,
+    )
+    from app.core.data_research.catalog.mysql import MySqlCatalogCollector
+    from app.core.data_research.catalog.postgresql import (
+        PostgreSqlCatalogCollector,
+    )
+    from app.core.data_research.catalog.service import CatalogCollectionService
+    from app.core.data_research.errors import IngestionSourceInvalid
+    from app.core.data_source.runtime import get_data_source_manager
+
+    manager = get_data_source_manager()
+
+    def collector_resolver(database_type):
+        if database_type == "postgresql":
+            return PostgreSqlCatalogCollector()
+        if database_type == "mysql":
+            return MySqlCatalogCollector()
+        raise IngestionSourceInvalid(
+            f"database type {database_type} is not supported"
+        )
+
+    collector = CatalogCollectionService(
+        manager,
+        definition_resolver=manager.definitions.get,
+        collector_resolver=collector_resolver,
+    )
+    return CatalogIngestionExecutor(
+        get_ingestion_service(),
+        collector,
+        get_catalog_snapshot_repository(),
+        commit=db.session.commit,
+        rollback=db.session.rollback,
+    )
+
+
 def get_data_element_service():
     from app.core.data_research.data_elements import DataElementService
     from app.core.data_research.repository import SqlAlchemyDataElementRepository
@@ -158,6 +221,8 @@ def _record(record):
         "parameters": dict(record.parameters or {}),
         "statistics": dict(record.statistics or {}),
         "last_error": record.last_error,
+        "attempt_count": int(record.attempt_count or 0),
+        "failure_stage": record.failure_stage,
         "actor_uid": record.actor_uid,
         "created_at": record.created_at.isoformat() if record.created_at else None,
         "updated_at": record.updated_at.isoformat() if record.updated_at else None,
@@ -166,6 +231,24 @@ def _record(record):
     }
 
 
+def _catalog_snapshot(record):
+    return {
+        "uid": str(record.uid),
+        "job_uid": str(record.job_uid),
+        "source_uid": str(record.source_uid),
+        "attempt": int(record.attempt),
+        "database_type": record.database_type,
+        "content_hash": record.content_hash,
+        "snapshot": dict(record.snapshot or {}),
+        "evidence_count": int(record.evidence_count),
+        "created_at": (
+            record.created_at.isoformat()
+            if record.created_at
+            else None
+        ),
+    }
+
+
 def _element(record):
     return {
         "uid": str(record.uid),
@@ -262,9 +345,15 @@ def _error(error):
 def create_ingestion_job():
     payload = request.get_json(silent=True) or {}
     try:
+        actor_uid = _identity().get("id") or _identity().get("sub")
+        if payload.get("job_type") == "catalog_collect":
+            get_database_source_registration_service().ensure(
+                payload.get("source_uid"),
+                actor_uid=actor_uid,
+            )
         record, created = get_ingestion_service().create_job(
             payload,
-            actor_uid=_identity().get("id") or _identity().get("sub"),
+            actor_uid=actor_uid,
         )
         return jsonify(success(_record(record))), 201 if created else 200
     except Exception as error:
@@ -295,6 +384,48 @@ def get_ingestion_job(job_uid):
         return _error(error)
 
 
+@bp.route("/ingestion-jobs/<job_uid>/execute", methods=["POST"])
+def execute_ingestion_job(job_uid):
+    try:
+        record = get_catalog_ingestion_executor().execute(job_uid)
+        return jsonify(success(_record(record))), 200
+    except Exception as error:
+        return _error(error)
+
+
+@bp.route(
+    "/ingestion-jobs/<job_uid>/catalog-snapshots",
+    methods=["GET"],
+)
+def list_catalog_snapshots(job_uid):
+    try:
+        records = get_catalog_snapshot_repository().list(job_uid)
+        return jsonify(
+            success(
+                {
+                    "records": [
+                        _catalog_snapshot(record)
+                        for record in records
+                    ],
+                    "total": len(records),
+                }
+            )
+        ), 200
+    except Exception as error:
+        return _error(error)
+
+
+@bp.route("/ingestion-jobs/<job_uid>/evidence", methods=["GET"])
+def list_ingestion_job_evidence(job_uid):
+    try:
+        records = get_evidence_service().list_for_job(job_uid)
+        return jsonify(
+            success({"records": records, "total": len(records)})
+        ), 200
+    except Exception as error:
+        return _error(error)
+
+
 @bp.route("/ingestion-jobs/<job_uid>/retry", methods=["POST"])
 def retry_ingestion_job(job_uid):
     try:

app/core/data_research/artifacts.py

@@ -71,6 +71,19 @@ class EvidenceService:
         data["excerpt"] = redact_excerpt(data.get("excerpt"))
         return data
 
+    def list_for_job(self, job_uid):
+        from app.models.data_research import EvidenceFragment
+
+        records = self.session.query(EvidenceFragment).filter_by(
+            job_uid=str(job_uid)
+        ).order_by(EvidenceFragment.created_at.asc()).all()
+        result = []
+        for model in records:
+            data = model.to_dict()
+            data["excerpt"] = redact_excerpt(data.get("excerpt"))
+            result.append(data)
+        return result
+
     def download(self, uid):
         from app.models.data_research import EvidenceFragment, SourceArtifact
 

app/core/data_research/catalog/base.py

@@ -9,6 +9,14 @@ from app.core.data_research.catalog.models import (
 )
 
 
+EMPTY_SCOPE_SENTINEL = "__dataops_empty_scope__"
+
+
+def expanding_text_values(values):
+    result = list(values)
+    return result or [EMPTY_SCOPE_SENTINEL]
+
+
 class BaseCatalogCollector:
     database_type = ""
 

app/core/data_research/catalog/execution.py

@@ -0,0 +1,127 @@
+from __future__ import annotations
+
+from app.core.data_research.errors import (
+    IngestionPayloadInvalid,
+    InvalidJobTransition,
+)
+from app.core.data_research.catalog.models import CatalogScope
+
+
+_SCOPE_FIELDS = (
+    "include_schemas",
+    "exclude_schemas",
+    "include_tables",
+    "exclude_tables",
+)
+
+
+def catalog_scope_from_parameters(parameters):
+    parameters = parameters or {}
+    if not isinstance(parameters, dict):
+        raise IngestionPayloadInvalid("parameters must be an object")
+    values = {}
+    for name in _SCOPE_FIELDS:
+        raw = parameters.get(name, [])
+        if not isinstance(raw, (list, tuple)):
+            raise IngestionPayloadInvalid(f"{name} must be a list")
+        if any(not isinstance(item, str) or not item.strip() for item in raw):
+            raise IngestionPayloadInvalid(
+                f"{name} must contain non-empty text values"
+            )
+        values[name] = tuple(item.strip() for item in raw)
+    return CatalogScope(**values)
+
+
+class CatalogIngestionExecutor:
+    def __init__(
+        self,
+        ingestion,
+        collector,
+        snapshots,
+        *,
+        commit=lambda: None,
+        rollback=lambda: None,
+    ):
+        self.ingestion = ingestion
+        self.collector = collector
+        self.snapshots = snapshots
+        self.commit = commit
+        self.rollback = rollback
+
+    @staticmethod
+    def _report(snapshot_record):
+        assets = tuple(snapshot_record.snapshot.get("assets") or ())
+        field_count = sum(
+            len(asset.get("fields") or ())
+            for asset in assets
+        )
+        return {
+            "snapshot_uid": snapshot_record.uid,
+            "content_hash": snapshot_record.content_hash,
+            "asset_count": len(assets),
+            "field_count": field_count,
+            "evidence_count": snapshot_record.evidence_count,
+        }
+
+    def execute(self, job_uid):
+        job = self.ingestion.get_job(job_uid)
+        if job.job_type != "catalog_collect":
+            raise IngestionPayloadInvalid(
+                "only catalog_collect jobs can use catalog execution"
+            )
+        scope = catalog_scope_from_parameters(job.parameters)
+        if job.status == "awaiting_review":
+            return job
+        if job.status == "created":
+            job = self.ingestion.transition(job.uid, "queued")
+        if job.status == "queued":
+            job = self.ingestion.transition(job.uid, "extracting")
+        if job.status not in {"extracting", "normalizing", "matching"}:
+            raise InvalidJobTransition(
+                f"{job.status} -> execute is not allowed"
+            )
+
+        try:
+            snapshot_record = self.snapshots.find(
+                job.uid,
+                job.attempt_count,
+            )
+            if job.status in {"extracting", "normalizing"} and snapshot_record is None:
+                snapshot = self.collector.collect(job.source_uid, scope)
+                if job.status == "extracting":
+                    job = self.ingestion.transition(job.uid, "normalizing")
+                snapshot_record = self.snapshots.persist(
+                    job.uid,
+                    job.attempt_count,
+                    snapshot,
+                )
+                self.commit()
+            if job.status == "normalizing":
+                job = self.ingestion.transition(job.uid, "matching")
+            if job.status == "matching":
+                if snapshot_record is None:
+                    raise RuntimeError(
+                        "catalog snapshot is missing for the current attempt"
+                    )
+                job = self.ingestion.transition(
+                    job.uid,
+                    "awaiting_review",
+                    statistics=self._report(snapshot_record),
+                )
+            return job
+        except Exception as error:
+            self.rollback()
+            current = self.ingestion.get_job(job.uid)
+            if current.status in {
+                "queued",
+                "extracting",
+                "normalizing",
+                "matching",
+            }:
+                self.ingestion.transition(
+                    current.uid,
+                    "failed",
+                    statistics=current.statistics,
+                    error=error,
+                )
+            raise

app/core/data_research/catalog/models.py

@@ -66,3 +66,16 @@ class CatalogDiff:
     removed: tuple[CatalogField, ...] = ()
     changed: tuple[CatalogFieldChange, ...] = ()
     renamed: tuple[CatalogFieldChange, ...] = ()
+
+
+@dataclass(frozen=True)
+class CatalogSnapshotRecord:
+    uid: str
+    job_uid: str
+    source_uid: str
+    attempt: int
+    database_type: str
+    content_hash: str
+    snapshot: dict[str, Any]
+    evidence_count: int
+    created_at: Any = None

app/core/data_research/catalog/mysql.py

@@ -2,7 +2,10 @@ from __future__ import annotations
 
 from sqlalchemy import bindparam, text
 
-from app.core.data_research.catalog.base import BaseCatalogCollector
+from app.core.data_research.catalog.base import (
+    BaseCatalogCollector,
+    expanding_text_values,
+)
 
 
 MYSQL_CATALOG_SQL = """
@@ -37,8 +40,8 @@ class MySqlCatalogCollector(BaseCatalogCollector):
         )
         parameters = {
             "database_name": str(database_name),
-            "include_tables": list(scope.include_tables),
-            "exclude_tables": list(scope.exclude_tables),
+            "include_tables": expanding_text_values(scope.include_tables),
+            "exclude_tables": expanding_text_values(scope.exclude_tables),
             "include_table_count": len(scope.include_tables),
             "exclude_table_count": len(scope.exclude_tables),
         }

app/core/data_research/catalog/postgresql.py

@@ -2,7 +2,10 @@ from __future__ import annotations
 
 from sqlalchemy import bindparam, text
 
-from app.core.data_research.catalog.base import BaseCatalogCollector
+from app.core.data_research.catalog.base import (
+    BaseCatalogCollector,
+    expanding_text_values,
+)
 
 
 POSTGRESQL_CATALOG_SQL = """
@@ -43,10 +46,10 @@ class PostgreSqlCatalogCollector(BaseCatalogCollector):
             bindparam("exclude_tables", expanding=True),
         )
         parameters = {
-            "include_schemas": list(scope.include_schemas),
-            "exclude_schemas": list(scope.exclude_schemas),
-            "include_tables": list(scope.include_tables),
-            "exclude_tables": list(scope.exclude_tables),
+            "include_schemas": expanding_text_values(scope.include_schemas),
+            "exclude_schemas": expanding_text_values(scope.exclude_schemas),
+            "include_tables": expanding_text_values(scope.include_tables),
+            "exclude_tables": expanding_text_values(scope.exclude_tables),
             "include_schema_count": len(scope.include_schemas),
             "exclude_schema_count": len(scope.exclude_schemas),
             "include_table_count": len(scope.include_tables),

app/core/data_research/errors.py

@@ -17,6 +17,11 @@ class IngestionJobNotFound(DataResearchError):
     http_status = 404
 
 
+class IngestionSourceInvalid(DataResearchError):
+    code = "INGESTION_SOURCE_INVALID"
+    http_status = 422
+
+
 class DataElementInvalid(DataResearchError):
     code = "DATA_ELEMENT_INVALID"
 

app/core/data_research/ingestion.py

@@ -157,6 +157,18 @@ class IngestionService:
             status=target_status,
             statistics=dict(statistics or record.statistics),
             last_error=_sanitize_error(error),
+            attempt_count=(
+                record.attempt_count + 1
+                if target_status == "extracting"
+                else record.attempt_count
+            ),
+            failure_stage=(
+                record.status
+                if target_status == "failed"
+                else None
+                if target_status == "queued"
+                else record.failure_stage
+            ),
             updated_at=now,
             started_at=(now if target_status == "extracting" and record.started_at is None else record.started_at),
             finished_at=(now if target_status in TERMINAL_STATUSES | {"failed", "partial"} else None),

app/core/data_research/models.py

@@ -28,6 +28,8 @@ class IngestionJobRecord:
     status: str = "created"
     statistics: dict[str, Any] = field(default_factory=dict)
     last_error: str | None = None
+    attempt_count: int = 0
+    failure_stage: str | None = None
     force_rerun: bool = False
     created_at: datetime | None = None
     updated_at: datetime | None = None

app/core/data_research/repository.py

@@ -1,16 +1,69 @@
 from __future__ import annotations
 
-from dataclasses import replace
+from dataclasses import asdict, replace
 
+from app.core.common.identifiers import new_governance_uid
 from app.models.data_research import (
+    CatalogSnapshot,
     CandidateDecisionRecord,
     DataElement,
     DataElementVersion,
+    EvidenceFragment,
     ExtractionCandidate,
     IngestionJob,
+    IngestionSource,
 )
 
 
+class SqlAlchemyIngestionSourceRepository:
+    def __init__(self, session):
+        self.session = session
+
+    @staticmethod
+    def _record(model):
+        from app.core.data_research.sources import IngestionSourceRecord
+
+        return IngestionSourceRecord(
+            uid=str(model.uid),
+            source_type=model.source_type,
+            name=model.name,
+            config=dict(model.config or {}),
+            permission_scope=dict(model.permission_scope or {}),
+            status=model.status,
+            created_by=model.created_by,
+            created_at=model.created_at,
+            updated_at=model.updated_at,
+        )
+
+    def get(self, uid):
+        model = self.session.get(IngestionSource, str(uid))
+        return self._record(model) if model is not None else None
+
+    def save(self, record):
+        model = self.session.get(IngestionSource, str(record.uid))
+        if model is None:
+            model = IngestionSource(
+                uid=record.uid,
+                source_type=record.source_type,
+                name=record.name,
+                config=record.config,
+                permission_scope=record.permission_scope,
+                status=record.status,
+                created_by=record.created_by,
+                created_at=record.created_at,
+                updated_at=record.updated_at,
+            )
+            self.session.add(model)
+        else:
+            model.name = record.name
+            model.config = record.config
+            model.permission_scope = record.permission_scope
+            model.status = record.status
+            model.updated_at = record.updated_at
+        self.session.flush()
+        return self._record(model)
+
+
 class SqlAlchemyIngestionJobRepository:
     def __init__(self, session):
         self.session = session
@@ -31,6 +84,8 @@ class SqlAlchemyIngestionJobRepository:
             status=model.status,
             statistics=dict(model.statistics or {}),
             last_error=model.last_error,
+            attempt_count=int(model.attempt_count or 0),
+            failure_stage=model.failure_stage,
             force_rerun=bool(model.force_rerun),
             created_at=model.created_at,
             updated_at=model.updated_at,
@@ -74,6 +129,8 @@ class SqlAlchemyIngestionJobRepository:
             parameters=record.parameters,
             statistics=record.statistics,
             last_error=record.last_error,
+            attempt_count=record.attempt_count,
+            failure_stage=record.failure_stage,
             actor_uid=record.actor_uid,
             force_rerun=record.force_rerun,
             started_at=record.started_at,
@@ -91,6 +148,8 @@ class SqlAlchemyIngestionJobRepository:
             "status",
             "statistics",
             "last_error",
+            "attempt_count",
+            "failure_stage",
             "started_at",
             "finished_at",
             "updated_at",
@@ -100,6 +159,88 @@ class SqlAlchemyIngestionJobRepository:
         return self._record(model)
 
 
+class SqlAlchemyCatalogSnapshotRepository:
+    def __init__(self, session, *, uid_factory=new_governance_uid):
+        self.session = session
+        self.uid_factory = uid_factory
+
+    def _record(self, model):
+        from app.core.data_research.catalog.models import CatalogSnapshotRecord
+
+        evidence_count = self.session.query(EvidenceFragment).filter_by(
+            job_uid=str(model.job_uid)
+        ).filter(
+            EvidenceFragment.locator["attempt"].astext
+            == str(model.attempt)
+        ).count()
+        return CatalogSnapshotRecord(
+            uid=str(model.uid),
+            job_uid=str(model.job_uid),
+            source_uid=str(model.source_uid),
+            attempt=int(model.attempt),
+            database_type=model.database_type,
+            content_hash=model.content_hash,
+            snapshot=dict(model.snapshot or {}),
+            evidence_count=int(evidence_count),
+            created_at=model.created_at,
+        )
+
+    def find(self, job_uid, attempt):
+        model = self.session.query(CatalogSnapshot).filter_by(
+            job_uid=str(job_uid),
+            attempt=int(attempt),
+        ).first()
+        return self._record(model) if model is not None else None
+
+    def persist(self, job_uid, attempt, snapshot):
+        existing = self.find(job_uid, attempt)
+        if existing is not None:
+            return existing
+        model = CatalogSnapshot(
+            uid=self.uid_factory(),
+            job_uid=str(job_uid),
+            source_uid=str(snapshot.data_source_uid),
+            attempt=int(attempt),
+            database_type=snapshot.database_type,
+            content_hash=snapshot.content_hash,
+            snapshot=asdict(snapshot),
+        )
+        self.session.add(model)
+        for asset in snapshot.assets:
+            for field in asset.fields:
+                self.session.add(
+                    EvidenceFragment(
+                        uid=self.uid_factory(),
+                        job_uid=str(job_uid),
+                        locator={
+                            "kind": "database.column",
+                            "data_source_uid": str(snapshot.data_source_uid),
+                            "database_type": snapshot.database_type,
+                            "schema": field.schema,
+                            "table": field.asset,
+                            "column": field.name,
+                            "ordinal_position": int(field.ordinal_position),
+                            "attempt": int(attempt),
+                        },
+                        excerpt=(
+                            f"{field.schema}.{field.asset}.{field.name} "
+                            f"{field.data_type}"
+                        ),
+                        confidence=1.0,
+                    )
+                )
+        self.session.flush()
+        return self._record(model)
+
+    def list(self, job_uid):
+        return [
+            self._record(model)
+            for model in self.session.query(CatalogSnapshot).filter_by(
+                job_uid=str(job_uid)
+            ).order_by(CatalogSnapshot.attempt.desc()).all()
+        ]
+
+
 class SqlAlchemyDataElementRepository:
     def __init__(self, session):
         self.session = session

app/core/data_research/sources.py

@@ -0,0 +1,96 @@
+from __future__ import annotations
+
+from dataclasses import dataclass, replace
+from typing import Any, Callable
+
+from app.core.common.timezone_utils import now_china_naive
+from app.core.data_research.errors import IngestionSourceInvalid
+
+
+@dataclass(frozen=True)
+class IngestionSourceRecord:
+    uid: str
+    source_type: str
+    name: str
+    config: dict[str, Any]
+    permission_scope: dict[str, Any]
+    status: str
+    created_by: str | None
+    created_at: Any = None
+    updated_at: Any = None
+
+
+class DatabaseSourceRegistrationService:
+    SUPPORTED_DATABASES = frozenset({"postgresql", "mysql"})
+
+    def __init__(
+        self,
+        repository,
+        *,
+        definition_resolver: Callable[[str], Any],
+        clock: Callable[[], Any] = now_china_naive,
+        commit: Callable[[], Any] = lambda: None,
+        rollback: Callable[[], Any] = lambda: None,
+    ):
+        self.repository = repository
+        self.definition_resolver = definition_resolver
+        self.clock = clock
+        self.commit = commit
+        self.rollback = rollback
+
+    def ensure(self, data_source_uid, actor_uid):
+        uid = str(data_source_uid or "").strip()
+        definition = self.definition_resolver(uid) if uid else None
+        if definition is None:
+            raise IngestionSourceInvalid("data source was not found")
+        if not bool(definition.status):
+            raise IngestionSourceInvalid("data source is disabled")
+        database_type = str(definition.database_type or "").strip().lower()
+        if database_type not in self.SUPPORTED_DATABASES:
+            raise IngestionSourceInvalid(
+                f"database type {database_type or 'unknown'} is not supported"
+            )
+
+        now = self.clock()
+        existing = self.repository.get(uid)
+        if existing is not None and existing.source_type != "database":
+            raise IngestionSourceInvalid(
+                "data source UID is already used by another source type"
+            )
+        config = {
+            "database_type": database_type,
+            "database": str(definition.database or ""),
+            "schema": str(definition.schema or "") or None,
+        }
+        name = str(
+            definition.name_zh or definition.name_en or uid
+        ).strip()
+        if existing is None:
+            record = IngestionSourceRecord(
+                uid=uid,
+                source_type="database",
+                name=name,
+                config=config,
+                permission_scope={},
+                status="active",
+                created_by=actor_uid,
+                created_at=now,
+                updated_at=now,
+            )
+            created = True
+        else:
+            record = replace(
+                existing,
+                name=name,
+                config=config,
+                status="active",
+                updated_at=now,
+            )
+            created = False
+        try:
+            saved = self.repository.save(record)
+            self.commit()
+            return saved, created
+        except Exception:
+            self.rollback()
+            raise

app/core/data_source/service.py

@@ -222,11 +222,19 @@ class DataSourceService:
 
     def list(self, payload):
         payload = payload if isinstance(payload, dict) else {}
+
+        def optional_text(name):
+            value = payload.get(name)
+            if value is None:
+                return None
+            normalized = str(value).strip()
+            return normalized or None
+
         filters = DataSourceFilters(
-            uid=payload.get("uid"),
-            name_en=payload.get("name_en"),
-            name_zh=payload.get("name_zh"),
-            database_type=payload.get("type"),
+            uid=optional_text("uid"),
+            name_en=optional_text("name_en"),
+            name_zh=optional_text("name_zh"),
+            database_type=optional_text("type"),
             status=payload.get("status"),
         )
         return self.definitions.list(filters)

app/models/data_research.py

@@ -115,6 +115,10 @@ class IngestionJob(db.Model):
             "idempotency_key",
             name="uq_ingestion_job_idempotency_key",
         ),
+        db.CheckConstraint(
+            "attempt_count >= 0",
+            name="ck_ingestion_job_attempt_count",
+        ),
         {"schema": "public"},
     )
 
@@ -135,6 +139,8 @@ class IngestionJob(db.Model):
     parameters = db.Column(JSONB, nullable=False, default=dict)
     statistics = db.Column(JSONB, nullable=False, default=dict)
     last_error = db.Column(db.String(1000))
+    attempt_count = db.Column(db.Integer, nullable=False, default=0)
+    failure_stage = db.Column(db.String(30))
     actor_uid = db.Column(db.String(100))
     force_rerun = db.Column(db.Boolean, nullable=False, default=False)
     created_at = db.Column(db.DateTime, nullable=False, default=now_china_naive)
@@ -153,6 +159,8 @@ class IngestionJob(db.Model):
             "parameters": dict(self.parameters or {}),
             "statistics": dict(self.statistics or {}),
             "last_error": self.last_error,
+            "attempt_count": int(self.attempt_count or 0),
+            "failure_stage": self.failure_stage,
             "actor_uid": self.actor_uid,
             "created_at": _iso(self.created_at),
             "updated_at": _iso(self.updated_at),
@@ -161,6 +169,59 @@ class IngestionJob(db.Model):
         }
 
 
+class CatalogSnapshot(db.Model):
+    __tablename__ = "catalog_snapshots"
+    __table_args__ = (
+        db.CheckConstraint(
+            "attempt > 0",
+            name="ck_catalog_snapshot_attempt",
+        ),
+        db.UniqueConstraint(
+            "job_uid",
+            "attempt",
+            name="uq_catalog_snapshot_job_attempt",
+        ),
+        {"schema": "public"},
+    )
+
+    uid = db.Column(
+        UUID(as_uuid=False),
+        primary_key=True,
+        default=new_governance_uid,
+    )
+    job_uid = db.Column(
+        UUID(as_uuid=False),
+        db.ForeignKey("public.ingestion_jobs.uid", ondelete="CASCADE"),
+        nullable=False,
+    )
+    source_uid = db.Column(
+        UUID(as_uuid=False),
+        db.ForeignKey("public.ingestion_sources.uid"),
+        nullable=False,
+    )
+    attempt = db.Column(db.Integer, nullable=False)
+    database_type = db.Column(db.String(20), nullable=False)
+    content_hash = db.Column(db.String(64), nullable=False)
+    snapshot = db.Column(JSONB, nullable=False)
+    created_at = db.Column(
+        db.DateTime,
+        nullable=False,
+        default=now_china_naive,
+    )
+
+    def to_dict(self) -> dict[str, Any]:
+        return {
+            "uid": str(self.uid),
+            "job_uid": str(self.job_uid),
+            "source_uid": str(self.source_uid),
+            "attempt": int(self.attempt),
+            "database_type": self.database_type,
+            "content_hash": self.content_hash,
+            "snapshot": dict(self.snapshot or {}),
+            "created_at": _iso(self.created_at),
+        }
+
+
 class EvidenceFragment(db.Model):
     __tablename__ = "evidence_fragments"
     __table_args__ = (

deployment/app/api/data_development/routes.py

@@ -27,6 +27,69 @@ def get_ingestion_service():
     )
 
 
+def get_database_source_registration_service():
+    from app.core.data_research.repository import (
+        SqlAlchemyIngestionSourceRepository,
+    )
+    from app.core.data_research.sources import (
+        DatabaseSourceRegistrationService,
+    )
+    from app.core.data_source.runtime import get_data_source_manager
+
+    manager = get_data_source_manager()
+    return DatabaseSourceRegistrationService(
+        SqlAlchemyIngestionSourceRepository(db.session),
+        definition_resolver=manager.definitions.get,
+        commit=db.session.commit,
+        rollback=db.session.rollback,
+    )
+
+
+def get_catalog_snapshot_repository():
+    from app.core.data_research.repository import (
+        SqlAlchemyCatalogSnapshotRepository,
+    )
+
+    return SqlAlchemyCatalogSnapshotRepository(db.session)
+
+
+def get_catalog_ingestion_executor():
+    from app.core.data_research.catalog.execution import (
+        CatalogIngestionExecutor,
+    )
+    from app.core.data_research.catalog.mysql import MySqlCatalogCollector
+    from app.core.data_research.catalog.postgresql import (
+        PostgreSqlCatalogCollector,
+    )
+    from app.core.data_research.catalog.service import CatalogCollectionService
+    from app.core.data_research.errors import IngestionSourceInvalid
+    from app.core.data_source.runtime import get_data_source_manager
+
+    manager = get_data_source_manager()
+
+    def collector_resolver(database_type):
+        if database_type == "postgresql":
+            return PostgreSqlCatalogCollector()
+        if database_type == "mysql":
+            return MySqlCatalogCollector()
+        raise IngestionSourceInvalid(
+            f"database type {database_type} is not supported"
+        )
+
+    collector = CatalogCollectionService(
+        manager,
+        definition_resolver=manager.definitions.get,
+        collector_resolver=collector_resolver,
+    )
+    return CatalogIngestionExecutor(
+        get_ingestion_service(),
+        collector,
+        get_catalog_snapshot_repository(),
+        commit=db.session.commit,
+        rollback=db.session.rollback,
+    )
+
+
 def get_data_element_service():
     from app.core.data_research.data_elements import DataElementService
     from app.core.data_research.repository import SqlAlchemyDataElementRepository
@@ -158,6 +221,8 @@ def _record(record):
         "parameters": dict(record.parameters or {}),
         "statistics": dict(record.statistics or {}),
         "last_error": record.last_error,
+        "attempt_count": int(record.attempt_count or 0),
+        "failure_stage": record.failure_stage,
         "actor_uid": record.actor_uid,
         "created_at": record.created_at.isoformat() if record.created_at else None,
         "updated_at": record.updated_at.isoformat() if record.updated_at else None,
@@ -166,6 +231,24 @@ def _record(record):
     }
 
 
+def _catalog_snapshot(record):
+    return {
+        "uid": str(record.uid),
+        "job_uid": str(record.job_uid),
+        "source_uid": str(record.source_uid),
+        "attempt": int(record.attempt),
+        "database_type": record.database_type,
+        "content_hash": record.content_hash,
+        "snapshot": dict(record.snapshot or {}),
+        "evidence_count": int(record.evidence_count),
+        "created_at": (
+            record.created_at.isoformat()
+            if record.created_at
+            else None
+        ),
+    }
+
+
 def _element(record):
     return {
         "uid": str(record.uid),
@@ -262,9 +345,15 @@ def _error(error):
 def create_ingestion_job():
     payload = request.get_json(silent=True) or {}
     try:
+        actor_uid = _identity().get("id") or _identity().get("sub")
+        if payload.get("job_type") == "catalog_collect":
+            get_database_source_registration_service().ensure(
+                payload.get("source_uid"),
+                actor_uid=actor_uid,
+            )
         record, created = get_ingestion_service().create_job(
             payload,
-            actor_uid=_identity().get("id") or _identity().get("sub"),
+            actor_uid=actor_uid,
         )
         return jsonify(success(_record(record))), 201 if created else 200
     except Exception as error:
@@ -295,6 +384,48 @@ def get_ingestion_job(job_uid):
         return _error(error)
 
 
+@bp.route("/ingestion-jobs/<job_uid>/execute", methods=["POST"])
+def execute_ingestion_job(job_uid):
+    try:
+        record = get_catalog_ingestion_executor().execute(job_uid)
+        return jsonify(success(_record(record))), 200
+    except Exception as error:
+        return _error(error)
+
+
+@bp.route(
+    "/ingestion-jobs/<job_uid>/catalog-snapshots",
+    methods=["GET"],
+)
+def list_catalog_snapshots(job_uid):
+    try:
+        records = get_catalog_snapshot_repository().list(job_uid)
+        return jsonify(
+            success(
+                {
+                    "records": [
+                        _catalog_snapshot(record)
+                        for record in records
+                    ],
+                    "total": len(records),
+                }
+            )
+        ), 200
+    except Exception as error:
+        return _error(error)
+
+
+@bp.route("/ingestion-jobs/<job_uid>/evidence", methods=["GET"])
+def list_ingestion_job_evidence(job_uid):
+    try:
+        records = get_evidence_service().list_for_job(job_uid)
+        return jsonify(
+            success({"records": records, "total": len(records)})
+        ), 200
+    except Exception as error:
+        return _error(error)
+
+
 @bp.route("/ingestion-jobs/<job_uid>/retry", methods=["POST"])
 def retry_ingestion_job(job_uid):
     try:

deployment/app/core/data_research/artifacts.py

@@ -71,6 +71,19 @@ class EvidenceService:
         data["excerpt"] = redact_excerpt(data.get("excerpt"))
         return data
 
+    def list_for_job(self, job_uid):
+        from app.models.data_research import EvidenceFragment
+
+        records = self.session.query(EvidenceFragment).filter_by(
+            job_uid=str(job_uid)
+        ).order_by(EvidenceFragment.created_at.asc()).all()
+        result = []
+        for model in records:
+            data = model.to_dict()
+            data["excerpt"] = redact_excerpt(data.get("excerpt"))
+            result.append(data)
+        return result
+
     def download(self, uid):
         from app.models.data_research import EvidenceFragment, SourceArtifact
 

deployment/app/core/data_research/catalog/base.py

@@ -9,6 +9,14 @@ from app.core.data_research.catalog.models import (
 )
 
 
+EMPTY_SCOPE_SENTINEL = "__dataops_empty_scope__"
+
+
+def expanding_text_values(values):
+    result = list(values)
+    return result or [EMPTY_SCOPE_SENTINEL]
+
+
 class BaseCatalogCollector:
     database_type = ""
 

deployment/app/core/data_research/catalog/execution.py

@@ -0,0 +1,127 @@
+from __future__ import annotations
+
+from app.core.data_research.errors import (
+    IngestionPayloadInvalid,
+    InvalidJobTransition,
+)
+from app.core.data_research.catalog.models import CatalogScope
+
+
+_SCOPE_FIELDS = (
+    "include_schemas",
+    "exclude_schemas",
+    "include_tables",
+    "exclude_tables",
+)
+
+
+def catalog_scope_from_parameters(parameters):
+    parameters = parameters or {}
+    if not isinstance(parameters, dict):
+        raise IngestionPayloadInvalid("parameters must be an object")
+    values = {}
+    for name in _SCOPE_FIELDS:
+        raw = parameters.get(name, [])
+        if not isinstance(raw, (list, tuple)):
+            raise IngestionPayloadInvalid(f"{name} must be a list")
+        if any(not isinstance(item, str) or not item.strip() for item in raw):
+            raise IngestionPayloadInvalid(
+                f"{name} must contain non-empty text values"
+            )
+        values[name] = tuple(item.strip() for item in raw)
+    return CatalogScope(**values)
+
+
+class CatalogIngestionExecutor:
+    def __init__(
+        self,
+        ingestion,
+        collector,
+        snapshots,
+        *,
+        commit=lambda: None,
+        rollback=lambda: None,
+    ):
+        self.ingestion = ingestion
+        self.collector = collector
+        self.snapshots = snapshots
+        self.commit = commit
+        self.rollback = rollback
+
+    @staticmethod
+    def _report(snapshot_record):
+        assets = tuple(snapshot_record.snapshot.get("assets") or ())
+        field_count = sum(
+            len(asset.get("fields") or ())
+            for asset in assets
+        )
+        return {
+            "snapshot_uid": snapshot_record.uid,
+            "content_hash": snapshot_record.content_hash,
+            "asset_count": len(assets),
+            "field_count": field_count,
+            "evidence_count": snapshot_record.evidence_count,
+        }
+
+    def execute(self, job_uid):
+        job = self.ingestion.get_job(job_uid)
+        if job.job_type != "catalog_collect":
+            raise IngestionPayloadInvalid(
+                "only catalog_collect jobs can use catalog execution"
+            )
+        scope = catalog_scope_from_parameters(job.parameters)
+        if job.status == "awaiting_review":
+            return job
+        if job.status == "created":
+            job = self.ingestion.transition(job.uid, "queued")
+        if job.status == "queued":
+            job = self.ingestion.transition(job.uid, "extracting")
+        if job.status not in {"extracting", "normalizing", "matching"}:
+            raise InvalidJobTransition(
+                f"{job.status} -> execute is not allowed"
+            )
+
+        try:
+            snapshot_record = self.snapshots.find(
+                job.uid,
+                job.attempt_count,
+            )
+            if job.status in {"extracting", "normalizing"} and snapshot_record is None:
+                snapshot = self.collector.collect(job.source_uid, scope)
+                if job.status == "extracting":
+                    job = self.ingestion.transition(job.uid, "normalizing")
+                snapshot_record = self.snapshots.persist(
+                    job.uid,
+                    job.attempt_count,
+                    snapshot,
+                )
+                self.commit()
+            if job.status == "normalizing":
+                job = self.ingestion.transition(job.uid, "matching")
+            if job.status == "matching":
+                if snapshot_record is None:
+                    raise RuntimeError(
+                        "catalog snapshot is missing for the current attempt"
+                    )
+                job = self.ingestion.transition(
+                    job.uid,
+                    "awaiting_review",
+                    statistics=self._report(snapshot_record),
+                )
+            return job
+        except Exception as error:
+            self.rollback()
+            current = self.ingestion.get_job(job.uid)
+            if current.status in {
+                "queued",
+                "extracting",
+                "normalizing",
+                "matching",
+            }:
+                self.ingestion.transition(
+                    current.uid,
+                    "failed",
+                    statistics=current.statistics,
+                    error=error,
+                )
+            raise

deployment/app/core/data_research/catalog/models.py

@@ -66,3 +66,16 @@ class CatalogDiff:
     removed: tuple[CatalogField, ...] = ()
     changed: tuple[CatalogFieldChange, ...] = ()
     renamed: tuple[CatalogFieldChange, ...] = ()
+
+
+@dataclass(frozen=True)
+class CatalogSnapshotRecord:
+    uid: str
+    job_uid: str
+    source_uid: str
+    attempt: int
+    database_type: str
+    content_hash: str
+    snapshot: dict[str, Any]
+    evidence_count: int
+    created_at: Any = None

deployment/app/core/data_research/catalog/mysql.py

@@ -2,7 +2,10 @@ from __future__ import annotations
 
 from sqlalchemy import bindparam, text
 
-from app.core.data_research.catalog.base import BaseCatalogCollector
+from app.core.data_research.catalog.base import (
+    BaseCatalogCollector,
+    expanding_text_values,
+)
 
 
 MYSQL_CATALOG_SQL = """
@@ -37,8 +40,8 @@ class MySqlCatalogCollector(BaseCatalogCollector):
         )
         parameters = {
             "database_name": str(database_name),
-            "include_tables": list(scope.include_tables),
-            "exclude_tables": list(scope.exclude_tables),
+            "include_tables": expanding_text_values(scope.include_tables),
+            "exclude_tables": expanding_text_values(scope.exclude_tables),
             "include_table_count": len(scope.include_tables),
             "exclude_table_count": len(scope.exclude_tables),
         }

deployment/app/core/data_research/catalog/postgresql.py

@@ -2,7 +2,10 @@ from __future__ import annotations
 
 from sqlalchemy import bindparam, text
 
-from app.core.data_research.catalog.base import BaseCatalogCollector
+from app.core.data_research.catalog.base import (
+    BaseCatalogCollector,
+    expanding_text_values,
+)
 
 
 POSTGRESQL_CATALOG_SQL = """
@@ -43,10 +46,10 @@ class PostgreSqlCatalogCollector(BaseCatalogCollector):
             bindparam("exclude_tables", expanding=True),
         )
         parameters = {
-            "include_schemas": list(scope.include_schemas),
-            "exclude_schemas": list(scope.exclude_schemas),
-            "include_tables": list(scope.include_tables),
-            "exclude_tables": list(scope.exclude_tables),
+            "include_schemas": expanding_text_values(scope.include_schemas),
+            "exclude_schemas": expanding_text_values(scope.exclude_schemas),
+            "include_tables": expanding_text_values(scope.include_tables),
+            "exclude_tables": expanding_text_values(scope.exclude_tables),
             "include_schema_count": len(scope.include_schemas),
             "exclude_schema_count": len(scope.exclude_schemas),
             "include_table_count": len(scope.include_tables),

deployment/app/core/data_research/errors.py

@@ -17,6 +17,11 @@ class IngestionJobNotFound(DataResearchError):
     http_status = 404
 
 
+class IngestionSourceInvalid(DataResearchError):
+    code = "INGESTION_SOURCE_INVALID"
+    http_status = 422
+
+
 class DataElementInvalid(DataResearchError):
     code = "DATA_ELEMENT_INVALID"
 

deployment/app/core/data_research/ingestion.py

@@ -157,6 +157,18 @@ class IngestionService:
             status=target_status,
             statistics=dict(statistics or record.statistics),
             last_error=_sanitize_error(error),
+            attempt_count=(
+                record.attempt_count + 1
+                if target_status == "extracting"
+                else record.attempt_count
+            ),
+            failure_stage=(
+                record.status
+                if target_status == "failed"
+                else None
+                if target_status == "queued"
+                else record.failure_stage
+            ),
             updated_at=now,
             started_at=(now if target_status == "extracting" and record.started_at is None else record.started_at),
             finished_at=(now if target_status in TERMINAL_STATUSES | {"failed", "partial"} else None),

deployment/app/core/data_research/models.py

@@ -28,6 +28,8 @@ class IngestionJobRecord:
     status: str = "created"
     statistics: dict[str, Any] = field(default_factory=dict)
     last_error: str | None = None
+    attempt_count: int = 0
+    failure_stage: str | None = None
     force_rerun: bool = False
     created_at: datetime | None = None
     updated_at: datetime | None = None

deployment/app/core/data_research/repository.py

@@ -1,16 +1,69 @@
 from __future__ import annotations
 
-from dataclasses import replace
+from dataclasses import asdict, replace
 
+from app.core.common.identifiers import new_governance_uid
 from app.models.data_research import (
+    CatalogSnapshot,
     CandidateDecisionRecord,
     DataElement,
     DataElementVersion,
+    EvidenceFragment,
     ExtractionCandidate,
     IngestionJob,
+    IngestionSource,
 )
 
 
+class SqlAlchemyIngestionSourceRepository:
+    def __init__(self, session):
+        self.session = session
+
+    @staticmethod
+    def _record(model):
+        from app.core.data_research.sources import IngestionSourceRecord
+
+        return IngestionSourceRecord(
+            uid=str(model.uid),
+            source_type=model.source_type,
+            name=model.name,
+            config=dict(model.config or {}),
+            permission_scope=dict(model.permission_scope or {}),
+            status=model.status,
+            created_by=model.created_by,
+            created_at=model.created_at,
+            updated_at=model.updated_at,
+        )
+
+    def get(self, uid):
+        model = self.session.get(IngestionSource, str(uid))
+        return self._record(model) if model is not None else None
+
+    def save(self, record):
+        model = self.session.get(IngestionSource, str(record.uid))
+        if model is None:
+            model = IngestionSource(
+                uid=record.uid,
+                source_type=record.source_type,
+                name=record.name,
+                config=record.config,
+                permission_scope=record.permission_scope,
+                status=record.status,
+                created_by=record.created_by,
+                created_at=record.created_at,
+                updated_at=record.updated_at,
+            )
+            self.session.add(model)
+        else:
+            model.name = record.name
+            model.config = record.config
+            model.permission_scope = record.permission_scope
+            model.status = record.status
+            model.updated_at = record.updated_at
+        self.session.flush()
+        return self._record(model)
+
+
 class SqlAlchemyIngestionJobRepository:
     def __init__(self, session):
         self.session = session
@@ -31,6 +84,8 @@ class SqlAlchemyIngestionJobRepository:
             status=model.status,
             statistics=dict(model.statistics or {}),
             last_error=model.last_error,
+            attempt_count=int(model.attempt_count or 0),
+            failure_stage=model.failure_stage,
             force_rerun=bool(model.force_rerun),
             created_at=model.created_at,
             updated_at=model.updated_at,
@@ -74,6 +129,8 @@ class SqlAlchemyIngestionJobRepository:
             parameters=record.parameters,
             statistics=record.statistics,
             last_error=record.last_error,
+            attempt_count=record.attempt_count,
+            failure_stage=record.failure_stage,
             actor_uid=record.actor_uid,
             force_rerun=record.force_rerun,
             started_at=record.started_at,
@@ -91,6 +148,8 @@ class SqlAlchemyIngestionJobRepository:
             "status",
             "statistics",
             "last_error",
+            "attempt_count",
+            "failure_stage",
             "started_at",
             "finished_at",
             "updated_at",
@@ -100,6 +159,88 @@ class SqlAlchemyIngestionJobRepository:
         return self._record(model)
 
 
+class SqlAlchemyCatalogSnapshotRepository:
+    def __init__(self, session, *, uid_factory=new_governance_uid):
+        self.session = session
+        self.uid_factory = uid_factory
+
+    def _record(self, model):
+        from app.core.data_research.catalog.models import CatalogSnapshotRecord
+
+        evidence_count = self.session.query(EvidenceFragment).filter_by(
+            job_uid=str(model.job_uid)
+        ).filter(
+            EvidenceFragment.locator["attempt"].astext
+            == str(model.attempt)
+        ).count()
+        return CatalogSnapshotRecord(
+            uid=str(model.uid),
+            job_uid=str(model.job_uid),
+            source_uid=str(model.source_uid),
+            attempt=int(model.attempt),
+            database_type=model.database_type,
+            content_hash=model.content_hash,
+            snapshot=dict(model.snapshot or {}),
+            evidence_count=int(evidence_count),
+            created_at=model.created_at,
+        )
+
+    def find(self, job_uid, attempt):
+        model = self.session.query(CatalogSnapshot).filter_by(
+            job_uid=str(job_uid),
+            attempt=int(attempt),
+        ).first()
+        return self._record(model) if model is not None else None
+
+    def persist(self, job_uid, attempt, snapshot):
+        existing = self.find(job_uid, attempt)
+        if existing is not None:
+            return existing
+        model = CatalogSnapshot(
+            uid=self.uid_factory(),
+            job_uid=str(job_uid),
+            source_uid=str(snapshot.data_source_uid),
+            attempt=int(attempt),
+            database_type=snapshot.database_type,
+            content_hash=snapshot.content_hash,
+            snapshot=asdict(snapshot),
+        )
+        self.session.add(model)
+        for asset in snapshot.assets:
+            for field in asset.fields:
+                self.session.add(
+                    EvidenceFragment(
+                        uid=self.uid_factory(),
+                        job_uid=str(job_uid),
+                        locator={
+                            "kind": "database.column",
+                            "data_source_uid": str(snapshot.data_source_uid),
+                            "database_type": snapshot.database_type,
+                            "schema": field.schema,
+                            "table": field.asset,
+                            "column": field.name,
+                            "ordinal_position": int(field.ordinal_position),
+                            "attempt": int(attempt),
+                        },
+                        excerpt=(
+                            f"{field.schema}.{field.asset}.{field.name} "
+                            f"{field.data_type}"
+                        ),
+                        confidence=1.0,
+                    )
+                )
+        self.session.flush()
+        return self._record(model)
+
+    def list(self, job_uid):
+        return [
+            self._record(model)
+            for model in self.session.query(CatalogSnapshot).filter_by(
+                job_uid=str(job_uid)
+            ).order_by(CatalogSnapshot.attempt.desc()).all()
+        ]
+
+
 class SqlAlchemyDataElementRepository:
     def __init__(self, session):
         self.session = session

deployment/app/core/data_research/sources.py

@@ -0,0 +1,96 @@
+from __future__ import annotations
+
+from dataclasses import dataclass, replace
+from typing import Any, Callable
+
+from app.core.common.timezone_utils import now_china_naive
+from app.core.data_research.errors import IngestionSourceInvalid
+
+
+@dataclass(frozen=True)
+class IngestionSourceRecord:
+    uid: str
+    source_type: str
+    name: str
+    config: dict[str, Any]
+    permission_scope: dict[str, Any]
+    status: str
+    created_by: str | None
+    created_at: Any = None
+    updated_at: Any = None
+
+
+class DatabaseSourceRegistrationService:
+    SUPPORTED_DATABASES = frozenset({"postgresql", "mysql"})
+
+    def __init__(
+        self,
+        repository,
+        *,
+        definition_resolver: Callable[[str], Any],
+        clock: Callable[[], Any] = now_china_naive,
+        commit: Callable[[], Any] = lambda: None,
+        rollback: Callable[[], Any] = lambda: None,
+    ):
+        self.repository = repository
+        self.definition_resolver = definition_resolver
+        self.clock = clock
+        self.commit = commit
+        self.rollback = rollback
+
+    def ensure(self, data_source_uid, actor_uid):
+        uid = str(data_source_uid or "").strip()
+        definition = self.definition_resolver(uid) if uid else None
+        if definition is None:
+            raise IngestionSourceInvalid("data source was not found")
+        if not bool(definition.status):
+            raise IngestionSourceInvalid("data source is disabled")
+        database_type = str(definition.database_type or "").strip().lower()
+        if database_type not in self.SUPPORTED_DATABASES:
+            raise IngestionSourceInvalid(
+                f"database type {database_type or 'unknown'} is not supported"
+            )
+
+        now = self.clock()
+        existing = self.repository.get(uid)
+        if existing is not None and existing.source_type != "database":
+            raise IngestionSourceInvalid(
+                "data source UID is already used by another source type"
+            )
+        config = {
+            "database_type": database_type,
+            "database": str(definition.database or ""),
+            "schema": str(definition.schema or "") or None,
+        }
+        name = str(
+            definition.name_zh or definition.name_en or uid
+        ).strip()
+        if existing is None:
+            record = IngestionSourceRecord(
+                uid=uid,
+                source_type="database",
+                name=name,
+                config=config,
+                permission_scope={},
+                status="active",
+                created_by=actor_uid,
+                created_at=now,
+                updated_at=now,
+            )
+            created = True
+        else:
+            record = replace(
+                existing,
+                name=name,
+                config=config,
+                status="active",
+                updated_at=now,
+            )
+            created = False
+        try:
+            saved = self.repository.save(record)
+            self.commit()
+            return saved, created
+        except Exception:
+            self.rollback()
+            raise

deployment/app/core/data_source/service.py

@@ -222,11 +222,19 @@ class DataSourceService:
 
     def list(self, payload):
         payload = payload if isinstance(payload, dict) else {}
+
+        def optional_text(name):
+            value = payload.get(name)
+            if value is None:
+                return None
+            normalized = str(value).strip()
+            return normalized or None
+
         filters = DataSourceFilters(
-            uid=payload.get("uid"),
-            name_en=payload.get("name_en"),
-            name_zh=payload.get("name_zh"),
-            database_type=payload.get("type"),
+            uid=optional_text("uid"),
+            name_en=optional_text("name_en"),
+            name_zh=optional_text("name_zh"),
+            database_type=optional_text("type"),
             status=payload.get("status"),
         )
         return self.definitions.list(filters)

deployment/app/models/data_research.py

@@ -115,6 +115,10 @@ class IngestionJob(db.Model):
             "idempotency_key",
             name="uq_ingestion_job_idempotency_key",
         ),
+        db.CheckConstraint(
+            "attempt_count >= 0",
+            name="ck_ingestion_job_attempt_count",
+        ),
         {"schema": "public"},
     )
 
@@ -135,6 +139,8 @@ class IngestionJob(db.Model):
     parameters = db.Column(JSONB, nullable=False, default=dict)
     statistics = db.Column(JSONB, nullable=False, default=dict)
     last_error = db.Column(db.String(1000))
+    attempt_count = db.Column(db.Integer, nullable=False, default=0)
+    failure_stage = db.Column(db.String(30))
     actor_uid = db.Column(db.String(100))
     force_rerun = db.Column(db.Boolean, nullable=False, default=False)
     created_at = db.Column(db.DateTime, nullable=False, default=now_china_naive)
@@ -153,6 +159,8 @@ class IngestionJob(db.Model):
             "parameters": dict(self.parameters or {}),
             "statistics": dict(self.statistics or {}),
             "last_error": self.last_error,
+            "attempt_count": int(self.attempt_count or 0),
+            "failure_stage": self.failure_stage,
             "actor_uid": self.actor_uid,
             "created_at": _iso(self.created_at),
             "updated_at": _iso(self.updated_at),
@@ -161,6 +169,59 @@ class IngestionJob(db.Model):
         }
 
 
+class CatalogSnapshot(db.Model):
+    __tablename__ = "catalog_snapshots"
+    __table_args__ = (
+        db.CheckConstraint(
+            "attempt > 0",
+            name="ck_catalog_snapshot_attempt",
+        ),
+        db.UniqueConstraint(
+            "job_uid",
+            "attempt",
+            name="uq_catalog_snapshot_job_attempt",
+        ),
+        {"schema": "public"},
+    )
+
+    uid = db.Column(
+        UUID(as_uuid=False),
+        primary_key=True,
+        default=new_governance_uid,
+    )
+    job_uid = db.Column(
+        UUID(as_uuid=False),
+        db.ForeignKey("public.ingestion_jobs.uid", ondelete="CASCADE"),
+        nullable=False,
+    )
+    source_uid = db.Column(
+        UUID(as_uuid=False),
+        db.ForeignKey("public.ingestion_sources.uid"),
+        nullable=False,
+    )
+    attempt = db.Column(db.Integer, nullable=False)
+    database_type = db.Column(db.String(20), nullable=False)
+    content_hash = db.Column(db.String(64), nullable=False)
+    snapshot = db.Column(JSONB, nullable=False)
+    created_at = db.Column(
+        db.DateTime,
+        nullable=False,
+        default=now_china_naive,
+    )
+
+    def to_dict(self) -> dict[str, Any]:
+        return {
+            "uid": str(self.uid),
+            "job_uid": str(self.job_uid),
+            "source_uid": str(self.source_uid),
+            "attempt": int(self.attempt),
+            "database_type": self.database_type,
+            "content_hash": self.content_hash,
+            "snapshot": dict(self.snapshot or {}),
+            "created_at": _iso(self.created_at),
+        }
+
+
 class EvidenceFragment(db.Model):
     __tablename__ = "evidence_fragments"
     __table_args__ = (

docs/DATAOPS_PHASE1_3_MONTH_WORK_PLAN_20260729.md

@@ -155,7 +155,7 @@ P2 不阻塞第一阶段验收。没有完成的 P2 功能必须保留接口和
 | WP-00 | 进行中 | 基准提交 `0d6f746`；第一阶段分支 `codex/dataops-phase1-equipment-governance` | 企业负责人、真实数据范围和验收责任人仍需项目现场确认 |
 | WP-01 | 工程完成 | 合并提交 `cab9c1f`；后端 839 项回归、前端构建、OpenAPI 142 项和 Docker 本体链路通过 | `deployment/app` 历史发布副本仍有基线差异，正式交付前按 WP-13 统一收口 |
 | WP-02 | 工程完成，待企业配置 | 本地三级 RBAC；设备责任矩阵；唯一最终设备资产管理员；修订、并发冲突和审计快照；OpenAPI 144 项 | 需要企业提供实际设备资产管理员、治理人员和查看者名单后配置并完成 UAT |
-| WP-03 | 待启动 | 已具备 PostgreSQL/MySQL 连接池、多源采集和证据链底座 | 需要企业设备台账库、维修库只读账号、网络和数据字典 |
+| WP-03 | 工程完成，待企业接入 | 数据源自动登记；只读目录执行；幂等与主动重采；尝试次数、失败阶段和脱敏诊断；每次执行不可变目录快照；字段级来源证据；本地隔离 PostgreSQL/MySQL 双源实测；OpenAPI 147 项 | 需要企业设备台账库、维修库只读账号、网络、采集范围和数据字典；当前同步执行满足示范版，异步 Worker 与断点续跑保留为后续增强 |
 
 ## 7. 12 周执行计划
 

docs/FUNCTION_MODULE_CENSUS_20260726.md

@@ -500,9 +500,9 @@ DataOps Platform 当前已经具备较完整的“治理对象 → 知识服务
 | CON-11 | 数据连接 / 实时流 | Kafka、Pulsar 的 Topic、Schema 和血缘采集 | 规划中 |
 | CON-12 | 数据连接 / API 与 SaaS | REST、Web Service 和第三方 SaaS 连接器 | 规划中 |
 | CON-13 | 数据连接 / BI 系统 | BI 数据集、报表、仪表盘和指标依赖采集 | 规划中 |
-| CON-14 | 采集控制 / 采集来源 | 统一定义数据库、文件、应用和流式来源 | 隔离分支完成，待合入 |
-| CON-15 | 采集控制 / 采集任务 | 异步任务、幂等、重试、取消、断点续跑和统计 | 隔离分支完成，待合入 |
-| CON-16 | 采集控制 / 证据定位 | 保存表/字段、文件页码、单元格和来源证据 | 隔离分支完成，待合入 |
+| CON-14 | 采集控制 / 采集来源 | 统一定义数据库、文件、应用和流式来源 | 部分建设 |
+| CON-15 | 采集控制 / 采集任务 | 异步任务、幂等、重试、取消、断点续跑和统计 | 部分建设 |
+| CON-16 | 采集控制 / 证据定位 | 保存表/字段、文件页码、单元格和来源证据 | 已建设 |
 | CON-17 | 企业侧执行 / 边缘网关 | 在客户内网执行采集、画像、质量和受控查询 | 规划中 |
 | CON-18 | 企业侧执行 / 数据不出域 | 只向 SaaS 控制面传输脱敏元数据、统计、血缘和证据 | 规划中 |
 | CON-19 | 企业侧执行 / 近期明细 | 企业侧保存近期设备运行明细，默认保留一年 | 规划中 |

docs/architecture/DATA_MODEL.md

@@ -139,7 +139,8 @@ flowchart LR
 | `datasource_credential_audit_events` | `data_source_uid`, `credential_version`, `event_type`, `actor_uid`, `safe_detail` | 不含秘密的凭据及连接池审计 |
 | `ingestion_sources` | `uid`, `source_type`, `config`, `permission_scope` | 数据库、文件、DDL 采集来源 |
 | `source_artifacts` | `source_uid`, `content_hash`, `storage_ref`, `parser_version` | MinIO 原件/工件索引与哈希去重 |
-| `ingestion_jobs` | `idempotency_key`, `status`, `statistics`, `last_error` | 可重试采集状态机 |
+| `ingestion_jobs` | `idempotency_key`, `status`, `attempt_count`, `failure_stage`, `statistics`, `last_error` | 可重复执行、可诊断、可重试的采集状态机 |
+| `catalog_snapshots` | `job_uid`, `source_uid`, `attempt`, `content_hash`, `snapshot` | 每次数据库目录采集的不可变结构快照 |
 | `evidence_fragments` | `artifact_uid`, `locator`, `excerpt`, `confidence` | 页、表、行列、坐标级证据 |
 | `extraction_candidates` | `normalized_data`, `evidence_uids`, `confidence`, `status` | 解析候选项 |
 | `data_elements` | `code`, `current_version`, `status`, `business_domain_uids` | 稳定数据元素身份与生命周期 |

docs/architecture/OPENAPI.yaml

@@ -3,7 +3,7 @@ info:
   title: "DataOps Platform API（当前代码基线）"
   version: "2026-07-16"
   description: "由 scripts/generate_openapi.py 从 app/api/*/routes.py 生成。请求体与响应细节仍以现有专项 API 文档和代码为准。"
-x-route-count: 144
+x-route-count: 147
 servers:
   - url: "http://localhost:15500"
     description: "全本地隔离测试后端"
@@ -1870,6 +1870,88 @@ paths:
             application/json:
               schema:
                 $ref: '#/components/schemas/ApiEnvelope'
+  "/api/development/v1/ingestion-jobs/{job_uid}/catalog-snapshots":
+    get:
+      tags: [data_development]
+      operationId: data_development_list_catalog_snapshots_get
+      summary: "list catalog snapshots"
+      x-source: "app/api/data_development/routes.py"
+      parameters:
+        - name: job_uid
+          in: path
+          required: true
+          schema:
+            type: string
+      responses:
+        "200":
+          description: "请求已由当前实现处理"
+          content:
+            application/json:
+              schema:
+                $ref: '#/components/schemas/ApiEnvelope'
+        default:
+          description: "错误响应"
+          content:
+            application/json:
+              schema:
+                $ref: '#/components/schemas/ApiEnvelope'
+  "/api/development/v1/ingestion-jobs/{job_uid}/evidence":
+    get:
+      tags: [data_development]
+      operationId: data_development_list_ingestion_job_evidence_get
+      summary: "list ingestion job evidence"
+      x-source: "app/api/data_development/routes.py"
+      parameters:
+        - name: job_uid
+          in: path
+          required: true
+          schema:
+            type: string
+      responses:
+        "200":
+          description: "请求已由当前实现处理"
+          content:
+            application/json:
+              schema:
+                $ref: '#/components/schemas/ApiEnvelope'
+        default:
+          description: "错误响应"
+          content:
+            application/json:
+              schema:
+                $ref: '#/components/schemas/ApiEnvelope'
+  "/api/development/v1/ingestion-jobs/{job_uid}/execute":
+    post:
+      tags: [data_development]
+      operationId: data_development_execute_ingestion_job_post
+      summary: "execute ingestion job"
+      x-source: "app/api/data_development/routes.py"
+      parameters:
+        - name: job_uid
+          in: path
+          required: true
+          schema:
+            type: string
+      requestBody:
+        required: false
+        content:
+          application/json:
+            schema:
+              type: object
+              additionalProperties: true
+      responses:
+        "200":
+          description: "请求已由当前实现处理"
+          content:
+            application/json:
+              schema:
+                $ref: '#/components/schemas/ApiEnvelope'
+        default:
+          description: "错误响应"
+          content:
+            application/json:
+              schema:
+                $ref: '#/components/schemas/ApiEnvelope'
   "/api/development/v1/ingestion-jobs/{job_uid}/retry":
     post:
       tags: [data_development]

docs/superpowers/plans/2026-07-29-wp03-enterprise-datasource-ingestion.md

@@ -0,0 +1,177 @@
+# WP-03 Enterprise Data Source Ingestion Implementation Plan
+
+> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking.
+
+**Goal:** Connect the existing PostgreSQL/MySQL data-source runtime to the data-research ingestion control plane so a configured enterprise database can be registered, collected repeatedly, diagnosed, retried, and traced to an immutable catalog snapshot and field evidence.
+
+**Architecture:** Keep `app/core/data_source` as the credential-safe connection boundary and `app/core/data_research` as the ingestion control plane. Database collection runs synchronously behind an explicit job execution endpoint for the demonstration release, while persisted job state, attempt numbers, catalog snapshots, and evidence make later worker extraction possible without changing the API contract.
+
+**Tech Stack:** Flask, SQLAlchemy, PostgreSQL JSONB, Neo4j-backed data-source definitions, Vue 2, Vuetify, pytest.
+
+## Global Constraints
+
+- Only PostgreSQL and MySQL database catalog collection are in WP-03.
+- Production sources use the existing read-only `metadata_collection` connection purpose.
+- Credentials and connection strings must never enter ingestion source config, job parameters, snapshots, evidence, API responses, or logs.
+- A repeated canonical create reuses the existing job; a deliberate new collection uses `force_rerun`.
+- A retry preserves prior attempts and starts a new numbered attempt.
+- Real enterprise connectivity remains an external acceptance gate until the customer supplies network access, read-only accounts, scope, and data dictionaries.
+- Work continues on `codex/dataops-phase1-equipment-governance`; no push or deployment is authorized.
+
+---
+
+### Task 1: Database Source Registration
+
+**Files:**
+- Create: `app/core/data_research/sources.py`
+- Modify: `app/core/data_research/repository.py`
+- Modify: `app/api/data_development/routes.py`
+- Test: `tests/data_research/test_database_source_registration.py`
+- Test: `tests/data_research/test_development_api.py`
+
+**Interfaces:**
+- Consumes: `DataSourceConnectionManager.definitions.get(uid)` and `IngestionSource`.
+- Produces: `DatabaseSourceRegistrationService.ensure(data_source_uid, actor_uid)`.
+
+- [x] **Step 1: Write failing tests**
+
+Cover successful PostgreSQL/MySQL registration, idempotent refresh, disabled or unsupported definitions, secret-free stored config, and automatic registration before a `catalog_collect` job is created.
+
+- [x] **Step 2: Verify RED**
+
+Run:
+
+```bash
+PYTHONPATH=. .venv/bin/pytest -q tests/data_research/test_database_source_registration.py tests/data_research/test_development_api.py
+```
+
+Expected: failure because the registration service and route integration do not exist.
+
+- [x] **Step 3: Implement the minimal registration boundary**
+
+Store only database type, database name, and default schema. Reuse the external data-source UID as the ingestion source UID so jobs, connection pools, audit records, and evidence share one stable identifier.
+
+- [x] **Step 4: Verify GREEN**
+
+Run the Task 1 test command and expect all tests to pass.
+
+### Task 2: Attempt-Aware Catalog Execution and Evidence
+
+**Files:**
+- Create: `app/core/data_research/catalog/execution.py`
+- Modify: `app/core/data_research/catalog/models.py`
+- Modify: `app/core/data_research/ingestion.py`
+- Modify: `app/core/data_research/models.py`
+- Modify: `app/core/data_research/repository.py`
+- Modify: `app/core/data_research/artifacts.py`
+- Modify: `app/models/data_research.py`
+- Create: `migrations/versions/20260729_280_catalog_ingestion_execution.py`
+- Test: `tests/data_research/test_catalog_execution.py`
+- Test: `tests/data_research/test_ingestion_service.py`
+- Test: `tests/data_research/test_ingestion_models.py`
+- Test: `tests/test_database_migrations.py`
+
+**Interfaces:**
+- Consumes: `CatalogCollectionService.collect(data_source_uid, CatalogScope)` and `IngestionService.transition`.
+- Produces: `CatalogIngestionExecutor.execute(job_uid)`, `CatalogSnapshotRecord`, and `SqlAlchemyCatalogSnapshotRepository`.
+
+- [x] **Step 1: Write failing tests**
+
+Cover attempt incrementing, failure-stage capture, sanitized errors, valid scope parsing, successful state progression, immutable per-attempt snapshots, field evidence locators, failed collection, retry, and same-attempt resume.
+
+- [x] **Step 2: Verify RED**
+
+Run:
+
+```bash
+PYTHONPATH=. .venv/bin/pytest -q tests/data_research/test_catalog_execution.py tests/data_research/test_ingestion_service.py tests/data_research/test_ingestion_models.py tests/test_database_migrations.py
+```
+
+Expected: failure because execution, snapshot persistence, and attempt fields do not exist.
+
+- [x] **Step 3: Implement the minimal execution path**
+
+Move jobs through `queued → extracting → normalizing → matching → awaiting_review`, increment attempts at extraction start, persist one catalog snapshot per job attempt, and persist one evidence fragment per collected field. On failure, persist the sanitized diagnostic and the stage that failed before returning an error.
+
+- [x] **Step 4: Verify GREEN**
+
+Run the Task 2 test command and expect all tests to pass.
+
+### Task 3: Management API and Operator UI
+
+**Files:**
+- Modify: `app/api/data_development/routes.py`
+- Modify: `app/core/system/permissions.py`
+- Modify: `frontend/src/api/dataDevelopment.js`
+- Modify: `frontend/src/views/dataGovernance/development/ingestion.vue`
+- Modify: `frontend/src/views/dataGovernance/development/tasks.vue`
+- Test: `tests/data_research/test_development_api.py`
+- Test: `tests/data_research/test_development_frontend_contract.py`
+- Test: `tests/test_permission_matrix.py`
+
+**Interfaces:**
+- Produces: `POST /api/development/v1/ingestion-jobs/{job_uid}/execute`, `GET /api/development/v1/ingestion-jobs/{job_uid}/catalog-snapshots`, and `GET /api/development/v1/ingestion-jobs/{job_uid}/evidence`.
+
+- [x] **Step 1: Write failing tests**
+
+Cover permission classification, execution response, snapshot/evidence responses, secret-free serialization, selectable configured data sources, deliberate rerun, attempt/failure diagnostics, and retry controls.
+
+- [x] **Step 2: Verify RED**
+
+Run:
+
+```bash
+PYTHONPATH=. .venv/bin/pytest -q tests/data_research/test_development_api.py tests/data_research/test_development_frontend_contract.py tests/test_permission_matrix.py
+```
+
+Expected: failure because the endpoints and UI contract do not exist.
+
+- [x] **Step 3: Implement API and UI**
+
+Use the existing configured data-source list rather than requiring users to type a UID. Execute a newly created database catalog job explicitly, expose progress and diagnostic fields in the task list, and keep evidence/snapshot reads available to viewers.
+
+- [x] **Step 4: Verify GREEN**
+
+Run the Task 3 test command and expect all tests to pass.
+
+### Task 4: Contracts, Status Ledger, and Release Gate
+
+**Files:**
+- Modify: `docs/architecture/OPENAPI.yaml`
+- Modify: `docs/architecture/DATA_MODEL.md`
+- Modify: `docs/FUNCTION_MODULE_CENSUS_20260726.md`
+- Modify: `docs/DATAOPS_PHASE1_3_MONTH_WORK_PLAN_20260729.md`
+
+**Interfaces:**
+- Produces: generated API inventory, current data model, and an evidence-backed WP-03 status that separates engineering completion from enterprise connectivity acceptance.
+
+- [x] **Step 1: Regenerate and verify contracts**
+
+Run:
+
+```bash
+.venv/bin/python scripts/generate_openapi.py
+PYTHONPATH=. .venv/bin/pytest -q tests/test_architecture_artifacts.py
+```
+
+- [x] **Step 2: Update status documents**
+
+Record engineering evidence and leave the two real enterprise source connections as externally blocked until customer inputs are supplied.
+
+- [x] **Step 3: Run complete verification**
+
+Run the focused integration tests, full backend suite, frontend production build, migration upgrade, browser flow, and `git diff --check`.
+
+Verification evidence:
+
+- Backend: 860 passed, 41 skipped, and 59 subtests passed.
+- Real local database integration: PostgreSQL and MySQL both passed.
+- Frontend: production build completed with zero errors; existing dependency, bundle-size, CSS ordering, and console-statement warnings remain.
+- Migration: local Docker database is at `20260729_280 (head)`.
+- Browser: configured PostgreSQL source produced one table, two field evidence rows, and an `awaiting_review` task with attempt count 1; browser console reported zero errors and zero warnings.
+- Release subset: every WP-03 backend file copied into `deployment/app` matches the canonical file under `app`; the historical full release-copy reconciliation remains WP-13 scope.
+- Hygiene: `git diff --check` completed without findings.
+
+- [x] **Step 4: Commit**
+
+Create one independently reversible WP-03 engineering commit. Do not push.

frontend/src/api/dataDevelopment.js

@@ -6,8 +6,11 @@ const ONTOLOGY_BASE = '/development/v1/ontologies'
 export const createIngestionJob = params => http.post(BASE, params)
 export const getIngestionJobs = params => http.get(BASE, params)
 export const getIngestionJob = uid => http.get(`${BASE}/${uid}`)
+export const executeIngestionJob = uid => http.post(`${BASE}/${uid}/execute`)
 export const retryIngestionJob = uid => http.post(`${BASE}/${uid}/retry`)
 export const cancelIngestionJob = uid => http.post(`${BASE}/${uid}/cancel`)
+export const getCatalogSnapshots = uid => http.get(`${BASE}/${uid}/catalog-snapshots`)
+export const getJobEvidence = uid => http.get(`${BASE}/${uid}/evidence`)
 export const getOntologies = () => http.get(ONTOLOGY_BASE)
 export const getOntology = uid => http.get(`${ONTOLOGY_BASE}/${uid}`)
 export const getOntologyGraph = uid => http.get(`${ONTOLOGY_BASE}/${uid}/graph`)
@@ -67,8 +70,11 @@ export default {
   createIngestionJob,
   getIngestionJobs,
   getIngestionJob,
+  executeIngestionJob,
   retryIngestionJob,
   cancelIngestionJob,
+  getCatalogSnapshots,
+  getJobEvidence,
   getOntologies,
   getOntology,
   getOntologyGraph,

frontend/src/views/dataGovernance/development/ingestion.vue

@@ -1,44 +1,205 @@
 <template>
   <div class="pa-6">
-    <h1 class="text-h4 mb-2">新建数据采集</h1>
-    <p class="text--secondary">支持数据库直连解析 DDL，或上传 SQL、XLS、XLSX、CSV、DOCX、PDF、PNG、JPG/JPEG 文件。</p>
+    <div class="d-flex align-start mb-6">
+      <div>
+        <h1 class="text-h4 mb-2">新建数据采集</h1>
+        <p class="text--secondary mb-0">
+          数据库采集只读取目录元数据，不读取业务表数据，也不会在任务中保存连接凭据。
+        </p>
+      </div>
+      <v-spacer />
+      <v-chip color="success" outlined>只读采集</v-chip>
+    </div>
+
+    <v-alert type="info" text class="mb-6">
+      首期支持 PostgreSQL 和 MySQL。请先在“数据源管理”中配置并验证连接。
+    </v-alert>
+
     <v-stepper v-model="step" vertical>
-      <v-stepper-step :complete="step > 1" step="1">选择信息来源</v-stepper-step>
+      <v-stepper-step :complete="step > 1" step="1">
+        选择信息来源
+      </v-stepper-step>
       <v-stepper-content step="1">
-        <v-radio-group v-model="sourceType" row><v-radio label="数据库目录" value="database" /><v-radio label="DDL 文本" value="ddl" /><v-radio label="文件上传" value="file" /></v-radio-group>
+        <v-radio-group v-model="sourceType" row>
+          <v-radio label="数据库目录" value="database" />
+          <v-radio label="DDL 文本" value="ddl" />
+          <v-radio label="文件上传" value="file" />
+        </v-radio-group>
         <v-btn color="primary" @click="step = 2">下一步</v-btn>
       </v-stepper-content>
-      <v-stepper-step :complete="step > 2" step="2">配置范围与文件</v-stepper-step>
+
+      <v-stepper-step :complete="step > 2" step="2">
+        配置采集范围
+      </v-stepper-step>
       <v-stepper-content step="2">
-        <v-text-field v-model="sourceUid" label="数据源 UID" />
-        <v-file-input v-if="sourceType === 'file'" v-model="file" label="选择治理文件" accept=".sql,.xls,.xlsx,.csv,.docx,.pdf,.png,.jpg,.jpeg" />
-        <v-textarea v-if="sourceType === 'ddl'" v-model="ddl" label="DDL" outlined />
-        <v-chip-group><v-chip v-for="format in formats" :key="format" small>{{ format }}</v-chip></v-chip-group>
-        <v-btn text @click="step = 1">上一步</v-btn><v-btn color="primary" @click="submit">创建任务</v-btn>
+        <v-select
+          v-if="sourceType === 'database'"
+          v-model="sourceUid"
+          :items="dataSources"
+          :loading="loadingSources"
+          item-value="uid"
+          item-text="display_name"
+          label="选择已配置数据源"
+          outlined
+          :rules="[value => !!value || '请选择数据源']"
+        />
+        <v-text-field
+          v-else
+          v-model="sourceUid"
+          label="采集来源 UID"
+          outlined
+        />
+
+        <template v-if="sourceType === 'database'">
+          <v-row>
+            <v-col cols="12" md="6">
+              <v-text-field
+                v-model="includeSchemas"
+                label="包含的 Schema"
+                hint="多个名称使用英文逗号分隔；留空表示按数据源默认范围"
+                persistent-hint
+                outlined
+              />
+            </v-col>
+            <v-col cols="12" md="6">
+              <v-text-field
+                v-model="includeTables"
+                label="包含的数据表"
+                hint="多个名称使用英文逗号分隔；留空表示采集范围内全部表"
+                persistent-hint
+                outlined
+              />
+            </v-col>
+          </v-row>
+          <v-checkbox
+            v-model="forceRerun"
+            label="重新采集并保留一次新的执行记录"
+          />
+        </template>
+
+        <v-file-input
+          v-if="sourceType === 'file'"
+          v-model="file"
+          label="选择治理文件"
+          accept=".sql,.xls,.xlsx,.csv,.docx,.pdf,.png,.jpg,.jpeg"
+          outlined
+        />
+        <v-textarea
+          v-if="sourceType === 'ddl'"
+          v-model="ddl"
+          label="DDL"
+          outlined
+        />
+        <v-chip-group>
+          <v-chip v-for="format in formats" :key="format" small>
+            {{ format }}
+          </v-chip>
+        </v-chip-group>
+
+        <v-btn text @click="step = 1">上一步</v-btn>
+        <v-btn
+          color="primary"
+          :loading="submitting"
+          :disabled="!sourceUid"
+          @click="submit"
+        >
+          创建并开始采集
+        </v-btn>
       </v-stepper-content>
     </v-stepper>
   </div>
 </template>
 
 <script>
-import { createIngestionJob, uploadSourceFile } from '@/api/dataDevelopment'
+import {
+  createIngestionJob,
+  executeIngestionJob,
+  uploadSourceFile
+} from '@/api/dataDevelopment'
+import { getDatasourceList } from '@/api/dataOrigin'
 
 export default {
   name: 'DataResearchIngestion',
-  data: () => ({ step: 1, sourceType: 'file', sourceUid: '', file: null, ddl: '', readOnly: false, formats: ['SQL', 'XLS', 'XLSX', 'CSV', 'DOCX', 'PDF', 'PNG', 'JPG/JPEG'] }),
+  data: () => ({
+    step: 1,
+    sourceType: 'database',
+    sourceUid: '',
+    dataSources: [],
+    loadingSources: false,
+    submitting: false,
+    includeSchemas: '',
+    includeTables: '',
+    forceRerun: false,
+    file: null,
+    ddl: '',
+    formats: ['PostgreSQL', 'MySQL', 'SQL', 'XLS', 'XLSX', 'CSV', 'DOCX', 'PDF', 'PNG', 'JPG/JPEG']
+  }),
+  created () {
+    this.loadDataSources()
+  },
   methods: {
+    csvValues (value) {
+      return String(value || '')
+        .split(',')
+        .map(item => item.trim())
+        .filter(Boolean)
+    },
+    async loadDataSources () {
+      this.loadingSources = true
+      try {
+        const response = await getDatasourceList({ status: true })
+        const records = (response.data && response.data.data_source) || []
+        this.dataSources = records
+          .filter(item => item.status && ['postgresql', 'mysql'].includes(item.type))
+          .map(item => ({
+            ...item,
+            display_name: `${item.name_zh || item.name_en} · ${String(item.type).toUpperCase()} · ${item.database}`
+          }))
+      } catch (error) {
+        this.$snackbar.error(error)
+      } finally {
+        this.loadingSources = false
+      }
+    },
     async submit () {
-      let artifactUid = null
-      if (this.file) {
-        const data = new FormData()
-        data.append('source_uid', this.sourceUid)
-        data.append('parser_version', 'auto-v1')
-        data.append('file', this.file)
-        const artifact = await uploadSourceFile(data)
-        artifactUid = artifact.data.uid
+      this.submitting = true
+      try {
+        let artifactUid = null
+        if (this.file) {
+          const data = new FormData()
+          data.append('source_uid', this.sourceUid)
+          data.append('parser_version', 'auto-v1')
+          data.append('file', this.file)
+          const artifact = await uploadSourceFile(data)
+          artifactUid = artifact.data.uid
+        }
+        const databaseJob = this.sourceType === 'database'
+        const response = await createIngestionJob({
+          source_uid: this.sourceUid,
+          artifact_uid: artifactUid,
+          job_type: databaseJob ? 'catalog_collect' : 'file_extract',
+          parser_version: databaseJob ? 'catalog-v1' : 'auto-v1',
+          force_rerun: databaseJob && this.forceRerun,
+          parameters: databaseJob
+            ? {
+                include_schemas: this.csvValues(this.includeSchemas),
+                exclude_schemas: [],
+                include_tables: this.csvValues(this.includeTables),
+                exclude_tables: []
+              }
+            : (this.ddl ? { ddl: this.ddl } : {})
+        })
+        const job = response.data
+        if (databaseJob && ['created', 'queued'].includes(job.status)) {
+          await executeIngestionJob(job.uid)
+        }
+        this.$snackbar.success('采集任务已创建')
+        this.$router.push('/data-governance/development/tasks')
+      } catch (error) {
+        this.$snackbar.error(error)
+      } finally {
+        this.submitting = false
       }
-      await createIngestionJob({ source_uid: this.sourceUid, artifact_uid: artifactUid, job_type: this.sourceType === 'database' ? 'catalog_collect' : 'file_extract', parser_version: 'auto-v1', parameters: this.ddl ? { ddl: this.ddl } : {} })
-      this.$router.push('/data-governance/development/tasks')
     }
   }
 }

frontend/src/views/dataGovernance/development/tasks.vue

@@ -1,21 +1,222 @@
 <template>
-  <div class="pa-6"><div class="d-flex align-center mb-4"><h1 class="text-h4">研发任务</h1><v-spacer /><v-btn color="primary" @click="$router.push('/data-governance/development/ingestion')">新建采集</v-btn></div>
+  <div class="pa-6">
+    <div class="d-flex align-center mb-4">
+      <div>
+        <h1 class="text-h4 mb-1">采集任务</h1>
+        <div class="text--secondary">查看采集进度、失败诊断、目录快照和来源证据。</div>
+      </div>
+      <v-spacer />
+      <v-btn
+        color="primary"
+        @click="$router.push('/data-governance/development/ingestion')"
+      >
+        新建采集
+      </v-btn>
+    </div>
+
     <v-data-table :headers="headers" :items="items" :loading="loading">
-      <template v-slot:[`item.status`]="{ item }"><v-chip small :color="colors[item.status]">{{ labels[item.status] || item.status }}</v-chip></template>
-      <template v-slot:[`item.actions`]="{ item }"><v-btn text color="primary" @click="$router.push(`/data-governance/development/review?job_uid=${item.uid}`)">查看</v-btn></template>
+      <template v-slot:[`item.status`]="{ item }">
+        <v-chip small :color="colors[item.status]">
+          {{ labels[item.status] || item.status }}
+        </v-chip>
+      </template>
+      <template v-slot:[`item.attempt_count`]="{ item }">
+        {{ item.attempt_count || 0 }}
+      </template>
+      <template v-slot:[`item.failure_stage`]="{ item }">
+        <div v-if="item.failure_stage">
+          <div>{{ stageLabels[item.failure_stage] || item.failure_stage }}</div>
+          <div class="text-caption error--text diagnostic">
+            {{ item.last_error }}
+          </div>
+        </div>
+        <span v-else>—</span>
+      </template>
+      <template v-slot:[`item.report`]="{ item }">
+        <template v-if="item.statistics && item.statistics.snapshot_uid">
+          {{ item.statistics.asset_count || 0 }} 张表 /
+          {{ item.statistics.field_count || 0 }} 个字段
+        </template>
+        <span v-else>—</span>
+      </template>
+      <template v-slot:[`item.actions`]="{ item }">
+        <v-btn
+          v-if="item.status === 'failed'"
+          text
+          color="warning"
+          :loading="runningUid === item.uid"
+          @click="retryAndRun(item)"
+        >
+          重试
+        </v-btn>
+        <v-btn
+          v-if="item.statistics && item.statistics.snapshot_uid"
+          text
+          color="primary"
+          @click="showEvidence(item)"
+        >
+          来源证据
+        </v-btn>
+        <v-btn
+          text
+          color="primary"
+          @click="$router.push(`/data-governance/development/review?job_uid=${item.uid}`)"
+        >
+          查看
+        </v-btn>
+      </template>
     </v-data-table>
-    <div class="d-none">排队中 解析中 待评审 已发布 失败 已取消</div>
+
+    <v-dialog v-model="evidenceDialog" max-width="860">
+      <v-card>
+        <v-card-title>目录快照与来源证据</v-card-title>
+        <v-card-text>
+          <v-alert v-if="selectedSnapshot" type="success" text>
+            第 {{ selectedSnapshot.attempt }} 次采集，
+            已保存 {{ selectedSnapshot.evidence_count }} 条字段证据。
+            内容摘要：{{ selectedSnapshot.content_hash }}
+          </v-alert>
+          <v-simple-table>
+            <thead>
+              <tr>
+                <th>Schema</th>
+                <th>数据表</th>
+                <th>字段</th>
+                <th>位置</th>
+              </tr>
+            </thead>
+            <tbody>
+              <tr v-for="record in evidenceRecords" :key="record.uid">
+                <td>{{ record.locator.schema }}</td>
+                <td>{{ record.locator.table }}</td>
+                <td>{{ record.locator.column }}</td>
+                <td>第 {{ record.locator.ordinal_position }} 列</td>
+              </tr>
+            </tbody>
+          </v-simple-table>
+        </v-card-text>
+        <v-card-actions>
+          <v-spacer />
+          <v-btn text @click="evidenceDialog = false">关闭</v-btn>
+        </v-card-actions>
+      </v-card>
+    </v-dialog>
+
+    <div class="d-none">排队中 解析中 待评审 已发布 失败 已取消 尝试次数 失败阶段 来源证据</div>
   </div>
 </template>
 
 <script>
-import { getIngestionJobs } from '@/api/dataDevelopment'
+import {
+  executeIngestionJob,
+  getCatalogSnapshots,
+  getIngestionJobs,
+  getJobEvidence,
+  retryIngestionJob
+} from '@/api/dataDevelopment'
 
 export default {
   name: 'DataResearchTasks',
-  data: () => ({ loading: false, readOnly: false, items: [], timer: null, labels: { queued: '排队中', extracting: '解析中', normalizing: '解析中', matching: '解析中', awaiting_review: '待评审', published: '已发布', failed: '失败', cancelled: '已取消' }, colors: { published: 'success', failed: 'error', awaiting_review: 'warning' }, headers: [{ text: '任务 UID', value: 'uid' }, { text: '类型', value: 'job_type' }, { text: '状态', value: 'status' }, { text: '操作', value: 'actions', sortable: false }] }),
-  created () { this.load(); this.timer = setInterval(this.load, 5000) },
-  beforeDestroy () { clearInterval(this.timer) },
-  methods: { async load () { this.loading = true; try { const response = await getIngestionJobs(); this.items = (response.data && response.data.records) || [] } finally { this.loading = false } } }
+  data: () => ({
+    loading: false,
+    items: [],
+    timer: null,
+    runningUid: null,
+    evidenceDialog: false,
+    selectedSnapshot: null,
+    evidenceRecords: [],
+    labels: {
+      created: '已创建',
+      queued: '排队中',
+      extracting: '采集中',
+      normalizing: '整理中',
+      matching: '匹配中',
+      awaiting_review: '待评审',
+      partial: '部分完成',
+      published: '已发布',
+      failed: '失败',
+      cancelled: '已取消'
+    },
+    stageLabels: {
+      queued: '排队',
+      extracting: '连接与采集',
+      normalizing: '目录整理',
+      matching: '字段匹配'
+    },
+    colors: {
+      published: 'success',
+      awaiting_review: 'warning',
+      failed: 'error',
+      cancelled: 'grey'
+    },
+    headers: [
+      { text: '任务 UID', value: 'uid' },
+      { text: '类型', value: 'job_type' },
+      { text: '状态', value: 'status' },
+      { text: '尝试次数', value: 'attempt_count' },
+      { text: '失败阶段', value: 'failure_stage' },
+      { text: '采集报告', value: 'report', sortable: false },
+      { text: '操作', value: 'actions', sortable: false }
+    ]
+  }),
+  created () {
+    this.load()
+    this.timer = setInterval(this.load, 5000)
+  },
+  beforeDestroy () {
+    clearInterval(this.timer)
+  },
+  methods: {
+    async load () {
+      this.loading = true
+      try {
+        const response = await getIngestionJobs()
+        this.items = (response.data && response.data.records) || []
+      } finally {
+        this.loading = false
+      }
+    },
+    async retryAndRun (item) {
+      this.runningUid = item.uid
+      try {
+        await retryIngestionJob(item.uid)
+        await executeIngestionJob(item.uid)
+        this.$snackbar.success('采集重试已完成')
+        await this.load()
+      } catch (error) {
+        this.$snackbar.error(error)
+        await this.load()
+      } finally {
+        this.runningUid = null
+      }
+    },
+    async showEvidence (item) {
+      try {
+        const [snapshots, evidence] = await Promise.all([
+          getCatalogSnapshots(item.uid),
+          getJobEvidence(item.uid)
+        ])
+        this.selectedSnapshot = (
+          snapshots.data &&
+          snapshots.data.records &&
+          snapshots.data.records[0]
+        ) || null
+        this.evidenceRecords = (
+          evidence.data &&
+          evidence.data.records
+        ) || []
+        this.evidenceDialog = true
+      } catch (error) {
+        this.$snackbar.error(error)
+      }
+    }
+  }
 }
 </script>
+
+<style lang="scss" scoped>
+.diagnostic {
+  max-width: 280px;
+  white-space: normal;
+}
+</style>

migrations/versions/20260729_280_catalog_ingestion_execution.py

@@ -0,0 +1,45 @@
+"""Add attempt-aware database catalog ingestion evidence."""
+
+from alembic import op
+
+
+revision = "20260729_280"
+down_revision = "20260729_270"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    op.execute(
+        """
+        ALTER TABLE public.ingestion_jobs
+            ADD COLUMN attempt_count INTEGER NOT NULL DEFAULT 0,
+            ADD COLUMN failure_stage VARCHAR(30);
+
+        ALTER TABLE public.ingestion_jobs
+            ADD CONSTRAINT ck_ingestion_job_attempt_count
+            CHECK (attempt_count >= 0);
+
+        CREATE TABLE public.catalog_snapshots (
+            uid UUID PRIMARY KEY,
+            job_uid UUID NOT NULL
+                REFERENCES public.ingestion_jobs(uid) ON DELETE CASCADE,
+            source_uid UUID NOT NULL
+                REFERENCES public.ingestion_sources(uid),
+            attempt INTEGER NOT NULL CHECK (attempt > 0),
+            database_type VARCHAR(20) NOT NULL,
+            content_hash CHAR(64) NOT NULL,
+            snapshot JSONB NOT NULL,
+            created_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP,
+            UNIQUE (job_uid, attempt)
+        );
+
+        CREATE INDEX idx_catalog_snapshots_source_created
+            ON public.catalog_snapshots(source_uid, created_at DESC);
+        """
+    )
+
+
+def downgrade() -> None:
+    # Catalog execution evidence is retained across application rollback.
+    pass

tests/data_research/test_catalog_collectors.py

@@ -93,6 +93,36 @@ def test_mysql_collector_uses_information_schema_and_bound_database():
     assert snapshot.database_type == "mysql"
 
 
+def test_collectors_bind_text_sentinel_for_empty_expanding_filters():
+    from app.core.data_research.catalog.base import EMPTY_SCOPE_SENTINEL
+    from app.core.data_research.catalog.models import CatalogScope
+    from app.core.data_research.catalog.mysql import MySqlCatalogCollector
+    from app.core.data_research.catalog.postgresql import PostgreSqlCatalogCollector
+
+    postgres = FakeConnection(sample_rows())
+    mysql = FakeConnection(sample_rows())
+
+    PostgreSqlCatalogCollector().collect(
+        postgres,
+        data_source_uid="source-1",
+        scope=CatalogScope(include_schemas=("public",)),
+    )
+    MySqlCatalogCollector().collect(
+        mysql,
+        data_source_uid="source-2",
+        database_name="analytics",
+        scope=CatalogScope(),
+    )
+
+    postgres_parameters = postgres.calls[0][1]
+    mysql_parameters = mysql.calls[0][1]
+    assert postgres_parameters["exclude_schemas"] == [EMPTY_SCOPE_SENTINEL]
+    assert postgres_parameters["include_tables"] == [EMPTY_SCOPE_SENTINEL]
+    assert postgres_parameters["exclude_tables"] == [EMPTY_SCOPE_SENTINEL]
+    assert mysql_parameters["include_tables"] == [EMPTY_SCOPE_SENTINEL]
+    assert mysql_parameters["exclude_tables"] == [EMPTY_SCOPE_SENTINEL]
+
+
 def test_catalog_service_uses_read_only_metadata_collection_purpose():
     from app.core.data_research.catalog.models import CatalogScope
     from app.core.data_research.catalog.postgresql import PostgreSqlCatalogCollector

tests/data_research/test_catalog_execution.py

@@ -0,0 +1,187 @@
+from __future__ import annotations
+
+from dataclasses import asdict
+
+import pytest
+
+from tests.data_research.test_ingestion_service import (
+    MemoryJobRepository,
+    payload,
+)
+
+
+class MemorySnapshotRepository:
+    def __init__(self):
+        self.records = {}
+
+    def find(self, job_uid, attempt):
+        return self.records.get((str(job_uid), int(attempt)))
+
+    def persist(self, job_uid, attempt, snapshot):
+        from app.core.data_research.catalog.models import CatalogSnapshotRecord
+
+        key = (str(job_uid), int(attempt))
+        existing = self.records.get(key)
+        if existing is not None:
+            return existing
+        field_count = sum(len(asset.fields) for asset in snapshot.assets)
+        record = CatalogSnapshotRecord(
+            uid=f"snapshot-{attempt}",
+            job_uid=str(job_uid),
+            source_uid=snapshot.data_source_uid,
+            attempt=int(attempt),
+            database_type=snapshot.database_type,
+            content_hash=snapshot.content_hash,
+            snapshot=asdict(snapshot),
+            evidence_count=field_count,
+        )
+        self.records[key] = record
+        return record
+
+    def list(self, job_uid):
+        return [
+            record
+            for (candidate_uid, _attempt), record in sorted(self.records.items())
+            if candidate_uid == str(job_uid)
+        ]
+
+
+def catalog_snapshot():
+    from app.core.data_research.catalog.models import (
+        CatalogAsset,
+        CatalogField,
+        CatalogSnapshot,
+    )
+
+    return CatalogSnapshot(
+        data_source_uid="00000000-0000-0000-0000-000000000001",
+        database_type="postgresql",
+        assets=(
+            CatalogAsset(
+                key="source:asset.equipment",
+                schema="asset",
+                name="equipment",
+                asset_type="table",
+                fields=(
+                    CatalogField(
+                        key="source:asset.equipment.equipment_code",
+                        schema="asset",
+                        asset="equipment",
+                        name="equipment_code",
+                        ordinal_position=1,
+                        data_type="varchar",
+                        nullable=False,
+                    ),
+                    CatalogField(
+                        key="source:asset.equipment.location_code",
+                        schema="asset",
+                        asset="equipment",
+                        name="location_code",
+                        ordinal_position=2,
+                        data_type="varchar",
+                        nullable=True,
+                    ),
+                ),
+            ),
+        ),
+    )
+
+
+class CatalogCollector:
+    def __init__(self, result=None, error=None):
+        self.result = result or catalog_snapshot()
+        self.error = error
+        self.calls = []
+
+    def collect(self, source_uid, scope):
+        self.calls.append((source_uid, scope))
+        if self.error is not None:
+            raise self.error
+        return self.result
+
+
+def catalog_payload(**overrides):
+    return payload(
+        artifact_uid=None,
+        job_type="catalog_collect",
+        parser_version="catalog-v1",
+        parameters={
+            "include_schemas": ["asset"],
+            "exclude_schemas": ["pg_catalog"],
+            "include_tables": ["equipment"],
+            "exclude_tables": [],
+        },
+        **overrides,
+    )
+
+
+def test_catalog_execution_persists_snapshot_evidence_and_job_report():
+    from app.core.data_research.catalog.execution import CatalogIngestionExecutor
+    from app.core.data_research.ingestion import IngestionService
+
+    jobs = MemoryJobRepository()
+    ingestion = IngestionService(jobs)
+    job, _ = ingestion.create_job(catalog_payload(), actor_uid="editor-1")
+    collector = CatalogCollector()
+    snapshots = MemorySnapshotRepository()
+    executor = CatalogIngestionExecutor(ingestion, collector, snapshots)
+
+    completed = executor.execute(job.uid)
+
+    assert completed.status == "awaiting_review"
+    assert completed.attempt_count == 1
+    assert completed.statistics == {
+        "snapshot_uid": "snapshot-1",
+        "content_hash": catalog_snapshot().content_hash,
+        "asset_count": 1,
+        "field_count": 2,
+        "evidence_count": 2,
+    }
+    assert snapshots.find(job.uid, 1).attempt == 1
+    assert collector.calls[0][1].include_schemas == ("asset",)
+
+    replayed = executor.execute(job.uid)
+    assert replayed == completed
+    assert len(collector.calls) == 1
+
+
+def test_failed_catalog_collection_records_stage_and_can_retry():
+    from app.core.data_research.catalog.execution import CatalogIngestionExecutor
+    from app.core.data_research.ingestion import IngestionService
+
+    jobs = MemoryJobRepository()
+    ingestion = IngestionService(jobs)
+    job, _ = ingestion.create_job(catalog_payload(), actor_uid="editor-1")
+    collector = CatalogCollector(
+        error=RuntimeError("password=clear-secret connection refused")
+    )
+    snapshots = MemorySnapshotRepository()
+    executor = CatalogIngestionExecutor(ingestion, collector, snapshots)
+
+    with pytest.raises(RuntimeError, match="connection refused"):
+        executor.execute(job.uid)
+
+    failed = ingestion.get_job(job.uid)
+    assert failed.status == "failed"
+    assert failed.attempt_count == 1
+    assert failed.failure_stage == "extracting"
+    assert "clear-secret" not in failed.last_error
+    assert snapshots.list(job.uid) == []
+
+    ingestion.retry(job.uid)
+    collector.error = None
+    completed = executor.execute(job.uid)
+
+    assert completed.status == "awaiting_review"
+    assert completed.attempt_count == 2
+    assert snapshots.find(job.uid, 2).attempt == 2
+
+
+def test_catalog_scope_rejects_non_list_or_non_text_values_before_collection():
+    from app.core.data_research.catalog.execution import catalog_scope_from_parameters
+    from app.core.data_research.errors import IngestionPayloadInvalid
+
+    with pytest.raises(IngestionPayloadInvalid, match="include_schemas"):
+        catalog_scope_from_parameters({"include_schemas": "asset"})
+    with pytest.raises(IngestionPayloadInvalid, match="include_tables"):
+        catalog_scope_from_parameters({"include_tables": ["equipment", 42]})

tests/data_research/test_database_source_registration.py

@@ -0,0 +1,92 @@
+from __future__ import annotations
+
+from types import SimpleNamespace
+
+import pytest
+
+
+class MemorySourceRepository:
+    def __init__(self):
+        self.records = {}
+
+    def get(self, uid):
+        return self.records.get(str(uid))
+
+    def save(self, record):
+        self.records[str(record.uid)] = record
+        return record
+
+
+def definition(**overrides):
+    values = {
+        "uid": "00000000-0000-0000-0000-000000000001",
+        "name_en": "equipment_registry",
+        "name_zh": "设备台账库",
+        "database_type": "postgresql",
+        "database": "equipment",
+        "schema": "asset",
+        "status": True,
+        "host": "db.internal.example",
+        "credential_ref": "vault://must-not-copy",
+    }
+    values.update(overrides)
+    return SimpleNamespace(**values)
+
+
+def test_database_source_registration_is_idempotent_and_secret_free():
+    from app.core.data_research.sources import DatabaseSourceRegistrationService
+
+    repository = MemorySourceRepository()
+    current = definition()
+    commits = []
+    service = DatabaseSourceRegistrationService(
+        repository,
+        definition_resolver=lambda _uid: current,
+        commit=lambda: commits.append("commit"),
+    )
+
+    first, first_created = service.ensure(current.uid, actor_uid="editor-1")
+    second, second_created = service.ensure(current.uid, actor_uid="editor-2")
+
+    assert first_created is True
+    assert second_created is False
+    assert second.uid == current.uid
+    assert second.name == "设备台账库"
+    assert second.config == {
+        "database_type": "postgresql",
+        "database": "equipment",
+        "schema": "asset",
+    }
+    assert "host" not in second.config
+    assert "credential" not in str(second.config).lower()
+    assert commits == ["commit", "commit"]
+
+
+@pytest.mark.parametrize(
+    ("candidate", "message"),
+    [
+        (None, "not found"),
+        (definition(status=False), "disabled"),
+        (definition(database_type="oracle"), "not supported"),
+    ],
+)
+def test_database_source_registration_rejects_unusable_definitions(
+    candidate,
+    message,
+):
+    from app.core.data_research.errors import IngestionSourceInvalid
+    from app.core.data_research.sources import DatabaseSourceRegistrationService
+
+    repository = MemorySourceRepository()
+    service = DatabaseSourceRegistrationService(
+        repository,
+        definition_resolver=lambda _uid: candidate,
+    )
+
+    with pytest.raises(IngestionSourceInvalid, match=message):
+        service.ensure(
+            "00000000-0000-0000-0000-000000000001",
+            actor_uid="editor-1",
+        )
+
+    assert repository.records == {}

tests/data_research/test_development_api.py

@@ -31,6 +31,75 @@ class FakeDevelopmentService:
         return replace(self.record, status="cancelled")
 
 
+class FakeSourceRegistrar:
+    def __init__(self):
+        self.actions = []
+
+    def ensure(self, source_uid, actor_uid):
+        self.actions.append((source_uid, actor_uid))
+        return object(), True
+
+
+class FakeCatalogExecutor:
+    def __init__(self, record):
+        self.record = record
+        self.actions = []
+
+    def execute(self, uid):
+        self.actions.append(uid)
+        return replace(
+            self.record,
+            status="awaiting_review",
+            attempt_count=1,
+            statistics={
+                "snapshot_uid": "snapshot-1",
+                "asset_count": 2,
+                "field_count": 8,
+                "evidence_count": 8,
+            },
+        )
+
+
+class FakeCatalogSnapshotRepository:
+    def list(self, job_uid):
+        from app.core.data_research.catalog.models import CatalogSnapshotRecord
+
+        return [
+            CatalogSnapshotRecord(
+                uid="snapshot-1",
+                job_uid=job_uid,
+                source_uid="00000000-0000-0000-0000-000000000001",
+                attempt=1,
+                database_type="postgresql",
+                content_hash="b" * 64,
+                snapshot={
+                    "data_source_uid": "00000000-0000-0000-0000-000000000001",
+                    "database_type": "postgresql",
+                    "assets": [],
+                },
+                evidence_count=8,
+            )
+        ]
+
+
+class FakeEvidenceService:
+    def list_for_job(self, job_uid):
+        return [
+            {
+                "uid": "evidence-1",
+                "job_uid": job_uid,
+                "locator": {
+                    "kind": "database.column",
+                    "schema": "asset",
+                    "table": "equipment",
+                    "column": "equipment_code",
+                },
+                "excerpt": "password=[redacted]",
+                "confidence": 1.0,
+            }
+        ]
+
+
 @pytest.fixture()
 def development_client(monkeypatch):
     from flask import request
@@ -51,6 +120,8 @@ def development_client(monkeypatch):
         parameters={"schema": "public"},
     )
     service = FakeDevelopmentService(record)
+    registrar = FakeSourceRegistrar()
+    executor = FakeCatalogExecutor(record)
 
     def identity():
         header = request.headers.get("Authorization", "")
@@ -66,8 +137,30 @@ def development_client(monkeypatch):
 
     monkeypatch.setattr(permissions, "authenticate_request", identity)
     monkeypatch.setattr(routes, "get_ingestion_service", lambda: service)
+    monkeypatch.setattr(
+        routes,
+        "get_database_source_registration_service",
+        lambda: registrar,
+    )
+    monkeypatch.setattr(
+        routes,
+        "get_catalog_ingestion_executor",
+        lambda: executor,
+    )
+    monkeypatch.setattr(
+        routes,
+        "get_catalog_snapshot_repository",
+        lambda: FakeCatalogSnapshotRepository(),
+    )
+    monkeypatch.setattr(
+        routes,
+        "get_evidence_service",
+        lambda: FakeEvidenceService(),
+    )
     app = create_app()
     app.config.update(TESTING=True)
+    service.registrar = registrar
+    service.executor = executor
     return app.test_client(), service
 
 
@@ -138,3 +231,60 @@ def test_retry_is_admin_only_and_cancel_is_owner_or_admin(development_client):
     assert admin_retry.status_code == 200
     assert other_cancel.status_code == 403
     assert owner_cancel.status_code == 200
+
+
+def test_catalog_job_registers_source_then_executes_with_report(
+    development_client,
+):
+    client, service = development_client
+    payload = {
+        **job_payload(),
+        "artifact_uid": None,
+        "job_type": "catalog_collect",
+        "parser_version": "catalog-v1",
+    }
+
+    created = client.post(
+        "/api/development/v1/ingestion-jobs",
+        json=payload,
+        headers={"Authorization": "Bearer editor"},
+    )
+    executed = client.post(
+        "/api/development/v1/ingestion-jobs/"
+        "00000000-0000-0000-0000-000000000010/execute",
+        headers={"Authorization": "Bearer editor"},
+    )
+
+    assert created.status_code == 201
+    assert service.registrar.actions == [
+        ("00000000-0000-0000-0000-000000000001", "editor-1")
+    ]
+    assert executed.status_code == 200
+    assert executed.get_json()["data"]["status"] == "awaiting_review"
+    assert executed.get_json()["data"]["attempt_count"] == 1
+    assert service.executor.actions == [
+        "00000000-0000-0000-0000-000000000010"
+    ]
+
+
+def test_catalog_snapshots_and_evidence_are_readable_without_secrets(
+    development_client,
+):
+    client, _service = development_client
+    uid = "00000000-0000-0000-0000-000000000010"
+
+    snapshots = client.get(
+        f"/api/development/v1/ingestion-jobs/{uid}/catalog-snapshots",
+        headers={"Authorization": "Bearer viewer"},
+    )
+    evidence = client.get(
+        f"/api/development/v1/ingestion-jobs/{uid}/evidence",
+        headers={"Authorization": "Bearer viewer"},
+    )
+
+    assert snapshots.status_code == 200
+    assert snapshots.get_json()["data"]["records"][0]["attempt"] == 1
+    assert evidence.status_code == 200
+    text = evidence.get_data(as_text=True)
+    assert "clear-secret" not in text
+    assert "equipment_code" in text

tests/data_research/test_development_frontend_contract.py

@@ -15,5 +15,27 @@ def test_data_development_frontend_api_covers_v60_job_contract():
     assert "getIngestionJob" in source
     assert "retryIngestionJob" in source
     assert "cancelIngestionJob" in source
+    assert "executeIngestionJob" in source
+    assert "getCatalogSnapshots" in source
+    assert "getJobEvidence" in source
     assert "password" not in source.lower()
     assert "credentials" not in source.lower()
+
+
+def test_database_ingestion_ui_uses_configured_sources_and_shows_diagnostics():
+    ingestion = (
+        ROOT / "frontend/src/views/dataGovernance/development/ingestion.vue"
+    ).read_text(encoding="utf-8")
+    tasks = (
+        ROOT / "frontend/src/views/dataGovernance/development/tasks.vue"
+    ).read_text(encoding="utf-8")
+
+    assert "getDatasourceList" in ingestion
+    assert "executeIngestionJob" in ingestion
+    assert "force_rerun" in ingestion
+    assert "选择已配置数据源" in ingestion
+    assert "尝试次数" in tasks
+    assert "失败阶段" in tasks
+    assert "来源证据" in tasks
+    assert "retryIngestionJob" in tasks
+    assert "executeIngestionJob" in tasks

tests/data_research/test_ingestion_models.py

@@ -40,6 +40,9 @@ def test_ingestion_job_has_idempotency_and_status_constraints():
     )
 
     assert ("idempotency_key",) in unique_columns
+    assert "attempt_count" in IngestionJob.__table__.c
+    assert "failure_stage" in IngestionJob.__table__.c
+    assert "attempt_count >= 0" in check_sql
     for status in (
         "created",
         "queued",
@@ -55,6 +58,22 @@ def test_ingestion_job_has_idempotency_and_status_constraints():
         assert status in check_sql
 
 
+def test_catalog_snapshot_has_one_immutable_record_per_job_attempt():
+    from app.models.data_research import CatalogSnapshot
+
+    constraints = list(CatalogSnapshot.__table__.constraints)
+    unique_columns = {
+        tuple(column.name for column in constraint.columns)
+        for constraint in constraints
+        if isinstance(constraint, UniqueConstraint)
+    }
+
+    assert CatalogSnapshot.__tablename__ == "catalog_snapshots"
+    assert CatalogSnapshot.__table__.c.job_uid.foreign_keys
+    assert ("job_uid", "attempt") in unique_columns
+    assert "content_hash" in CatalogSnapshot.__table__.c
+
+
 def test_model_serialization_redacts_control_plane_secrets():
     from app.models.data_research import IngestionSource, SourceArtifact
 

tests/data_research/test_ingestion_service.py

@@ -104,6 +104,8 @@ def test_failed_job_can_retry_and_running_job_can_cancel():
     service = IngestionService(repository)
     job, _ = service.create_job(payload(), actor_uid="editor-1")
     service.transition(job.uid, "queued")
+    extracting = service.transition(job.uid, "extracting")
+    assert extracting.attempt_count == 1
     failed = service.transition(
         job.uid,
         "failed",
@@ -113,10 +115,15 @@ def test_failed_job_can_retry_and_running_job_can_cancel():
     assert "clear-secret" not in failed.last_error
     assert "abc123" not in failed.last_error
     assert "[redacted]" in failed.last_error
+    assert failed.failure_stage == "extracting"
+    assert failed.attempt_count == 1
 
     retried = service.retry(job.uid)
     assert retried.status == "queued"
     assert retried.last_error is None
+    assert retried.failure_stage is None
+    extracting_again = service.transition(job.uid, "extracting")
+    assert extracting_again.attempt_count == 2
 
     cancelled = service.cancel(job.uid)
     assert cancelled.status == "cancelled"

tests/integration/test_catalog_ingestion_databases.py

@@ -0,0 +1,202 @@
+from __future__ import annotations
+
+import os
+import uuid
+
+import pytest
+from sqlalchemy.engine import make_url
+
+
+pytestmark = pytest.mark.integration
+
+
+class DefinitionRepository:
+    def __init__(self, definition):
+        self.definition = definition
+
+    def get(self, uid):
+        return self.definition if str(uid) == self.definition.uid else None
+
+
+class CredentialRepository:
+    def __init__(self, credential):
+        self.credential = credential
+
+    def get_active(self, _session, _uid, _version):
+        return self.credential
+
+
+@pytest.mark.parametrize(
+    ("database_type", "source_environment"),
+    [
+        ("postgresql", "TEST_SOURCE_POSTGRES_URL"),
+        ("mysql", "TEST_SOURCE_MYSQL_URL"),
+    ],
+)
+def test_real_database_catalog_collection_persists_attempt_and_evidence(
+    monkeypatch,
+    database_type,
+    source_environment,
+):
+    platform_url = os.environ.get("TEST_DATABASE_URL")
+    source_url = os.environ.get(source_environment)
+    if not platform_url or not source_url:
+        pytest.skip(
+            f"TEST_DATABASE_URL and {source_environment} are required"
+        )
+
+    monkeypatch.setenv("DATABASE_URL", platform_url)
+    from app import create_app, db
+    from app.config.config import datasource_pool_settings
+    from app.core.data_research.catalog.execution import (
+        CatalogIngestionExecutor,
+    )
+    from app.core.data_research.catalog.models import CatalogScope
+    from app.core.data_research.catalog.mysql import MySqlCatalogCollector
+    from app.core.data_research.catalog.postgresql import (
+        PostgreSqlCatalogCollector,
+    )
+    from app.core.data_research.catalog.service import CatalogCollectionService
+    from app.core.data_research.ingestion import IngestionService
+    from app.core.data_research.repository import (
+        SqlAlchemyCatalogSnapshotRepository,
+        SqlAlchemyIngestionJobRepository,
+    )
+    from app.core.data_source.adapters import adapter_for
+    from app.core.data_source.manager import DataSourceConnectionManager
+    from app.core.data_source.models import (
+        DataSourceCredential,
+        DataSourceDefinition,
+    )
+    from app.core.data_source.pool_registry import PoolRegistry
+    from app.models.data_research import (
+        CatalogSnapshot,
+        EvidenceFragment,
+        IngestionJob,
+        IngestionSource,
+    )
+
+    parsed = make_url(source_url)
+    source_uid = str(uuid.uuid4())
+    definition = DataSourceDefinition(
+        uid=source_uid,
+        name_en=f"acceptance-{database_type}",
+        database_type=database_type,
+        host=parsed.host,
+        port=parsed.port,
+        database=parsed.database,
+        schema="public",
+        credential_ref=source_uid,
+        credential_version=1,
+    )
+    credential = DataSourceCredential(
+        username=parsed.username,
+        password=parsed.password,
+    )
+    settings = datasource_pool_settings()
+    manager = DataSourceConnectionManager(
+        definitions=DefinitionRepository(definition),
+        credentials=CredentialRepository(credential),
+        platform_session=lambda: object(),
+        adapter_resolver=adapter_for,
+        registry=PoolRegistry({**settings, "drain_timeout": 30}),
+        settings_resolver=datasource_pool_settings,
+    )
+    app = create_app()
+    app.config.update(TESTING=True)
+    job_uid = None
+    try:
+        with app.app_context():
+            db.session.add(
+                IngestionSource(
+                    uid=source_uid,
+                    source_type="database",
+                    name=f"验收 {database_type}",
+                    config={
+                        "database_type": database_type,
+                        "database": parsed.database,
+                        "schema": "public",
+                    },
+                    permission_scope={},
+                    status="active",
+                    created_by="integration-test",
+                )
+            )
+            db.session.commit()
+            ingestion = IngestionService(
+                SqlAlchemyIngestionJobRepository(db.session),
+                commit=db.session.commit,
+                rollback=db.session.rollback,
+            )
+            job, _created = ingestion.create_job(
+                {
+                    "source_uid": source_uid,
+                    "job_type": "catalog_collect",
+                    "parser_version": "catalog-v1",
+                        "parameters": {
+                            "include_schemas": (
+                                ["public"]
+                                if database_type == "postgresql"
+                                else []
+                            ),
+                        "exclude_schemas": [],
+                        "include_tables": ["acceptance_customers"],
+                        "exclude_tables": [],
+                    },
+                    "force_rerun": True,
+                },
+                actor_uid="integration-test",
+            )
+            job_uid = job.uid
+            collection = CatalogCollectionService(
+                manager,
+                definition_resolver=lambda _uid: definition,
+                collector_resolver=lambda _type: (
+                    PostgreSqlCatalogCollector()
+                    if database_type == "postgresql"
+                    else MySqlCatalogCollector()
+                ),
+            )
+            executor = CatalogIngestionExecutor(
+                ingestion,
+                collection,
+                SqlAlchemyCatalogSnapshotRepository(db.session),
+                commit=db.session.commit,
+                rollback=db.session.rollback,
+            )
+
+            completed = executor.execute(job.uid)
+
+            assert completed.status == "awaiting_review"
+            assert completed.attempt_count == 1
+            assert completed.statistics["asset_count"] == 1
+            assert completed.statistics["field_count"] == 2
+            snapshot = db.session.query(CatalogSnapshot).filter_by(
+                job_uid=job.uid
+            ).one()
+            assert snapshot.attempt == 1
+            assert snapshot.snapshot["assets"][0]["name"] == (
+                "acceptance_customers"
+            )
+            evidence = db.session.query(EvidenceFragment).filter_by(
+                job_uid=job.uid
+            ).all()
+            assert {
+                item.locator["column"]
+                for item in evidence
+            } == {"id", "customer_name"}
+            assert all(
+                "password" not in str(item.locator).lower()
+                for item in evidence
+            )
+    finally:
+        manager.close()
+        with app.app_context():
+            if job_uid:
+                db.session.query(IngestionJob).filter_by(
+                    uid=job_uid
+                ).delete(synchronize_session=False)
+            db.session.query(IngestionSource).filter_by(
+                uid=source_uid
+            ).delete(synchronize_session=False)
+            db.session.commit()

tests/test_database_migrations.py

@@ -51,6 +51,7 @@ EXPECTED_UPGRADED_TABLES = {
     "governance_responsibility_scopes",
     "governance_responsibility_assignments",
     "governance_responsibility_audit_events",
+    "catalog_snapshots",
 }
 
 
@@ -78,6 +79,23 @@ def test_data_research_ingestion_migration_is_additive_and_constrained():
     assert "DROP TABLE" not in migration.upper()
 
 
+def test_catalog_execution_migration_adds_attempts_and_snapshots():
+    migration = (
+        ROOT
+        / "migrations"
+        / "versions"
+        / "20260729_280_catalog_ingestion_execution.py"
+    ).read_text(encoding="utf-8")
+
+    assert 'revision = "20260729_280"' in migration
+    assert 'down_revision = "20260729_270"' in migration
+    assert "ADD COLUMN attempt_count" in migration
+    assert "ADD COLUMN failure_stage" in migration
+    assert "CREATE TABLE public.catalog_snapshots" in migration
+    assert "UNIQUE (job_uid, attempt)" in migration
+    assert "DROP TABLE" not in migration.upper()
+
+
 def test_data_element_migration_adds_versioned_governance_tables():
     migration = (
         ROOT

tests/test_datasource_lifecycle_api.py

@@ -27,12 +27,14 @@ class FakeDefinitions:
         self.existing = existing
         self.saved = None
         self.deleted = None
+        self.last_filters = None
         self.fail_save = fail_save
 
     def get(self, uid):
         return self.existing if uid == UID else None
 
-    def list(self, _filters):
+    def list(self, filters):
+        self.last_filters = filters
         return [self.existing] if self.existing else []
 
     def save(self, definition):
@@ -201,6 +203,33 @@ def test_create_stores_secret_only_in_credential_store():
     assert events[0]["event_type"] == "datasource.credential_version_created"
 
 
+def test_list_ignores_blank_text_filters_from_search_form():
+    (
+        service,
+        _session,
+        definitions,
+        _credentials,
+        _adapter,
+        _manager,
+        _events,
+    ) = make_service(existing=existing_definition())
+
+    result = service.list(
+        {
+            "uid": "",
+            "name_en": " ",
+            "name_zh": "",
+            "type": "",
+        }
+    )
+
+    assert result == [definitions.existing]
+    assert definitions.last_filters.uid is None
+    assert definitions.last_filters.name_en is None
+    assert definitions.last_filters.name_zh is None
+    assert definitions.last_filters.database_type is None
+
+
 def test_update_reuses_omitted_credentials_and_invalidates_old_pool():
     (
         service,

tests/test_permission_matrix.py

@@ -33,6 +33,15 @@ def test_data_development_paths_have_specific_write_policies():
     assert permission_for_request(
         "/api/development/v1/ingestion-jobs/job-1/cancel", "POST"
     ) == ("ingestion:run",)
+    assert permission_for_request(
+        "/api/development/v1/ingestion-jobs/job-1/execute", "POST"
+    ) == ("ingestion:run",)
+    assert permission_for_request(
+        "/api/development/v1/ingestion-jobs/job-1/catalog-snapshots", "GET"
+    ) == ("governance:read",)
+    assert permission_for_request(
+        "/api/development/v1/ingestion-jobs/job-1/evidence", "GET"
+    ) == ("governance:read",)
     assert permission_for_request(
         "/api/development/v1/data-elements", "POST"
     ) == ("data-elements:edit",)