feat: deliver V61 structured governance lifecycle

app/api/data_development/routes.py

@@ -26,6 +26,31 @@ def get_ingestion_service():
     )
 
 
+def get_data_element_service():
+    from app.core.data_research.data_elements import DataElementService
+    from app.core.data_research.repository import SqlAlchemyDataElementRepository
+    from app.core.events.outbox import enqueue_outbox
+
+    return DataElementService(
+        SqlAlchemyDataElementRepository(db.session),
+        outbox_enqueue=lambda **event: enqueue_outbox(db.session, **event),
+    )
+
+
+def get_candidate_decision_service():
+    from app.core.data_research.candidate_decisions import CandidateDecisionService
+    from app.core.data_research.data_elements import DataElementService
+    from app.core.data_research.repository import (
+        SqlAlchemyCandidateDecisionRepository,
+        SqlAlchemyDataElementRepository,
+    )
+
+    return CandidateDecisionService(
+        SqlAlchemyCandidateDecisionRepository(db.session),
+        data_elements=DataElementService(SqlAlchemyDataElementRepository(db.session)),
+    )
+
+
 def _identity():
     return getattr(g, "current_user", {}) or {}
 
@@ -49,6 +74,32 @@ def _record(record):
     }
 
 
+def _element(record):
+    return {
+        "uid": str(record.uid),
+        "code": record.code,
+        "status": record.status,
+        "current_version": int(record.current_version),
+        "snapshot": dict(record.snapshot or {}),
+        "created_by": record.created_by,
+        "updated_by": record.updated_by,
+    }
+
+
+def _decision(record):
+    return {
+        "uid": str(record.uid),
+        "candidate_uid": str(record.candidate_uid),
+        "action": record.action,
+        "data_element_uid": (
+            str(record.data_element_uid) if record.data_element_uid else None
+        ),
+        "evidence_uids": list(record.evidence_uids),
+        "actor_uid": record.actor_uid,
+        "reason": record.reason,
+    }
+
+
 def _error(error):
     if isinstance(error, DataResearchError):
         return (
@@ -133,3 +184,53 @@ def cancel_ingestion_job(job_uid):
     except Exception as error:
         return _error(error)
 
+
+@bp.route("/data-elements", methods=["POST"])
+def create_data_element():
+    try:
+        record = get_data_element_service().create_draft(
+            request.get_json(silent=True) or {},
+            actor_uid=_identity().get("id") or _identity().get("sub"),
+        )
+        db.session.commit()
+        return jsonify(success(_element(record))), 201
+    except Exception as error:
+        db.session.rollback()
+        return _error(error)
+
+
+@bp.route("/data-elements/<element_uid>/transitions", methods=["POST"])
+def transition_data_element(element_uid):
+    payload = request.get_json(silent=True) or {}
+    identity = _identity()
+    target_status = str(payload.get("target_status") or "")
+    if (
+        target_status in {"published", "deprecated", "retired"}
+        and "data-elements:publish" not in set(identity.get("permissions") or [])
+    ):
+        return jsonify(failed("权限不足", code=403)), 403
+    try:
+        record = get_data_element_service().transition(
+            element_uid,
+            target_status,
+            expected_version=int(payload.get("expected_version")),
+            actor_uid=identity.get("id") or identity.get("sub"),
+        )
+        db.session.commit()
+        return jsonify(success(_element(record))), 200
+    except Exception as error:
+        db.session.rollback()
+        return _error(error)
+
+
+@bp.route("/candidate-decisions", methods=["POST"])
+def decide_candidates():
+    payload = request.get_json(silent=True) or {}
+    try:
+        records = get_candidate_decision_service().decide(
+            payload.get("decisions") or [],
+            actor_uid=_identity().get("id") or _identity().get("sub"),
+        )
+        return jsonify(success([_decision(record) for record in records])), 200
+    except Exception as error:
+        return _error(error)

app/core/data_research/candidate_decisions.py

@@ -0,0 +1,79 @@
+from __future__ import annotations
+
+from dataclasses import dataclass
+from typing import Any, Callable
+
+from app.core.common.identifiers import new_governance_uid
+from app.core.data_research.errors import CandidateDecisionInvalid
+
+
+ALLOWED_ACTIONS = frozenset({"reuse", "create", "map", "ignore"})
+
+
+@dataclass(frozen=True)
+class CandidateDecision:
+    uid: str
+    candidate_uid: str
+    action: str
+    data_element_uid: str | None
+    evidence_uids: tuple[str, ...]
+    actor_uid: str | None
+    reason: str | None = None
+
+
+class CandidateDecisionService:
+    def __init__(
+        self,
+        repository,
+        *,
+        data_elements,
+        uid_factory: Callable[[], str] = new_governance_uid,
+    ):
+        self.repository = repository
+        self.data_elements = data_elements
+        self.uid_factory = uid_factory
+
+    def _validate(self, item):
+        if not isinstance(item, dict):
+            raise CandidateDecisionInvalid("decision must be an object")
+        action = str(item.get("action") or "").strip()
+        if action not in ALLOWED_ACTIONS:
+            raise CandidateDecisionInvalid(f"unsupported action: {action}")
+        candidate_uid = str(item.get("candidate_uid") or "").strip()
+        candidate = self.repository.get(candidate_uid)
+        if candidate is None:
+            raise CandidateDecisionInvalid(f"candidate {candidate_uid} was not found")
+        data_element_uid = str(item.get("data_element_uid") or "").strip() or None
+        if action in {"reuse", "map"} and not data_element_uid:
+            raise CandidateDecisionInvalid("data_element_uid is required")
+        return item, candidate, action, data_element_uid
+
+    def decide(self, items, *, actor_uid):
+        try:
+            validated = [self._validate(item) for item in items]
+            decisions = []
+            for item, candidate, action, data_element_uid in validated:
+                evidence_uids = tuple(candidate.get("evidence_uids") or ())
+                if action == "create":
+                    payload = dict(candidate.get("normalized_data") or {})
+                    payload["evidence_uids"] = list(evidence_uids)
+                    element = self.data_elements.create_draft(payload, actor_uid)
+                    data_element_uid = str(element.uid)
+                decisions.append(
+                    CandidateDecision(
+                        uid=self.uid_factory(),
+                        candidate_uid=str(candidate["uid"]),
+                        action=action,
+                        data_element_uid=data_element_uid,
+                        evidence_uids=evidence_uids,
+                        actor_uid=actor_uid,
+                        reason=str(item.get("reason") or "").strip() or None,
+                    )
+                )
+            self.repository.save_decisions(decisions)
+            self.repository.commit()
+            return decisions
+        except Exception:
+            self.repository.rollback()
+            raise
+

app/core/data_research/catalog/__init__.py

@@ -0,0 +1,7 @@
+from app.core.data_research.catalog.service import (
+    CatalogCollectionService,
+    diff_snapshots,
+)
+
+__all__ = ["CatalogCollectionService", "diff_snapshots"]
+

app/core/data_research/catalog/base.py

@@ -0,0 +1,67 @@
+from __future__ import annotations
+
+from collections import OrderedDict
+
+from app.core.data_research.catalog.models import (
+    CatalogAsset,
+    CatalogField,
+    CatalogSnapshot,
+)
+
+
+class BaseCatalogCollector:
+    database_type = ""
+
+    @staticmethod
+    def _asset_type(value: str) -> str:
+        return "view" if "VIEW" in str(value).upper() else "table"
+
+    def _snapshot(self, rows, data_source_uid: str):
+        grouped = OrderedDict()
+        for row in rows:
+            schema = str(row["schema_name"])
+            asset_name = str(row["asset_name"])
+            asset_key = f"{data_source_uid}:{schema}.{asset_name}"
+            entry = grouped.setdefault(
+                asset_key,
+                {
+                    "schema": schema,
+                    "name": asset_name,
+                    "asset_type": self._asset_type(row["asset_type"]),
+                    "comment": row.get("asset_comment"),
+                    "fields": [],
+                },
+            )
+            field_name = str(row["column_name"])
+            entry["fields"].append(
+                CatalogField(
+                    key=f"{asset_key}.{field_name}",
+                    schema=schema,
+                    asset=asset_name,
+                    name=field_name,
+                    ordinal_position=int(row["ordinal_position"]),
+                    data_type=str(row["data_type"]),
+                    nullable=str(row["is_nullable"]).upper() == "YES",
+                    default=row.get("column_default"),
+                    comment=row.get("column_comment"),
+                )
+            )
+        assets = tuple(
+            CatalogAsset(
+                key=key,
+                schema=value["schema"],
+                name=value["name"],
+                asset_type=value["asset_type"],
+                comment=value["comment"],
+                fields=tuple(
+                    sorted(value["fields"], key=lambda item: item.ordinal_position)
+                ),
+            )
+            for key, value in grouped.items()
+        )
+        return CatalogSnapshot(
+            data_source_uid=str(data_source_uid),
+            database_type=self.database_type,
+            assets=assets,
+        )
+

app/core/data_research/catalog/models.py

@@ -0,0 +1,69 @@
+from __future__ import annotations
+
+import hashlib
+import json
+from dataclasses import asdict, dataclass, field
+from typing import Any
+
+
+@dataclass(frozen=True)
+class CatalogScope:
+    include_schemas: tuple[str, ...] = ()
+    exclude_schemas: tuple[str, ...] = ()
+    include_tables: tuple[str, ...] = ()
+    exclude_tables: tuple[str, ...] = ()
+
+
+@dataclass(frozen=True)
+class CatalogField:
+    key: str
+    schema: str
+    asset: str
+    name: str
+    ordinal_position: int
+    data_type: str
+    nullable: bool
+    default: str | None = None
+    comment: str | None = None
+
+
+@dataclass(frozen=True)
+class CatalogAsset:
+    key: str
+    schema: str
+    name: str
+    asset_type: str
+    fields: tuple[CatalogField, ...]
+    comment: str | None = None
+
+
+@dataclass(frozen=True)
+class CatalogSnapshot:
+    data_source_uid: str
+    database_type: str
+    assets: tuple[CatalogAsset, ...]
+
+    @property
+    def content_hash(self) -> str:
+        payload = json.dumps(
+            asdict(self),
+            ensure_ascii=False,
+            sort_keys=True,
+            separators=(",", ":"),
+        )
+        return hashlib.sha256(payload.encode("utf-8")).hexdigest()
+
+
+@dataclass(frozen=True)
+class CatalogFieldChange:
+    before: CatalogField
+    after: CatalogField
+
+
+@dataclass(frozen=True)
+class CatalogDiff:
+    added: tuple[CatalogField, ...] = ()
+    removed: tuple[CatalogField, ...] = ()
+    changed: tuple[CatalogFieldChange, ...] = ()
+    renamed: tuple[CatalogFieldChange, ...] = ()
+

app/core/data_research/catalog/mysql.py

@@ -0,0 +1,47 @@
+from __future__ import annotations
+
+from sqlalchemy import bindparam, text
+
+from app.core.data_research.catalog.base import BaseCatalogCollector
+
+
+MYSQL_CATALOG_SQL = """
+SELECT
+    c.TABLE_SCHEMA AS schema_name,
+    c.TABLE_NAME AS asset_name,
+    t.TABLE_TYPE AS asset_type,
+    t.TABLE_COMMENT AS asset_comment,
+    c.COLUMN_NAME AS column_name,
+    c.ORDINAL_POSITION AS ordinal_position,
+    c.COLUMN_TYPE AS data_type,
+    c.IS_NULLABLE AS is_nullable,
+    c.COLUMN_DEFAULT AS column_default,
+    c.COLUMN_COMMENT AS column_comment
+FROM information_schema.COLUMNS c
+JOIN information_schema.TABLES t
+  ON t.TABLE_SCHEMA = c.TABLE_SCHEMA AND t.TABLE_NAME = c.TABLE_NAME
+WHERE c.TABLE_SCHEMA = :database_name
+  AND (:include_table_count = 0 OR c.TABLE_NAME IN :include_tables)
+  AND (:exclude_table_count = 0 OR c.TABLE_NAME NOT IN :exclude_tables)
+ORDER BY c.TABLE_NAME, c.ORDINAL_POSITION
+"""
+
+
+class MySqlCatalogCollector(BaseCatalogCollector):
+    database_type = "mysql"
+
+    def collect(self, connection, *, data_source_uid, database_name, scope):
+        statement = text(MYSQL_CATALOG_SQL).bindparams(
+            bindparam("include_tables", expanding=True),
+            bindparam("exclude_tables", expanding=True),
+        )
+        parameters = {
+            "database_name": str(database_name),
+            "include_tables": list(scope.include_tables),
+            "exclude_tables": list(scope.exclude_tables),
+            "include_table_count": len(scope.include_tables),
+            "exclude_table_count": len(scope.exclude_tables),
+        }
+        rows = connection.execute(statement, parameters).mappings().all()
+        return self._snapshot(rows, str(data_source_uid))
+

app/core/data_research/catalog/postgresql.py

@@ -0,0 +1,57 @@
+from __future__ import annotations
+
+from sqlalchemy import bindparam, text
+
+from app.core.data_research.catalog.base import BaseCatalogCollector
+
+
+POSTGRESQL_CATALOG_SQL = """
+SELECT
+    c.table_schema AS schema_name,
+    c.table_name AS asset_name,
+    t.table_type AS asset_type,
+    obj_description(pc.oid) AS asset_comment,
+    c.column_name,
+    c.ordinal_position,
+    c.data_type,
+    c.is_nullable,
+    c.column_default,
+    col_description(pc.oid, c.ordinal_position) AS column_comment
+FROM information_schema.columns c
+JOIN information_schema.tables t
+  ON t.table_schema = c.table_schema AND t.table_name = c.table_name
+LEFT JOIN pg_catalog.pg_namespace pn ON pn.nspname = c.table_schema
+LEFT JOIN pg_catalog.pg_class pc
+  ON pc.relnamespace = pn.oid AND pc.relname = c.table_name
+WHERE c.table_schema NOT IN ('pg_catalog', 'information_schema')
+  AND (:include_schema_count = 0 OR c.table_schema IN :include_schemas)
+  AND (:exclude_schema_count = 0 OR c.table_schema NOT IN :exclude_schemas)
+  AND (:include_table_count = 0 OR c.table_name IN :include_tables)
+  AND (:exclude_table_count = 0 OR c.table_name NOT IN :exclude_tables)
+ORDER BY c.table_schema, c.table_name, c.ordinal_position
+"""
+
+
+class PostgreSqlCatalogCollector(BaseCatalogCollector):
+    database_type = "postgresql"
+
+    def collect(self, connection, *, data_source_uid, scope):
+        statement = text(POSTGRESQL_CATALOG_SQL).bindparams(
+            bindparam("include_schemas", expanding=True),
+            bindparam("exclude_schemas", expanding=True),
+            bindparam("include_tables", expanding=True),
+            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_schema_count": len(scope.include_schemas),
+            "exclude_schema_count": len(scope.exclude_schemas),
+            "include_table_count": len(scope.include_tables),
+            "exclude_table_count": len(scope.exclude_tables),
+        }
+        rows = connection.execute(statement, parameters).mappings().all()
+        return self._snapshot(rows, str(data_source_uid))
+

app/core/data_research/catalog/service.py

@@ -0,0 +1,99 @@
+from __future__ import annotations
+
+from app.core.data_research.catalog.models import (
+    CatalogDiff,
+    CatalogFieldChange,
+)
+
+
+def _fields(snapshot):
+    return {
+        field.key: field
+        for asset in snapshot.assets
+        for field in asset.fields
+    }
+
+
+def diff_snapshots(previous, current):
+    before = _fields(previous)
+    after = _fields(current)
+    changed = []
+    for key in sorted(set(before) & set(after)):
+        left = before[key]
+        right = after[key]
+        if (
+            left.data_type,
+            left.nullable,
+            left.default,
+            left.comment,
+        ) != (
+            right.data_type,
+            right.nullable,
+            right.default,
+            right.comment,
+        ):
+            changed.append(CatalogFieldChange(left, right))
+
+    removed = [before[key] for key in sorted(set(before) - set(after))]
+    added = [after[key] for key in sorted(set(after) - set(before))]
+    renamed = []
+    used_added = set()
+    used_removed = set()
+    for old in removed:
+        for new in added:
+            if new.key in used_added:
+                continue
+            if (
+                old.schema,
+                old.asset,
+                old.ordinal_position,
+                old.data_type,
+            ) == (
+                new.schema,
+                new.asset,
+                new.ordinal_position,
+                new.data_type,
+            ):
+                renamed.append(CatalogFieldChange(old, new))
+                used_removed.add(old.key)
+                used_added.add(new.key)
+                break
+    return CatalogDiff(
+        added=tuple(item for item in added if item.key not in used_added),
+        removed=tuple(item for item in removed if item.key not in used_removed),
+        changed=tuple(changed),
+        renamed=tuple(renamed),
+    )
+
+
+class CatalogCollectionService:
+    def __init__(self, manager, *, definition_resolver, collector_resolver):
+        self.manager = manager
+        self.definition_resolver = definition_resolver
+        self.collector_resolver = collector_resolver
+
+    def collect(self, data_source_uid, scope):
+        definition = self.definition_resolver(str(data_source_uid))
+        database_type = (
+            definition.get("database_type")
+            if isinstance(definition, dict)
+            else definition.database_type
+        )
+        database_name = (
+            definition.get("database")
+            if isinstance(definition, dict)
+            else definition.database
+        )
+        collector = self.collector_resolver(database_type)
+        with self.manager.connect(
+            str(data_source_uid),
+            "metadata_collection",
+        ) as connection:
+            kwargs = {
+                "data_source_uid": str(data_source_uid),
+                "scope": scope,
+            }
+            if database_type == "mysql":
+                kwargs["database_name"] = database_name
+            return collector.collect(connection, **kwargs)
+

app/core/data_research/data_elements.py

@@ -0,0 +1,118 @@
+from __future__ import annotations
+
+from dataclasses import dataclass, replace
+from typing import Any, Callable
+
+from app.core.common.identifiers import new_governance_uid
+from app.core.data_research.errors import (
+    DataElementInvalid,
+    DataElementNotFound,
+    DataElementTransitionInvalid,
+    DataElementVersionConflict,
+)
+
+
+ELEMENT_TRANSITIONS = {
+    "draft": frozenset({"in_review"}),
+    "in_review": frozenset({"draft", "published"}),
+    "published": frozenset({"deprecated"}),
+    "deprecated": frozenset({"published", "retired"}),
+    "retired": frozenset(),
+}
+
+
+@dataclass(frozen=True)
+class DataElementRecord:
+    uid: str
+    code: str
+    status: str
+    current_version: int
+    snapshot: dict[str, Any]
+    created_by: str | None = None
+    updated_by: str | None = None
+
+
+def _required(payload, name):
+    value = str(payload.get(name) or "").strip()
+    if not value:
+        raise DataElementInvalid(f"{name} is required")
+    return value
+
+
+class DataElementService:
+    def __init__(
+        self,
+        repository,
+        *,
+        uid_factory: Callable[[], str] = new_governance_uid,
+        outbox_enqueue: Callable[..., Any] = lambda **_event: None,
+    ):
+        self.repository = repository
+        self.uid_factory = uid_factory
+        self.outbox_enqueue = outbox_enqueue
+
+    def create_draft(self, payload, actor_uid):
+        if not isinstance(payload, dict):
+            raise DataElementInvalid("payload must be an object")
+        snapshot = {
+            "code": _required(payload, "code"),
+            "name_zh": _required(payload, "name_zh"),
+            "name_en": _required(payload, "name_en"),
+            "definition": _required(payload, "definition"),
+            "logical_type": _required(payload, "logical_type"),
+            "evidence_uids": list(payload.get("evidence_uids") or []),
+            "business_domain_uids": list(payload.get("business_domain_uids") or []),
+        }
+        for name in (
+            "aliases",
+            "format",
+            "value_domain",
+            "unit",
+            "sensitivity",
+            "quality_rules",
+        ):
+            if name in payload:
+                snapshot[name] = payload[name]
+        if self.repository.get_by_code(snapshot["code"]) is not None:
+            raise DataElementInvalid("code already exists")
+        return self.repository.save(
+            DataElementRecord(
+                uid=self.uid_factory(),
+                code=snapshot["code"],
+                status="draft",
+                current_version=1,
+                snapshot=snapshot,
+                created_by=actor_uid,
+                updated_by=actor_uid,
+            )
+        )
+
+    def transition(self, uid, target_status, *, expected_version, actor_uid):
+        record = self.repository.get(str(uid))
+        if record is None:
+            raise DataElementNotFound(f"data element {uid} was not found")
+        if int(expected_version) != int(record.current_version):
+            raise DataElementVersionConflict(
+                f"expected version {expected_version}, current version {record.current_version}"
+            )
+        if target_status not in ELEMENT_TRANSITIONS.get(record.status, frozenset()):
+            raise DataElementTransitionInvalid(
+                f"{record.status} -> {target_status} is not allowed"
+            )
+        updated = self.repository.save(
+            replace(
+                record,
+                status=target_status,
+                current_version=record.current_version + 1,
+                updated_by=actor_uid,
+            )
+        )
+        if target_status == "published":
+            self.outbox_enqueue(
+                aggregate_type="data_element",
+                aggregate_id=updated.uid,
+                event_type="data_element.version_published",
+                payload={"uid": updated.uid, "version": updated.current_version},
+            )
+        return updated
+

app/core/data_research/errors.py

@@ -16,3 +16,30 @@ class IngestionJobNotFound(DataResearchError):
     code = "INGESTION_JOB_NOT_FOUND"
     http_status = 404
 
+
+class DataElementInvalid(DataResearchError):
+    code = "DATA_ELEMENT_INVALID"
+
+
+class DataElementNotFound(DataResearchError):
+    code = "DATA_ELEMENT_NOT_FOUND"
+    http_status = 404
+
+
+class DataElementVersionConflict(DataResearchError):
+    code = "DATA_ELEMENT_VERSION_CONFLICT"
+    http_status = 409
+
+
+class DataElementTransitionInvalid(DataResearchError):
+    code = "DATA_ELEMENT_TRANSITION_INVALID"
+    http_status = 409
+
+
+class CandidateDecisionInvalid(DataResearchError):
+    code = "CANDIDATE_DECISION_INVALID"
+
+
+class DataElementProjectionInvalid(DataResearchError):
+    code = "DATA_ELEMENT_PROJECTION_INVALID"
+    http_status = 422

app/core/data_research/extractors/__init__.py

@@ -0,0 +1,4 @@
+from app.core.data_research.extractors.registry import default_registry
+
+__all__ = ["default_registry"]
+

app/core/data_research/extractors/base.py

@@ -0,0 +1,63 @@
+from __future__ import annotations
+
+import hashlib
+from dataclasses import dataclass, field
+from typing import Any
+
+
+@dataclass(frozen=True)
+class ExtractionContext:
+    filename: str
+    media_type: str
+    max_bytes: int = 25 * 1024 * 1024
+
+
+@dataclass(frozen=True)
+class ExtractionEvidence:
+    locator: dict[str, Any]
+    excerpt: str | None = None
+    confidence: float | None = None
+
+
+@dataclass(frozen=True)
+class ExtractedItem:
+    candidate_type: str
+    data: dict[str, Any]
+    evidence: tuple[ExtractionEvidence, ...] = ()
+    confidence: float = 1.0
+
+
+@dataclass(frozen=True)
+class ExtractionBatch:
+    parser_version: str
+    content_hash: str
+    items: tuple[ExtractedItem, ...]
+    warnings: tuple[str, ...] = ()
+    metadata: dict[str, Any] = field(default_factory=dict)
+
+
+class BaseExtractor:
+    parser_version = ""
+
+    def can_handle(self, media_type: str, filename: str) -> bool:
+        raise NotImplementedError
+
+    def extract(self, content: bytes, context: ExtractionContext) -> ExtractionBatch:
+        raise NotImplementedError
+
+    def _batch(
+        self,
+        content: bytes,
+        items: list[ExtractedItem],
+        *,
+        warnings: list[str] | None = None,
+        metadata: dict[str, Any] | None = None,
+    ) -> ExtractionBatch:
+        return ExtractionBatch(
+            parser_version=self.parser_version,
+            content_hash=hashlib.sha256(content).hexdigest(),
+            items=tuple(items),
+            warnings=tuple(warnings or ()),
+            metadata=dict(metadata or {}),
+        )
+

app/core/data_research/extractors/csv.py

@@ -0,0 +1,97 @@
+from __future__ import annotations
+
+import csv
+import io
+import re
+from pathlib import Path
+
+from app.core.data_research.extractors.base import (
+    BaseExtractor,
+    ExtractedItem,
+    ExtractionContext,
+    ExtractionEvidence,
+)
+
+
+_INTEGER = re.compile(r"^[+-]?\d+$")
+_DECIMAL = re.compile(r"^[+-]?(?:\d+\.\d*|\d*\.\d+)$")
+_BOOLEAN = frozenset({"true", "false", "yes", "no"})
+
+
+def infer_type(values) -> str:
+    present = [str(value).strip() for value in values if str(value).strip()]
+    if not present:
+        return "string"
+    lowered = [value.lower() for value in present]
+    if all(value in _BOOLEAN for value in lowered):
+        return "boolean"
+    if all(_INTEGER.fullmatch(value) for value in present):
+        return "integer"
+    if all(_INTEGER.fullmatch(value) or _DECIMAL.fullmatch(value) for value in present):
+        return "decimal"
+    return "string"
+
+
+class CsvExtractor(BaseExtractor):
+    parser_version = "csv-v1"
+
+    def can_handle(self, media_type: str, filename: str) -> bool:
+        return filename.lower().endswith(".csv") or media_type in {
+            "text/csv",
+            "application/csv",
+        }
+
+    def extract(self, content: bytes, context: ExtractionContext):
+        try:
+            text = content.decode("utf-8-sig")
+        except UnicodeDecodeError as exc:
+            raise ValueError("CSV file must be UTF-8 encoded") from exc
+        if not text.strip():
+            raise ValueError("CSV header is required")
+        try:
+            dialect = csv.Sniffer().sniff(text[:8192], delimiters=",;\t|")
+        except csv.Error:
+            dialect = csv.excel
+        rows = list(csv.reader(io.StringIO(text), dialect))
+        if not rows or not any(str(value).strip() for value in rows[0]):
+            raise ValueError("CSV header is required")
+        headers = [str(value).strip() for value in rows[0]]
+        data_rows = rows[1:]
+        columns = []
+        for index, name in enumerate(headers, start=1):
+            values = [row[index - 1] if len(row) >= index else "" for row in data_rows]
+            columns.append(
+                {
+                    "name_zh": "",
+                    "name_en": name,
+                    "data_type": infer_type(values),
+                    "nullable": any(not str(value).strip() for value in values),
+                    "evidence_locator": {
+                        "kind": "csv.header",
+                        "column": index,
+                        "name": name,
+                    },
+                }
+            )
+        item = ExtractedItem(
+            candidate_type="physical_asset",
+            data={
+                "table_info": {
+                    "name_zh": "",
+                    "name_en": Path(context.filename).stem,
+                },
+                "columns": columns,
+            },
+            evidence=(
+                ExtractionEvidence(
+                    locator={"kind": "csv.header", "row": 1},
+                    excerpt=",".join(headers),
+                ),
+            ),
+        )
+        return self._batch(
+            content,
+            [item],
+            metadata={"row_count": len(data_rows), "delimiter": dialect.delimiter},
+        )
+

app/core/data_research/extractors/excel.py

@@ -0,0 +1,79 @@
+from __future__ import annotations
+
+import io
+
+import pandas as pd
+
+from app.core.data_research.extractors.base import (
+    BaseExtractor,
+    ExtractedItem,
+    ExtractionContext,
+    ExtractionEvidence,
+)
+
+
+def _column_letter(index: int) -> str:
+    result = ""
+    while index:
+        index, remainder = divmod(index - 1, 26)
+        result = chr(65 + remainder) + result
+    return result
+
+
+def _pandas_type(series) -> str:
+    if pd.api.types.is_bool_dtype(series.dtype):
+        return "boolean"
+    if pd.api.types.is_integer_dtype(series.dtype):
+        return "integer"
+    if pd.api.types.is_float_dtype(series.dtype):
+        return "decimal"
+    if pd.api.types.is_datetime64_any_dtype(series.dtype):
+        return "datetime"
+    return "string"
+
+
+class ExcelExtractor(BaseExtractor):
+    parser_version = "excel-v1"
+
+    def can_handle(self, media_type: str, filename: str) -> bool:
+        lowered = filename.lower()
+        return lowered.endswith((".xlsx", ".xls")) or "spreadsheet" in media_type or "excel" in media_type
+
+    def extract(self, content: bytes, context: ExtractionContext):
+        workbook = pd.ExcelFile(io.BytesIO(content))
+        items = []
+        for sheet_name in workbook.sheet_names:
+            frame = pd.read_excel(workbook, sheet_name=sheet_name)
+            columns = []
+            last_row = max(1, len(frame) + 1)
+            for index, name in enumerate(frame.columns, start=1):
+                letter = _column_letter(index)
+                columns.append(
+                    {
+                        "name_zh": "",
+                        "name_en": str(name),
+                        "data_type": _pandas_type(frame[name]),
+                        "nullable": bool(frame[name].isna().any()),
+                        "evidence_locator": {
+                            "kind": "excel.cell_range",
+                            "sheet": sheet_name,
+                            "range": f"{letter}1:{letter}{last_row}",
+                        },
+                    }
+                )
+            items.append(
+                ExtractedItem(
+                    candidate_type="physical_asset",
+                    data={
+                        "table_info": {"name_zh": "", "name_en": sheet_name},
+                        "columns": columns,
+                    },
+                    evidence=(
+                        ExtractionEvidence(
+                            locator={"kind": "excel.sheet", "sheet": sheet_name},
+                        ),
+                    ),
+                )
+            )
+        return self._batch(content, items, metadata={"sheet_count": len(items)})
+

app/core/data_research/extractors/registry.py

@@ -0,0 +1,30 @@
+from __future__ import annotations
+
+from app.core.data_research.extractors.base import ExtractionContext
+from app.core.data_research.extractors.csv import CsvExtractor
+from app.core.data_research.extractors.excel import ExcelExtractor
+from app.core.data_research.extractors.sql import SqlExtractor
+
+
+class ExtractorRegistry:
+    def __init__(self, extractors):
+        self.extractors = tuple(extractors)
+
+    def resolve(self, media_type: str, filename: str):
+        for extractor in self.extractors:
+            if extractor.can_handle(media_type, filename):
+                return extractor
+        raise ValueError(f"unsupported source format: {filename}")
+
+    def extract(self, content: bytes, context: ExtractionContext):
+        if len(content) > int(context.max_bytes):
+            raise ValueError("source exceeds configured size limit")
+        return self.resolve(context.media_type, context.filename).extract(
+            content,
+            context,
+        )
+
+
+def default_registry():
+    return ExtractorRegistry((SqlExtractor(), CsvExtractor(), ExcelExtractor()))
+

app/core/data_research/extractors/sql.py

@@ -0,0 +1,42 @@
+from __future__ import annotations
+
+from app.core.data_research.extractors.base import (
+    BaseExtractor,
+    ExtractedItem,
+    ExtractionContext,
+    ExtractionEvidence,
+)
+from app.core.llm.ddl_parser import DDLParser
+
+
+class SqlExtractor(BaseExtractor):
+    parser_version = "sql-ddl-v1"
+
+    def can_handle(self, media_type: str, filename: str) -> bool:
+        return filename.lower().endswith(".sql") or media_type in {
+            "application/sql",
+            "text/sql",
+        }
+
+    def extract(self, content: bytes, context: ExtractionContext):
+        try:
+            sql_text = content.decode("utf-8-sig")
+        except UnicodeDecodeError as exc:
+            raise ValueError("SQL file must be UTF-8 encoded") from exc
+        tables = DDLParser()._parse_sql_ddl_fallback(sql_text)
+        if not tables:
+            raise ValueError("no supported CREATE TABLE or CREATE VIEW statements found")
+        items = [
+            ExtractedItem(
+                candidate_type="physical_asset",
+                data=table,
+                evidence=(
+                    ExtractionEvidence(
+                        locator={"kind": "sql.statement", "statement": index},
+                    ),
+                ),
+            )
+            for index, table in enumerate(tables, start=1)
+        ]
+        return self._batch(content, items)
+

app/core/data_research/graph_projection.py

@@ -0,0 +1,32 @@
+from __future__ import annotations
+
+from app.core.data_research.errors import DataElementProjectionInvalid
+
+
+class DataElementGraphProjector:
+    """Idempotent effective-graph projection for published data elements."""
+
+    STATEMENT = """
+    MERGE (element:DataElement {uid: $uid})
+    SET element.code = $code,
+        element.name_zh = $name_zh,
+        element.version = $version,
+        element.status = 'published'
+    """
+
+    def __init__(self, graph_session):
+        self.graph_session = graph_session
+
+    def project(self, payload):
+        if str(payload.get("status") or "") != "published":
+            raise DataElementProjectionInvalid(
+                "only a published data element version can be projected"
+            )
+        snapshot = dict(payload.get("snapshot") or {})
+        self.graph_session.run(
+            self.STATEMENT,
+            uid=str(payload["uid"]),
+            version=int(payload["version"]),
+            code=str(snapshot.get("code") or ""),
+            name_zh=str(snapshot.get("name_zh") or ""),
+        )

app/core/data_research/repository.py

@@ -2,7 +2,13 @@ from __future__ import annotations
 
 from dataclasses import replace
 
-from app.models.data_research import IngestionJob
+from app.models.data_research import (
+    CandidateDecisionRecord,
+    DataElement,
+    DataElementVersion,
+    ExtractionCandidate,
+    IngestionJob,
+)
 
 
 class SqlAlchemyIngestionJobRepository:
@@ -92,3 +98,94 @@ class SqlAlchemyIngestionJobRepository:
             setattr(model, name, getattr(record, name))
         self.session.flush()
         return self._record(model)
+
+
+class SqlAlchemyDataElementRepository:
+    def __init__(self, session):
+        self.session = session
+
+    @staticmethod
+    def _record(model):
+        from app.core.data_research.data_elements import DataElementRecord
+
+        version = max(model.versions, key=lambda item: item.version)
+        return DataElementRecord(
+            uid=str(model.uid),
+            code=model.code,
+            status=model.status,
+            current_version=int(model.current_version),
+            snapshot=dict(version.snapshot or {}),
+            created_by=model.created_by,
+            updated_by=version.created_by,
+        )
+
+    def get(self, uid):
+        model = self.session.get(DataElement, str(uid))
+        return self._record(model) if model is not None else None
+
+    def get_by_code(self, code):
+        model = self.session.query(DataElement).filter_by(code=str(code)).first()
+        return self._record(model) if model is not None else None
+
+    def save(self, record):
+        model = self.session.get(DataElement, str(record.uid))
+        if model is None:
+            model = DataElement(
+                uid=record.uid,
+                code=record.code,
+                current_version=record.current_version,
+                status=record.status,
+                business_domain_uids=list(
+                    record.snapshot.get("business_domain_uids") or []
+                ),
+                created_by=record.created_by,
+            )
+            self.session.add(model)
+        else:
+            model.current_version = record.current_version
+            model.status = record.status
+            model.business_domain_uids = list(
+                record.snapshot.get("business_domain_uids") or []
+            )
+        self.session.add(
+            DataElementVersion(
+                data_element_uid=record.uid,
+                version=record.current_version,
+                status=record.status,
+                snapshot=record.snapshot,
+                evidence_uids=list(record.snapshot.get("evidence_uids") or []),
+                created_by=record.updated_by or record.created_by,
+            )
+        )
+        self.session.flush()
+        return record
+
+
+class SqlAlchemyCandidateDecisionRepository:
+    def __init__(self, session):
+        self.session = session
+
+    def get(self, uid):
+        model = self.session.get(ExtractionCandidate, str(uid))
+        return model.to_dict() if model is not None else None
+
+    def save_decisions(self, decisions):
+        for decision in decisions:
+            self.session.add(
+                CandidateDecisionRecord(
+                    uid=decision.uid,
+                    candidate_uid=decision.candidate_uid,
+                    action=decision.action,
+                    data_element_uid=decision.data_element_uid,
+                    evidence_uids=list(decision.evidence_uids),
+                    actor_uid=decision.actor_uid,
+                    reason=decision.reason,
+                )
+            )
+        self.session.flush()
+
+    def commit(self):
+        self.session.commit()
+
+    def rollback(self):
+        self.session.rollback()

app/core/system/permissions.py

@@ -75,6 +75,12 @@ def permission_for_request(path: str, method: str) -> tuple[str, ...]:
             return (INGESTION_RUN,)
         if method == "POST":
             return (INGESTION_RUN,)
+    if path.startswith("/api/development/v1/data-elements"):
+        if method == "GET":
+            return (READ_GOVERNANCE,)
+        return (DATA_ELEMENTS_EDIT,)
+    if path.startswith("/api/development/v1/candidate-decisions"):
+        return (DATA_ELEMENTS_EDIT,)
     if path.startswith("/api/system/workbench"):
         return (READ_GOVERNANCE,)
     if (

app/models/__init__.py

@@ -8,6 +8,9 @@ from app.models.data_research import (
     IngestionJob,
     IngestionSource,
     SourceArtifact,
+    DataElement,
+    DataElementVersion,
+    CandidateDecisionRecord,
 )
 
 __all__ = [
@@ -20,4 +23,7 @@ __all__ = [
     "SourceArtifact",
     "EvidenceFragment",
     "ExtractionCandidate",
+    "DataElement",
+    "DataElementVersion",
+    "CandidateDecisionRecord",
 ]

app/models/data_research.py

@@ -239,3 +239,82 @@ class ExtractionCandidate(db.Model):
             "updated_at": _iso(self.updated_at),
         }
 
+
+class DataElement(db.Model):
+    __tablename__ = "data_elements"
+    __table_args__ = (
+        db.CheckConstraint(
+            "status IN ('draft','in_review','published','deprecated','retired')",
+            name="ck_data_element_status",
+        ),
+        {"schema": "public"},
+    )
+
+    uid = db.Column(UUID(as_uuid=False), primary_key=True, default=new_governance_uid)
+    code = db.Column(db.String(120), nullable=False, unique=True)
+    current_version = db.Column(db.Integer, nullable=False, default=1)
+    status = db.Column(db.String(20), nullable=False, default="draft")
+    business_domain_uids = db.Column(JSONB, nullable=False, default=list)
+    created_by = db.Column(db.String(100))
+    created_at = db.Column(db.DateTime, nullable=False, default=now_china_naive)
+    updated_at = db.Column(db.DateTime, nullable=False, default=now_china_naive)
+    versions = db.relationship(
+        "DataElementVersion",
+        lazy="selectin",
+        order_by="DataElementVersion.version",
+        cascade="all, delete-orphan",
+    )
+
+
+class DataElementVersion(db.Model):
+    __tablename__ = "data_element_versions"
+    __table_args__ = (
+        db.UniqueConstraint(
+            "data_element_uid",
+            "version",
+            name="uq_data_element_version",
+        ),
+        {"schema": "public"},
+    )
+
+    uid = db.Column(UUID(as_uuid=False), primary_key=True, default=new_governance_uid)
+    data_element_uid = db.Column(
+        UUID(as_uuid=False),
+        db.ForeignKey("public.data_elements.uid"),
+        nullable=False,
+    )
+    version = db.Column(db.Integer, nullable=False)
+    status = db.Column(db.String(20), nullable=False)
+    snapshot = db.Column(JSONB, nullable=False)
+    evidence_uids = db.Column(JSONB, nullable=False, default=list)
+    created_by = db.Column(db.String(100))
+    created_at = db.Column(db.DateTime, nullable=False, default=now_china_naive)
+
+
+class CandidateDecisionRecord(db.Model):
+    __tablename__ = "candidate_decisions"
+    __table_args__ = (
+        db.CheckConstraint(
+            "action IN ('reuse','create','map','ignore')",
+            name="ck_candidate_decision_action",
+        ),
+        db.UniqueConstraint("candidate_uid", name="uq_candidate_decision"),
+        {"schema": "public"},
+    )
+
+    uid = db.Column(UUID(as_uuid=False), primary_key=True, default=new_governance_uid)
+    candidate_uid = db.Column(
+        UUID(as_uuid=False),
+        db.ForeignKey("public.extraction_candidates.uid"),
+        nullable=False,
+    )
+    action = db.Column(db.String(20), nullable=False)
+    data_element_uid = db.Column(
+        UUID(as_uuid=False),
+        db.ForeignKey("public.data_elements.uid"),
+    )
+    evidence_uids = db.Column(JSONB, nullable=False, default=list)
+    payload = db.Column(JSONB, nullable=False, default=dict)
+    actor_uid = db.Column(db.String(100))
+    reason = db.Column(db.String(1000))
+    created_at = db.Column(db.DateTime, nullable=False, default=now_china_naive)

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: 121
+x-route-count: 124
 servers:
   - url: "http://localhost:15500"
     description: "全本地隔离测试后端"
@@ -1642,6 +1642,90 @@ paths:
             application/json:
               schema:
                 $ref: '#/components/schemas/ApiEnvelope'
+  "/api/development/v1/candidate-decisions":
+    post:
+      tags: [data_development]
+      operationId: data_development_decide_candidates_post
+      summary: "decide candidates"
+      x-source: "app/api/data_development/routes.py"
+      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/data-elements":
+    post:
+      tags: [data_development]
+      operationId: data_development_create_data_element_post
+      summary: "create data element"
+      x-source: "app/api/data_development/routes.py"
+      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/data-elements/{element_uid}/transitions":
+    post:
+      tags: [data_development]
+      operationId: data_development_transition_data_element_post
+      summary: "transition data element"
+      x-source: "app/api/data_development/routes.py"
+      parameters:
+        - name: element_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":
     get:
       tags: [data_development]

docs/superpowers/plans/2026-07-22-data-research-v60-v65-delivery-plan.md

@@ -85,7 +85,7 @@
 
 - [x] Run `PYTHONPATH=. .venv/bin/pytest -q tests/data_research tests/test_database_migrations.py tests/test_permission_matrix.py tests/test_architecture_artifacts.py`.
 - [x] Run `PYTHONPATH=. .venv/bin/pytest -q` and `git diff --check`.
-- [ ] Record evidence in `docs/validation/data-research-v60.md` and commit V60.
+- [x] Record evidence in `docs/validation/data-research-v60.md` and commit V60.
 
 ## V61 — Database catalog and structured files
 
@@ -103,10 +103,10 @@
 - Produces: `Extractor.can_handle(media_type, filename)`, `extract(stream, context) -> ExtractionBatch`, and evidence locators `sql.statement`, `csv.header`, `excel.sheet/cell_range`.
 - Consumes: existing `DDLParser` only as SQL fallback; deterministic extraction runs first.
 
-- [ ] Write tests for dialect-preserving DDL, multi-table SQL, BOM/delimiter CSV, empty/header-only CSV, multi-sheet Excel, parser version, content hash, and exact evidence positions.
-- [ ] Run the extractor tests and verify missing-registry failures.
-- [ ] Implement bounded streaming extractors and registry selection without reading arbitrary large files into unbounded memory.
-- [ ] Rerun focused tests and existing `tests/test_ddl_parser_view.py`.
+- [x] Write tests for dialect-preserving DDL, multi-table SQL, BOM/delimiter CSV, empty/header-only CSV, multi-sheet Excel, parser version, content hash, and exact evidence positions.
+- [x] Run the extractor tests and verify missing-registry failures.
+- [x] Implement bounded streaming extractors and registry selection without reading arbitrary large files into unbounded memory.
+- [x] Rerun focused tests and existing `tests/test_ddl_parser_view.py`.
 
 ### Task 5: Add PostgreSQL and MySQL catalog collectors with snapshot diff
 
@@ -122,10 +122,10 @@
 - Produces: `CatalogCollectionService.collect(data_source_uid, scope) -> CatalogSnapshot` and `diff(previous, current) -> CatalogDiff`.
 - Consumes: `DataSourceConnectionManager.connect(uid, "metadata_collection")` and allowlisted schema/table filters.
 
-- [ ] Write unit tests for generated allowlisted catalog statements, schema/table exclusions, stable field keys, no business-row reads, add/remove/type/rename diffs, and query timeout propagation.
-- [ ] Run unit tests and verify missing collector failures.
-- [ ] Implement dialect collectors and pure snapshot differ; never concatenate untrusted identifiers into SQL.
-- [ ] Run unit tests and, when source containers are available, the integration catalog test.
+- [x] Write unit tests for generated allowlisted catalog statements, schema/table exclusions, stable field keys, no business-row reads, add/remove/type/rename diffs, and query timeout propagation.
+- [x] Run unit tests and verify missing collector failures.
+- [x] Implement dialect collectors and pure snapshot differ; never concatenate untrusted identifiers into SQL.
+- [x] Run unit tests and, when source containers are available, the integration catalog test.
 
 ### Task 6: Implement data-element lifecycle and candidate decisions
 
@@ -142,16 +142,16 @@
 - Produces: lifecycle `candidate -> draft -> in_review -> published -> deprecated -> retired`.
 - Produces: batch decisions `reuse`, `create`, `map`, `ignore`, and outbox event `data_element.version_published`.
 
-- [ ] Write tests for required definitions, stable UID/version increments, optimistic concurrency, invalid lifecycle moves, batch decision atomicity, evidence retention, and no direct candidate publication.
-- [ ] Run focused tests and verify missing service failures.
-- [ ] Implement lifecycle/version services, extend API endpoints, and emit outbox events within the PostgreSQL transaction.
-- [ ] Rerun focused tests plus outbox and cross-store consistency tests.
+- [x] Write tests for required definitions, stable UID/version increments, optimistic concurrency, invalid lifecycle moves, batch decision atomicity, evidence retention, and no direct candidate publication.
+- [x] Run focused tests and verify missing service failures.
+- [x] Implement lifecycle/version services, extend API endpoints, and emit outbox events within the PostgreSQL transaction.
+- [x] Rerun focused tests plus outbox and cross-store consistency tests.
 
 ### V61 gate
 
-- [ ] Run all V61-focused tests and `tests/core/data_source`.
-- [ ] Run the complete Python suite and `git diff --check`.
-- [ ] Record evidence in `docs/validation/data-research-v61.md` and commit V61.
+- [x] Run all V61-focused tests and `tests/core/data_source`.
+- [x] Run the complete Python suite and `git diff --check`.
+- [x] Record evidence in `docs/validation/data-research-v61.md` and commit V61.
 
 ## V62 — Documents, scanned PDFs, and images
 

docs/validation/data-research-v61.md

@@ -0,0 +1,23 @@
+# V61 结构化采集与数据元素生命周期验证
+
+日期：2026-07-22
+
+## 交付范围
+
+- SQL、CSV、Excel 确定性解析器及统一注册表。
+- PostgreSQL、MySQL 元数据目录采集与快照差异计算。
+- 数据元素稳定 UID、版本化生命周期、乐观并发控制和发布 Outbox 事件。
+- 候选项 `reuse/create/map/ignore` 原子决策门禁。
+- 数据元素 HTTP 接口与仅发布版本写入 Neo4j 的幂等投影契约。
+
+## 测试证据
+
+- 结构化解析器与既有 DDL 解析回归：`9 passed`。
+- 数据库目录采集：`4 passed`。
+- 生命周期、候选决策、API、投影、权限与迁移：`19 passed, 1 skipped`。
+- V61 专项及数据源/Outbox/跨存储回归：`93 passed, 3 skipped`。
+- OpenAPI 契约：`7 passed`，共 124 个操作。
+- 全量后端回归：`320 passed, 23 skipped, 59 subtests passed`。
+- `git diff --check`：通过。
+
+跳过项均为需要外部 PostgreSQL、Neo4j 或 MCP 服务的环境型集成测试；单元级跨存储、Outbox 重试与幂等契约已通过。

migrations/versions/20260722_105_data_research_elements.py

@@ -0,0 +1,67 @@
+"""Add versioned data-element governance and candidate decisions."""
+
+from alembic import op
+
+
+revision = "20260722_105"
+down_revision = "20260722_100"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    op.execute(
+        """
+        CREATE TABLE IF NOT EXISTS public.data_elements (
+            uid UUID PRIMARY KEY,
+            code VARCHAR(120) NOT NULL UNIQUE,
+            current_version INTEGER NOT NULL DEFAULT 1,
+            status VARCHAR(20) NOT NULL DEFAULT 'draft'
+                CHECK (status IN ('draft','in_review','published','deprecated','retired')),
+            business_domain_uids JSONB NOT NULL DEFAULT '[]'::jsonb,
+            created_by VARCHAR(100),
+            created_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP,
+            updated_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP
+        );
+
+        CREATE TABLE IF NOT EXISTS public.data_element_versions (
+            uid UUID PRIMARY KEY,
+            data_element_uid UUID NOT NULL
+                REFERENCES public.data_elements(uid),
+            version INTEGER NOT NULL,
+            status VARCHAR(20) NOT NULL,
+            snapshot JSONB NOT NULL,
+            evidence_uids JSONB NOT NULL DEFAULT '[]'::jsonb,
+            created_by VARCHAR(100),
+            created_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP,
+            UNIQUE (data_element_uid, version)
+        );
+
+        CREATE TABLE IF NOT EXISTS public.candidate_decisions (
+            uid UUID PRIMARY KEY,
+            candidate_uid UUID NOT NULL
+                REFERENCES public.extraction_candidates(uid),
+            action VARCHAR(20) NOT NULL
+                CHECK (action IN ('reuse','create','map','ignore')),
+            data_element_uid UUID
+                REFERENCES public.data_elements(uid),
+            evidence_uids JSONB NOT NULL DEFAULT '[]'::jsonb,
+            payload JSONB NOT NULL DEFAULT '{}'::jsonb,
+            actor_uid VARCHAR(100),
+            reason VARCHAR(1000),
+            created_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP,
+            UNIQUE (candidate_uid)
+        );
+
+        CREATE INDEX IF NOT EXISTS idx_data_elements_status
+            ON public.data_elements(status, updated_at);
+        CREATE INDEX IF NOT EXISTS idx_data_element_versions_element
+            ON public.data_element_versions(data_element_uid, version DESC);
+        """
+    )
+
+
+def downgrade() -> None:
+    # Governance history is preserved during application rollback.
+    pass
+

tests/data_research/test_candidate_decisions.py

@@ -0,0 +1,120 @@
+from __future__ import annotations
+
+import pytest
+
+
+class CandidateRepository:
+    def __init__(self):
+        self.candidates = {
+            "candidate-1": {
+                "uid": "candidate-1",
+                "normalized_data": {
+                    "code": "ORDER_ID",
+                    "name_zh": "订单编号",
+                    "name_en": "order_id",
+                    "definition": "订单的唯一业务标识",
+                    "logical_type": "string",
+                },
+                "evidence_uids": ["evidence-1"],
+                "status": "candidate",
+            },
+            "candidate-2": {
+                "uid": "candidate-2",
+                "normalized_data": {"name_zh": "金额"},
+                "evidence_uids": ["evidence-2"],
+                "status": "candidate",
+            },
+        }
+        self.decisions = []
+        self.commits = 0
+        self.rollbacks = 0
+
+    def get(self, uid):
+        return self.candidates.get(uid)
+
+    def save_decisions(self, decisions):
+        self.decisions.extend(decisions)
+
+    def commit(self):
+        self.commits += 1
+
+    def rollback(self):
+        self.rollbacks += 1
+
+
+class ElementService:
+    def __init__(self):
+        self.created = []
+
+    def create_draft(self, payload, actor_uid):
+        self.created.append((payload, actor_uid))
+        return type("Element", (), {"uid": "element-new", "status": "draft"})()
+
+
+def test_batch_decisions_are_atomic_and_retain_evidence():
+    from app.core.data_research.candidate_decisions import CandidateDecisionService
+
+    repository = CandidateRepository()
+    elements = ElementService()
+    service = CandidateDecisionService(
+        repository,
+        data_elements=elements,
+        uid_factory=iter(("decision-1", "decision-2")).__next__,
+    )
+    decisions = service.decide(
+        [
+            {"candidate_uid": "candidate-1", "action": "create"},
+            {
+                "candidate_uid": "candidate-2",
+                "action": "map",
+                "data_element_uid": "element-existing",
+            },
+        ],
+        actor_uid="editor-1",
+    )
+
+    assert repository.commits == 1
+    assert repository.rollbacks == 0
+    assert decisions[0].data_element_uid == "element-new"
+    assert decisions[0].evidence_uids == ("evidence-1",)
+    assert elements.created[0][0]["evidence_uids"] == ["evidence-1"]
+    assert elements.created[0][0].get("status") is None
+    assert decisions[1].data_element_uid == "element-existing"
+
+
+def test_invalid_batch_rolls_back_without_partial_decisions():
+    from app.core.data_research.candidate_decisions import CandidateDecisionService
+    from app.core.data_research.errors import CandidateDecisionInvalid
+
+    repository = CandidateRepository()
+    service = CandidateDecisionService(
+        repository,
+        data_elements=ElementService(),
+    )
+
+    with pytest.raises(CandidateDecisionInvalid, match="data_element_uid"):
+        service.decide(
+            [
+                {"candidate_uid": "candidate-1", "action": "ignore"},
+                {"candidate_uid": "candidate-2", "action": "reuse"},
+            ],
+            actor_uid="editor-1",
+        )
+
+    assert repository.decisions == []
+    assert repository.commits == 0
+    assert repository.rollbacks == 1
+
+
+def test_candidate_decision_never_directly_publishes():
+    from app.core.data_research.candidate_decisions import CandidateDecisionService
+    from app.core.data_research.errors import CandidateDecisionInvalid
+
+    repository = CandidateRepository()
+    service = CandidateDecisionService(repository, data_elements=ElementService())
+
+    with pytest.raises(CandidateDecisionInvalid, match="unsupported action"):
+        service.decide(
+            [{"candidate_uid": "candidate-1", "action": "publish"}],
+            actor_uid="editor-1",
+        )

tests/data_research/test_catalog_collectors.py

@@ -0,0 +1,180 @@
+from __future__ import annotations
+
+from contextlib import contextmanager
+
+
+class FakeResult:
+    def __init__(self, rows):
+        self.rows = rows
+
+    def mappings(self):
+        return self
+
+    def all(self):
+        return list(self.rows)
+
+
+class FakeConnection:
+    def __init__(self, rows):
+        self.rows = rows
+        self.calls = []
+
+    def execute(self, statement, parameters):
+        self.calls.append((str(statement), dict(parameters)))
+        return FakeResult(self.rows)
+
+
+def sample_rows():
+    return [
+        {
+            "schema_name": "public",
+            "asset_name": "orders",
+            "asset_type": "BASE TABLE",
+            "asset_comment": "orders",
+            "column_name": "id",
+            "ordinal_position": 1,
+            "data_type": "integer",
+            "is_nullable": "NO",
+            "column_default": None,
+            "column_comment": "primary key",
+        },
+        {
+            "schema_name": "public",
+            "asset_name": "orders",
+            "asset_type": "BASE TABLE",
+            "asset_comment": "orders",
+            "column_name": "amount",
+            "ordinal_position": 2,
+            "data_type": "numeric",
+            "is_nullable": "YES",
+            "column_default": None,
+            "column_comment": None,
+        },
+    ]
+
+
+def test_postgresql_collector_uses_allowlisted_catalog_and_bound_scope():
+    from app.core.data_research.catalog.models import CatalogScope
+    from app.core.data_research.catalog.postgresql import PostgreSqlCatalogCollector
+
+    connection = FakeConnection(sample_rows())
+    snapshot = PostgreSqlCatalogCollector().collect(
+        connection,
+        data_source_uid="source-1",
+        scope=CatalogScope(include_schemas=("public", "unsafe' OR 1=1 --")),
+    )
+
+    statement, parameters = connection.calls[0]
+    assert "information_schema.columns" in statement
+    assert "pg_catalog" in statement
+    assert "unsafe' OR 1=1 --" not in statement
+    assert parameters["include_schemas"] == ["public", "unsafe' OR 1=1 --"]
+    assert snapshot.assets[0].fields[1].key == "source-1:public.orders.amount"
+    assert snapshot.assets[0].fields[1].nullable is True
+
+
+def test_mysql_collector_uses_information_schema_and_bound_database():
+    from app.core.data_research.catalog.models import CatalogScope
+    from app.core.data_research.catalog.mysql import MySqlCatalogCollector
+
+    connection = FakeConnection(sample_rows())
+    snapshot = MySqlCatalogCollector().collect(
+        connection,
+        data_source_uid="source-2",
+        database_name="analytics' OR 1=1 --",
+        scope=CatalogScope(include_tables=("orders",)),
+    )
+
+    statement, parameters = connection.calls[0]
+    assert "information_schema.COLUMNS" in statement
+    assert "analytics' OR 1=1 --" not in statement
+    assert parameters["database_name"] == "analytics' OR 1=1 --"
+    assert parameters["include_tables"] == ["orders"]
+    assert snapshot.database_type == "mysql"
+
+
+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
+    from app.core.data_research.catalog.service import CatalogCollectionService
+
+    connection = FakeConnection(sample_rows())
+
+    class Manager:
+        def __init__(self):
+            self.calls = []
+
+        @contextmanager
+        def connect(self, uid, purpose):
+            self.calls.append((uid, purpose))
+            yield connection
+
+    manager = Manager()
+    service = CatalogCollectionService(
+        manager,
+        definition_resolver=lambda _uid: {
+            "database_type": "postgresql",
+            "database": "warehouse",
+        },
+        collector_resolver=lambda _database_type: PostgreSqlCatalogCollector(),
+    )
+
+    snapshot = service.collect("source-1", CatalogScope())
+
+    assert manager.calls == [("source-1", "metadata_collection")]
+    assert snapshot.assets[0].name == "orders"
+
+
+def _snapshot(fields):
+    from app.core.data_research.catalog.models import (
+        CatalogAsset,
+        CatalogField,
+        CatalogSnapshot,
+    )
+
+    return CatalogSnapshot(
+        data_source_uid="source-1",
+        database_type="postgresql",
+        assets=(
+            CatalogAsset(
+                key="source-1:public.orders",
+                schema="public",
+                name="orders",
+                asset_type="table",
+                fields=tuple(
+                    CatalogField(
+                        key=f"source-1:public.orders.{name}",
+                        schema="public",
+                        asset="orders",
+                        name=name,
+                        ordinal_position=ordinal,
+                        data_type=data_type,
+                        nullable=True,
+                    )
+                    for name, ordinal, data_type in fields
+                ),
+            ),
+        ),
+    )
+
+
+def test_snapshot_diff_classifies_add_remove_type_change_and_rename():
+    from app.core.data_research.catalog.service import diff_snapshots
+
+    previous = _snapshot(
+        (("id", 1, "integer"), ("amount", 2, "numeric"), ("legacy", 3, "text"))
+    )
+    current = _snapshot(
+        (("id", 1, "bigint"), ("total", 2, "numeric"), ("created_at", 4, "timestamp"))
+    )
+
+    diff = diff_snapshots(previous, current)
+
+    assert [(item.before.name, item.after.name) for item in diff.renamed] == [
+        ("amount", "total")
+    ]
+    assert [(item.before.data_type, item.after.data_type) for item in diff.changed] == [
+        ("integer", "bigint")
+    ]
+    assert [item.name for item in diff.removed] == ["legacy"]
+    assert [item.name for item in diff.added] == ["created_at"]

tests/data_research/test_data_element_api.py

@@ -0,0 +1,146 @@
+from __future__ import annotations
+
+from dataclasses import replace
+
+import pytest
+
+from app.core.data_research.data_elements import DataElementRecord
+
+
+class FakeElementService:
+    def __init__(self):
+        self.record = DataElementRecord(
+            uid="element-1",
+            code="ORDER_ID",
+            status="draft",
+            current_version=1,
+            snapshot={"definition": "订单唯一标识"},
+            created_by="editor-1",
+            updated_by="editor-1",
+        )
+        self.actions = []
+
+    def create_draft(self, payload, actor_uid):
+        self.actions.append(("create", payload, actor_uid))
+        return self.record
+
+    def transition(self, uid, target_status, *, expected_version, actor_uid):
+        self.actions.append(("transition", uid, target_status, expected_version, actor_uid))
+        return replace(
+            self.record,
+            uid=uid,
+            status=target_status,
+            current_version=expected_version + 1,
+            updated_by=actor_uid,
+        )
+
+
+class FakeDecisionService:
+    def __init__(self):
+        self.calls = []
+
+    def decide(self, items, *, actor_uid):
+        self.calls.append((items, actor_uid))
+        return [
+            type(
+                "Decision",
+                (),
+                {
+                    "uid": "decision-1",
+                    "candidate_uid": "candidate-1",
+                    "action": "ignore",
+                    "data_element_uid": None,
+                    "evidence_uids": ("evidence-1",),
+                    "actor_uid": actor_uid,
+                    "reason": "重复字段",
+                },
+            )()
+        ]
+
+
+@pytest.fixture()
+def client(monkeypatch):
+    from flask import request
+
+    from app import create_app
+    from app.api.data_development import routes
+    from app.core.system import permissions
+
+    elements = FakeElementService()
+    decisions = FakeDecisionService()
+
+    def identity():
+        token = request.headers.get("Authorization", "")
+        if token == "Bearer editor":
+            return {"id": "editor-1", "roles": ["editor"]}
+        if token == "Bearer admin":
+            return {"id": "admin-1", "roles": ["admin"]}
+        return None
+
+    monkeypatch.setattr(permissions, "authenticate_request", identity)
+    monkeypatch.setattr(routes, "get_data_element_service", lambda: elements)
+    monkeypatch.setattr(routes, "get_candidate_decision_service", lambda: decisions)
+    app = create_app()
+    app.config.update(TESTING=True)
+    return app.test_client(), elements, decisions
+
+
+def test_editor_creates_draft_and_submits_review(client):
+    http, elements, _decisions = client
+    headers = {"Authorization": "Bearer editor"}
+    created = http.post(
+        "/api/development/v1/data-elements",
+        headers=headers,
+        json={"code": "ORDER_ID", "definition": "订单唯一标识"},
+    )
+    reviewed = http.post(
+        "/api/development/v1/data-elements/element-1/transitions",
+        headers=headers,
+        json={"target_status": "in_review", "expected_version": 1},
+    )
+
+    assert created.status_code == 201
+    assert created.get_json()["data"]["status"] == "draft"
+    assert reviewed.status_code == 200
+    assert reviewed.get_json()["data"]["status"] == "in_review"
+    assert elements.actions[-1] == (
+        "transition",
+        "element-1",
+        "in_review",
+        1,
+        "editor-1",
+    )
+
+
+def test_only_publisher_can_publish(client):
+    http, _elements, _decisions = client
+    payload = {"target_status": "published", "expected_version": 2}
+
+    denied = http.post(
+        "/api/development/v1/data-elements/element-1/transitions",
+        headers={"Authorization": "Bearer editor"},
+        json=payload,
+    )
+    allowed = http.post(
+        "/api/development/v1/data-elements/element-1/transitions",
+        headers={"Authorization": "Bearer admin"},
+        json=payload,
+    )
+
+    assert denied.status_code == 403
+    assert allowed.status_code == 200
+    assert allowed.get_json()["data"]["current_version"] == 3
+
+
+def test_candidate_decisions_are_editor_gated(client):
+    http, _elements, decisions = client
+    response = http.post(
+        "/api/development/v1/candidate-decisions",
+        headers={"Authorization": "Bearer editor"},
+        json={"decisions": [{"candidate_uid": "candidate-1", "action": "ignore"}]},
+    )
+
+    assert response.status_code == 200
+    assert response.get_json()["data"][0]["evidence_uids"] == ["evidence-1"]
+    assert decisions.calls[0][1] == "editor-1"
+

tests/data_research/test_data_element_lifecycle.py

@@ -0,0 +1,113 @@
+from __future__ import annotations
+
+from dataclasses import replace
+
+import pytest
+
+
+class MemoryElementRepository:
+    def __init__(self):
+        self.records = {}
+
+    def get(self, uid):
+        return self.records.get(uid)
+
+    def get_by_code(self, code):
+        return next((item for item in self.records.values() if item.code == code), None)
+
+    def save(self, record):
+        self.records[record.uid] = record
+        return record
+
+
+def valid_payload(**overrides):
+    payload = {
+        "code": "ORDER_ID",
+        "name_zh": "订单编号",
+        "name_en": "order_id",
+        "definition": "订单的唯一业务标识",
+        "logical_type": "string",
+        "evidence_uids": ["evidence-1"],
+        "business_domain_uids": ["domain-1"],
+    }
+    payload.update(overrides)
+    return payload
+
+
+def test_data_element_requires_definition_and_starts_as_draft():
+    from app.core.data_research.data_elements import DataElementService
+    from app.core.data_research.errors import DataElementInvalid
+
+    repository = MemoryElementRepository()
+    service = DataElementService(repository, uid_factory=lambda: "element-1")
+
+    with pytest.raises(DataElementInvalid, match="definition"):
+        service.create_draft(valid_payload(definition=""), actor_uid="editor-1")
+
+    element = service.create_draft(valid_payload(), actor_uid="editor-1")
+    assert element.uid == "element-1"
+    assert element.status == "draft"
+    assert element.current_version == 1
+    assert element.snapshot["evidence_uids"] == ["evidence-1"]
+
+
+def test_lifecycle_uses_optimistic_version_and_emits_publish_event():
+    from app.core.data_research.data_elements import DataElementService
+    from app.core.data_research.errors import DataElementVersionConflict
+
+    repository = MemoryElementRepository()
+    events = []
+    service = DataElementService(
+        repository,
+        uid_factory=lambda: "element-1",
+        outbox_enqueue=lambda **event: events.append(event),
+    )
+    element = service.create_draft(valid_payload(), actor_uid="editor-1")
+    review = service.transition(
+        element.uid,
+        "in_review",
+        expected_version=1,
+        actor_uid="editor-1",
+    )
+    published = service.transition(
+        element.uid,
+        "published",
+        expected_version=2,
+        actor_uid="admin-1",
+    )
+
+    assert review.current_version == 2
+    assert published.current_version == 3
+    assert events == [
+        {
+            "aggregate_type": "data_element",
+            "aggregate_id": "element-1",
+            "event_type": "data_element.version_published",
+            "payload": {"uid": "element-1", "version": 3},
+        }
+    ]
+    with pytest.raises(DataElementVersionConflict):
+        service.transition(
+            element.uid,
+            "deprecated",
+            expected_version=2,
+            actor_uid="admin-1",
+        )
+
+
+def test_invalid_lifecycle_transition_is_rejected():
+    from app.core.data_research.data_elements import DataElementService
+    from app.core.data_research.errors import DataElementTransitionInvalid
+
+    repository = MemoryElementRepository()
+    service = DataElementService(repository, uid_factory=lambda: "element-1")
+    element = service.create_draft(valid_payload(), actor_uid="editor-1")
+
+    with pytest.raises(DataElementTransitionInvalid, match="draft -> published"):
+        service.transition(
+            element.uid,
+            "published",
+            expected_version=1,
+            actor_uid="admin-1",
+        )
+

tests/data_research/test_graph_projection.py

@@ -0,0 +1,45 @@
+from __future__ import annotations
+
+import pytest
+
+
+class FakeGraph:
+    def __init__(self):
+        self.calls = []
+
+    def run(self, statement, **parameters):
+        self.calls.append((statement, parameters))
+
+
+def test_published_element_projection_is_uid_based_and_idempotent():
+    from app.core.data_research.graph_projection import DataElementGraphProjector
+
+    graph = FakeGraph()
+    projector = DataElementGraphProjector(graph)
+    payload = {
+        "uid": "element-1",
+        "version": 3,
+        "status": "published",
+        "snapshot": {"code": "ORDER_ID", "name_zh": "订单编号"},
+    }
+
+    projector.project(payload)
+    projector.project(payload)
+
+    assert len(graph.calls) == 2
+    assert "MERGE (element:DataElement {uid: $uid})" in graph.calls[0][0]
+    assert graph.calls[0][1]["uid"] == "element-1"
+    assert graph.calls[0][1]["version"] == 3
+
+
+def test_unpublished_element_is_never_projected():
+    from app.core.data_research.graph_projection import DataElementGraphProjector
+    from app.core.data_research.errors import DataElementProjectionInvalid
+
+    graph = FakeGraph()
+    with pytest.raises(DataElementProjectionInvalid, match="published"):
+        DataElementGraphProjector(graph).project(
+            {"uid": "element-1", "version": 2, "status": "in_review", "snapshot": {}}
+        )
+    assert graph.calls == []
+

tests/data_research/test_structured_extractors.py

@@ -0,0 +1,119 @@
+from __future__ import annotations
+
+import io
+
+import pandas as pd
+import pytest
+
+
+def test_sql_extractor_returns_multi_table_candidates_with_statement_evidence():
+    from app.core.data_research.extractors.base import ExtractionContext
+    from app.core.data_research.extractors.sql import SqlExtractor
+
+    content = b"""
+    CREATE TABLE customer (id INTEGER PRIMARY KEY, name VARCHAR(50));
+    CREATE TABLE orders (id INTEGER PRIMARY KEY, customer_id INTEGER NOT NULL);
+    """
+    batch = SqlExtractor().extract(
+        content,
+        ExtractionContext(filename="schema.sql", media_type="application/sql"),
+    )
+
+    assert batch.parser_version == "sql-ddl-v1"
+    assert len(batch.content_hash) == 64
+    assert [item.data["table_info"]["name_en"] for item in batch.items] == [
+        "customer",
+        "orders",
+    ]
+    assert batch.items[0].evidence[0].locator == {
+        "kind": "sql.statement",
+        "statement": 1,
+    }
+
+
+def test_csv_extractor_detects_bom_delimiter_and_column_types():
+    from app.core.data_research.extractors.base import ExtractionContext
+    from app.core.data_research.extractors.csv import CsvExtractor
+
+    content = "\ufeffid;amount;active\n1;12.5;true\n2;9.0;false\n".encode("utf-8")
+    batch = CsvExtractor().extract(
+        content,
+        ExtractionContext(filename="orders.csv", media_type="text/csv"),
+    )
+
+    table = batch.items[0]
+    assert table.data["table_info"]["name_en"] == "orders"
+    assert [(column["name_en"], column["data_type"]) for column in table.data["columns"]] == [
+        ("id", "integer"),
+        ("amount", "decimal"),
+        ("active", "boolean"),
+    ]
+    assert table.data["columns"][1]["evidence_locator"] == {
+        "kind": "csv.header",
+        "column": 2,
+        "name": "amount",
+    }
+
+
+def test_csv_header_only_is_valid_and_empty_file_is_rejected():
+    from app.core.data_research.extractors.base import ExtractionContext
+    from app.core.data_research.extractors.csv import CsvExtractor
+
+    context = ExtractionContext(filename="empty.csv", media_type="text/csv")
+    batch = CsvExtractor().extract(b"id,name\n", context)
+    assert [column["data_type"] for column in batch.items[0].data["columns"]] == [
+        "string",
+        "string",
+    ]
+
+    with pytest.raises(ValueError, match="header"):
+        CsvExtractor().extract(b"", context)
+
+
+def test_excel_extractor_returns_one_candidate_per_sheet_with_cell_evidence():
+    from app.core.data_research.extractors.base import ExtractionContext
+    from app.core.data_research.extractors.excel import ExcelExtractor
+
+    stream = io.BytesIO()
+    with pd.ExcelWriter(stream, engine="openpyxl") as writer:
+        pd.DataFrame({"customer_id": [1], "name": ["Ada"]}).to_excel(
+            writer, sheet_name="Customers", index=False
+        )
+        pd.DataFrame({"order_id": [9], "total": [3.5]}).to_excel(
+            writer, sheet_name="Orders", index=False
+        )
+
+    batch = ExcelExtractor().extract(
+        stream.getvalue(),
+        ExtractionContext(
+            filename="dictionary.xlsx",
+            media_type="application/vnd.openxmlformats-officedocument.spreadsheetml.sheet",
+        ),
+    )
+
+    assert [item.data["table_info"]["name_en"] for item in batch.items] == [
+        "Customers",
+        "Orders",
+    ]
+    assert batch.items[1].data["columns"][1]["evidence_locator"] == {
+        "kind": "excel.cell_range",
+        "sheet": "Orders",
+        "range": "B1:B2",
+    }
+
+
+def test_registry_selects_by_filename_and_enforces_input_limit():
+    from app.core.data_research.extractors.base import ExtractionContext
+    from app.core.data_research.extractors.registry import default_registry
+
+    registry = default_registry()
+    assert registry.resolve("text/csv", "orders.csv").parser_version == "csv-v1"
+    assert registry.resolve("application/sql", "schema.sql").parser_version == "sql-ddl-v1"
+
+    context = ExtractionContext(
+        filename="orders.csv",
+        media_type="text/csv",
+        max_bytes=4,
+    )
+    with pytest.raises(ValueError, match="size limit"):
+        registry.extract(b"id,name\n1,Ada\n", context)

tests/test_database_migrations.py

@@ -40,6 +40,9 @@ EXPECTED_UPGRADED_TABLES = {
     "ingestion_jobs",
     "evidence_fragments",
     "extraction_candidates",
+    "data_elements",
+    "data_element_versions",
+    "candidate_decisions",
 }
 
 
@@ -67,6 +70,22 @@ def test_data_research_ingestion_migration_is_additive_and_constrained():
     assert "DROP TABLE" not in migration.upper()
 
 
+def test_data_element_migration_adds_versioned_governance_tables():
+    migration = (
+        ROOT
+        / "migrations"
+        / "versions"
+        / "20260722_105_data_research_elements.py"
+    ).read_text(encoding="utf-8")
+
+    assert 'revision = "20260722_105"' in migration
+    assert 'down_revision = "20260722_100"' in migration
+    for table in ("data_elements", "data_element_versions", "candidate_decisions"):
+        assert f"CREATE TABLE IF NOT EXISTS public.{table}" in migration
+    assert "UNIQUE (data_element_uid, version)" in migration
+    assert "DROP TABLE" not in migration.upper()
+
+
 def test_alembic_configuration_is_environment_only():
     ini = (ROOT / "alembic.ini").read_text(encoding="utf-8")
     env = (ROOT / "migrations" / "env.py").read_text(encoding="utf-8")

tests/test_permission_matrix.py

@@ -31,6 +31,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/data-elements", "POST"
+    ) == ("data-elements:edit",)
+    assert permission_for_request(
+        "/api/development/v1/data-elements/element-1/transitions", "POST"
+    ) == ("data-elements:edit",)
+    assert permission_for_request(
+        "/api/development/v1/candidate-decisions", "POST"
+    ) == ("data-elements:edit",)
 
 
 def test_permissions_are_derived_server_side_from_roles():