feat: add V60 data research ingestion foundation

app/__init__.py

@@ -37,6 +37,7 @@ def create_app():
     from app.api.data_interface import bp as data_interface_bp
     from app.api.data_service import bp as data_service_bp
     from app.api.data_source import bp as data_source_bp
+    from app.api.data_development import bp as data_development_bp
     from app.api.graph import bp as graph_bp
     from app.api.meta_data import bp as meta_bp
     from app.api.system import bp as system_bp
@@ -46,6 +47,7 @@ def create_app():
     app.register_blueprint(graph_bp, url_prefix="/api/graph")
     app.register_blueprint(system_bp, url_prefix="/api/system")
     app.register_blueprint(data_source_bp, url_prefix="/api/datasource")
+    app.register_blueprint(data_development_bp, url_prefix="/api/development/v1")
     app.register_blueprint(data_flow_bp, url_prefix="/api/dataflow")
     app.register_blueprint(business_domain_bp, url_prefix="/api/bd")
     app.register_blueprint(data_factory_bp, url_prefix="/api/datafactory")

app/api/data_development/__init__.py

@@ -0,0 +1,7 @@
+from flask import Blueprint
+
+
+bp = Blueprint("data_development", __name__)
+
+from app.api.data_development import routes  # noqa: E402, F401
+

app/api/data_development/routes.py

@@ -0,0 +1,135 @@
+"""HTTP orchestration boundary for data-research ingestion jobs."""
+
+from __future__ import annotations
+
+import logging
+
+from flask import g, jsonify, request
+
+from app import db
+from app.api.data_development import bp
+from app.core.data_research.errors import DataResearchError
+from app.models.result import failed, success
+
+
+logger = logging.getLogger(__name__)
+
+
+def get_ingestion_service():
+    from app.core.data_research.ingestion import IngestionService
+    from app.core.data_research.repository import SqlAlchemyIngestionJobRepository
+
+    return IngestionService(
+        SqlAlchemyIngestionJobRepository(db.session),
+        commit=db.session.commit,
+        rollback=db.session.rollback,
+    )
+
+
+def _identity():
+    return getattr(g, "current_user", {}) or {}
+
+
+def _record(record):
+    return {
+        "uid": str(record.uid),
+        "source_uid": str(record.source_uid),
+        "artifact_uid": str(record.artifact_uid) if record.artifact_uid else None,
+        "job_type": record.job_type,
+        "parser_version": record.parser_version,
+        "status": record.status,
+        "parameters": dict(record.parameters or {}),
+        "statistics": dict(record.statistics or {}),
+        "last_error": record.last_error,
+        "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,
+        "started_at": record.started_at.isoformat() if record.started_at else None,
+        "finished_at": record.finished_at.isoformat() if record.finished_at else None,
+    }
+
+
+def _error(error):
+    if isinstance(error, DataResearchError):
+        return (
+            jsonify(
+                failed(
+                    str(error),
+                    code=error.http_status,
+                    error={"code": error.code},
+                )
+            ),
+            error.http_status,
+        )
+    logger.exception("data-research ingestion request failed")
+    return (
+        jsonify(
+            failed(
+                "数据采集任务处理失败",
+                code=500,
+                error={"code": "DATA_RESEARCH_ERROR"},
+            )
+        ),
+        500,
+    )
+
+
+@bp.route("/ingestion-jobs", methods=["POST"])
+def create_ingestion_job():
+    payload = request.get_json(silent=True) or {}
+    try:
+        record, created = get_ingestion_service().create_job(
+            payload,
+            actor_uid=_identity().get("id") or _identity().get("sub"),
+        )
+        return jsonify(success(_record(record))), 201 if created else 200
+    except Exception as error:
+        return _error(error)
+
+
+@bp.route("/ingestion-jobs", methods=["GET"])
+def list_ingestion_jobs():
+    filters = {
+        name: request.args.get(name)
+        for name in ("status", "source_uid")
+        if request.args.get(name)
+    }
+    try:
+        records = get_ingestion_service().list_jobs(filters)
+        return jsonify(
+            success({"records": [_record(item) for item in records], "total": len(records)})
+        ), 200
+    except Exception as error:
+        return _error(error)
+
+
+@bp.route("/ingestion-jobs/<job_uid>", methods=["GET"])
+def get_ingestion_job(job_uid):
+    try:
+        return jsonify(success(_record(get_ingestion_service().get_job(job_uid)))), 200
+    except Exception as error:
+        return _error(error)
+
+
+@bp.route("/ingestion-jobs/<job_uid>/retry", methods=["POST"])
+def retry_ingestion_job(job_uid):
+    try:
+        return jsonify(success(_record(get_ingestion_service().retry(job_uid)))), 200
+    except Exception as error:
+        return _error(error)
+
+
+@bp.route("/ingestion-jobs/<job_uid>/cancel", methods=["POST"])
+def cancel_ingestion_job(job_uid):
+    try:
+        service = get_ingestion_service()
+        record = service.get_job(job_uid)
+        identity = _identity()
+        permissions = set(identity.get("permissions") or [])
+        actor_uid = identity.get("id") or identity.get("sub")
+        if record.actor_uid != actor_uid and "ingestion:admin" not in permissions:
+            return jsonify(failed("权限不足", code=403)), 403
+        return jsonify(success(_record(service.cancel(job_uid)))), 200
+    except Exception as error:
+        return _error(error)
+

app/core/data_research/__init__.py

@@ -0,0 +1,6 @@
+"""Data-research ingestion, governance, and ontology services."""
+
+from app.core.data_research.ingestion import IngestionService
+
+__all__ = ["IngestionService"]
+

app/core/data_research/errors.py

@@ -0,0 +1,18 @@
+class DataResearchError(ValueError):
+    code = "DATA_RESEARCH_ERROR"
+    http_status = 400
+
+
+class IngestionPayloadInvalid(DataResearchError):
+    code = "INGESTION_PAYLOAD_INVALID"
+
+
+class InvalidJobTransition(DataResearchError):
+    code = "INGESTION_TRANSITION_INVALID"
+    http_status = 409
+
+
+class IngestionJobNotFound(DataResearchError):
+    code = "INGESTION_JOB_NOT_FOUND"
+    http_status = 404
+

app/core/data_research/ingestion.py

@@ -0,0 +1,182 @@
+from __future__ import annotations
+
+import hashlib
+import json
+import re
+from dataclasses import replace
+from typing import Any, Callable
+
+from app.core.common.identifiers import new_governance_uid
+from app.core.common.timezone_utils import now_china_naive
+from app.core.data_research.errors import (
+    IngestionJobNotFound,
+    IngestionPayloadInvalid,
+    InvalidJobTransition,
+)
+from app.core.data_research.models import IngestionJobRecord, IngestionJobSpec
+
+
+ALLOWED_TRANSITIONS = {
+    "created": frozenset({"queued", "cancelled"}),
+    "queued": frozenset({"extracting", "failed", "cancelled"}),
+    "extracting": frozenset({"normalizing", "failed", "cancelled"}),
+    "normalizing": frozenset({"matching", "failed", "cancelled"}),
+    "matching": frozenset(
+        {"awaiting_review", "partial", "failed", "cancelled"}
+    ),
+    "awaiting_review": frozenset({"published", "partial", "cancelled"}),
+    "partial": frozenset({"queued", "cancelled"}),
+    "failed": frozenset({"queued", "cancelled"}),
+    "published": frozenset(),
+    "cancelled": frozenset(),
+}
+TERMINAL_STATUSES = frozenset({"published", "cancelled"})
+_SECRET_ASSIGNMENT = re.compile(
+    r"(?i)\b(password|token|secret|authorization)\s*[:=]\s*[^\s,;]+"
+)
+
+
+def _required_text(payload: dict[str, Any], name: str) -> str:
+    value = str(payload.get(name) or "").strip()
+    if not value:
+        raise IngestionPayloadInvalid(f"{name} is required")
+    return value
+
+
+def _sanitize_error(error: Any) -> str | None:
+    if error is None:
+        return None
+    value = _SECRET_ASSIGNMENT.sub(lambda match: f"{match.group(1)}=[redacted]", str(error))
+    return value[:1000]
+
+
+class IngestionService:
+    def __init__(
+        self,
+        repository,
+        *,
+        uid_factory: Callable[[], str] = new_governance_uid,
+        nonce_factory: Callable[[], str] = new_governance_uid,
+        clock: Callable[[], Any] = now_china_naive,
+        commit: Callable[[], Any] = lambda: None,
+        rollback: Callable[[], Any] = lambda: None,
+    ):
+        self.repository = repository
+        self.uid_factory = uid_factory
+        self.nonce_factory = nonce_factory
+        self.clock = clock
+        self.commit = commit
+        self.rollback = rollback
+
+    @staticmethod
+    def _spec(payload: dict[str, Any]) -> IngestionJobSpec:
+        if not isinstance(payload, dict):
+            raise IngestionPayloadInvalid("payload must be an object")
+        parameters = payload.get("parameters", {})
+        if parameters is None:
+            parameters = {}
+        if not isinstance(parameters, dict):
+            raise IngestionPayloadInvalid("parameters must be an object")
+        artifact_uid = str(payload.get("artifact_uid") or "").strip() or None
+        return IngestionJobSpec(
+            source_uid=_required_text(payload, "source_uid"),
+            artifact_uid=artifact_uid,
+            job_type=_required_text(payload, "job_type"),
+            parser_version=_required_text(payload, "parser_version"),
+            parameters=dict(parameters),
+            force_rerun=bool(payload.get("force_rerun", False)),
+        )
+
+    def _idempotency_key(self, spec: IngestionJobSpec) -> str:
+        value = {
+            "source_uid": spec.source_uid,
+            "artifact_uid": spec.artifact_uid,
+            "job_type": spec.job_type,
+            "parser_version": spec.parser_version,
+            "parameters": spec.parameters,
+        }
+        if spec.force_rerun:
+            value["rerun_nonce"] = self.nonce_factory()
+        canonical = json.dumps(
+            value,
+            sort_keys=True,
+            ensure_ascii=False,
+            separators=(",", ":"),
+        )
+        return hashlib.sha256(canonical.encode("utf-8")).hexdigest()
+
+    def create_job(self, payload: dict[str, Any], actor_uid: str | None):
+        spec = self._spec(payload)
+        idempotency_key = self._idempotency_key(spec)
+        if not spec.force_rerun:
+            existing = self.repository.get_by_idempotency_key(idempotency_key)
+            if existing is not None:
+                return existing, False
+        now = self.clock()
+        record = IngestionJobRecord(
+            uid=self.uid_factory(),
+            source_uid=spec.source_uid,
+            artifact_uid=spec.artifact_uid,
+            job_type=spec.job_type,
+            parser_version=spec.parser_version,
+            idempotency_key=idempotency_key,
+            actor_uid=actor_uid,
+            parameters=spec.parameters,
+            force_rerun=spec.force_rerun,
+            created_at=now,
+            updated_at=now,
+        )
+        try:
+            saved = self.repository.add(record)
+            self.commit()
+            return saved, True
+        except Exception:
+            self.rollback()
+            raise
+
+    def _get(self, uid: str) -> IngestionJobRecord:
+        record = self.repository.get(str(uid))
+        if record is None:
+            raise IngestionJobNotFound(f"ingestion job {uid} was not found")
+        return record
+
+    def transition(
+        self,
+        uid: str,
+        target_status: str,
+        statistics: dict[str, Any] | None = None,
+        error: Any = None,
+    ) -> IngestionJobRecord:
+        record = self._get(uid)
+        target_status = str(target_status or "").strip()
+        if target_status not in ALLOWED_TRANSITIONS.get(record.status, frozenset()):
+            raise InvalidJobTransition(f"{record.status} -> {target_status} is not allowed")
+        now = self.clock()
+        updated = replace(
+            record,
+            status=target_status,
+            statistics=dict(statistics or record.statistics),
+            last_error=_sanitize_error(error),
+            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),
+        )
+        try:
+            saved = self.repository.save(updated)
+            self.commit()
+            return saved
+        except Exception:
+            self.rollback()
+            raise
+
+    def get_job(self, uid: str) -> IngestionJobRecord:
+        return self._get(uid)
+
+    def list_jobs(self, filters: dict[str, Any] | None = None):
+        return self.repository.list(filters or {})
+
+    def retry(self, uid: str) -> IngestionJobRecord:
+        return self.transition(uid, "queued")
+
+    def cancel(self, uid: str) -> IngestionJobRecord:
+        return self.transition(uid, "cancelled")

app/core/data_research/models.py

@@ -0,0 +1,42 @@
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+from datetime import datetime
+from typing import Any
+
+
+@dataclass(frozen=True)
+class IngestionJobSpec:
+    source_uid: str
+    job_type: str
+    parser_version: str
+    artifact_uid: str | None = None
+    parameters: dict[str, Any] = field(default_factory=dict)
+    force_rerun: bool = False
+
+
+@dataclass(frozen=True)
+class IngestionJobRecord:
+    uid: str
+    source_uid: str
+    job_type: str
+    parser_version: str
+    idempotency_key: str
+    actor_uid: str | None
+    artifact_uid: str | None = None
+    parameters: dict[str, Any] = field(default_factory=dict)
+    status: str = "created"
+    statistics: dict[str, Any] = field(default_factory=dict)
+    last_error: str | None = None
+    force_rerun: bool = False
+    created_at: datetime | None = None
+    updated_at: datetime | None = None
+    started_at: datetime | None = None
+    finished_at: datetime | None = None
+
+
+@dataclass(frozen=True)
+class JobTransition:
+    source_status: str
+    target_status: str
+

app/core/data_research/repository.py

@@ -0,0 +1,94 @@
+from __future__ import annotations
+
+from dataclasses import replace
+
+from app.models.data_research import IngestionJob
+
+
+class SqlAlchemyIngestionJobRepository:
+    def __init__(self, session):
+        self.session = session
+
+    @staticmethod
+    def _record(model):
+        from app.core.data_research.models import IngestionJobRecord
+
+        return IngestionJobRecord(
+            uid=str(model.uid),
+            source_uid=str(model.source_uid),
+            artifact_uid=str(model.artifact_uid) if model.artifact_uid else None,
+            job_type=model.job_type,
+            parser_version=model.parser_version,
+            idempotency_key=model.idempotency_key,
+            actor_uid=model.actor_uid,
+            parameters=dict(model.parameters or {}),
+            status=model.status,
+            statistics=dict(model.statistics or {}),
+            last_error=model.last_error,
+            force_rerun=bool(model.force_rerun),
+            created_at=model.created_at,
+            updated_at=model.updated_at,
+            started_at=model.started_at,
+            finished_at=model.finished_at,
+        )
+
+    def get(self, uid):
+        model = self.session.get(IngestionJob, str(uid))
+        return self._record(model) if model is not None else None
+
+    def get_by_idempotency_key(self, key):
+        model = self.session.query(IngestionJob).filter_by(
+            idempotency_key=str(key)
+        ).first()
+        return self._record(model) if model is not None else None
+
+    def list(self, filters=None):
+        filters = filters or {}
+        query = self.session.query(IngestionJob)
+        if filters.get("status"):
+            query = query.filter(IngestionJob.status == str(filters["status"]))
+        if filters.get("source_uid"):
+            query = query.filter(
+                IngestionJob.source_uid == str(filters["source_uid"])
+            )
+        return [
+            self._record(model)
+            for model in query.order_by(IngestionJob.created_at.desc()).all()
+        ]
+
+    def add(self, record):
+        model = IngestionJob(
+            uid=record.uid,
+            source_uid=record.source_uid,
+            artifact_uid=record.artifact_uid,
+            job_type=record.job_type,
+            status=record.status,
+            idempotency_key=record.idempotency_key,
+            parser_version=record.parser_version,
+            parameters=record.parameters,
+            statistics=record.statistics,
+            last_error=record.last_error,
+            actor_uid=record.actor_uid,
+            force_rerun=record.force_rerun,
+            started_at=record.started_at,
+            finished_at=record.finished_at,
+        )
+        self.session.add(model)
+        self.session.flush()
+        return self._record(model)
+
+    def save(self, record):
+        model = self.session.get(IngestionJob, str(record.uid))
+        if model is None:
+            return self.add(record)
+        for name in (
+            "status",
+            "statistics",
+            "last_error",
+            "started_at",
+            "finished_at",
+            "updated_at",
+        ):
+            setattr(model, name, getattr(record, name))
+        self.session.flush()
+        return self._record(model)

app/core/system/permissions.py

@@ -15,11 +15,26 @@ MANAGE_USERS = "users:manage"
 ACTIVATE_WORKFLOW = "workflow:activate"
 OPERATE_ORDERS = "orders:operate"
 DATASOURCE_POOL_MANAGE = "datasources:pools:manage"
+INGESTION_RUN = "ingestion:run"
+INGESTION_ADMIN = "ingestion:admin"
+EVIDENCE_DOWNLOAD = "evidence:download"
+DATA_ELEMENTS_EDIT = "data-elements:edit"
+DATA_ELEMENTS_PUBLISH = "data-elements:publish"
+ONTOLOGIES_EDIT = "ontologies:edit"
+ONTOLOGIES_PUBLISH = "ontologies:publish"
 
 ROLE_PERMISSIONS = {
     "viewer": frozenset({READ_GOVERNANCE}),
     "editor": frozenset(
-        {READ_GOVERNANCE, EDIT_GOVERNANCE, APPROVE_REVIEW, OPERATE_ORDERS}
+        {
+            READ_GOVERNANCE,
+            EDIT_GOVERNANCE,
+            APPROVE_REVIEW,
+            OPERATE_ORDERS,
+            INGESTION_RUN,
+            DATA_ELEMENTS_EDIT,
+            ONTOLOGIES_EDIT,
+        }
     ),
     "admin": frozenset(
         {
@@ -30,6 +45,13 @@ ROLE_PERMISSIONS = {
             ACTIVATE_WORKFLOW,
             OPERATE_ORDERS,
             DATASOURCE_POOL_MANAGE,
+            INGESTION_RUN,
+            INGESTION_ADMIN,
+            EVIDENCE_DOWNLOAD,
+            DATA_ELEMENTS_EDIT,
+            DATA_ELEMENTS_PUBLISH,
+            ONTOLOGIES_EDIT,
+            ONTOLOGIES_PUBLISH,
         }
     ),
 }
@@ -44,6 +66,15 @@ def permission_for_request(path: str, method: str) -> tuple[str, ...]:
         return (PUBLIC,)
     if path.startswith("/api/system/users"):
         return (MANAGE_USERS,)
+    if path.startswith("/api/development/v1/ingestion-jobs"):
+        if method == "GET":
+            return (READ_GOVERNANCE,)
+        if path.endswith("/retry"):
+            return (INGESTION_ADMIN,)
+        if path.endswith("/cancel"):
+            return (INGESTION_RUN,)
+        if method == "POST":
+            return (INGESTION_RUN,)
     if path.startswith("/api/system/workbench"):
         return (READ_GOVERNANCE,)
     if (

app/models/__init__.py

@@ -2,10 +2,22 @@
 
 from app.models.data_product import DataOrder, DataProduct
 from app.models.metadata_review import MetadataReviewRecord, MetadataVersionHistory
+from app.models.data_research import (
+    EvidenceFragment,
+    ExtractionCandidate,
+    IngestionJob,
+    IngestionSource,
+    SourceArtifact,
+)
 
 __all__ = [
     "DataOrder",
     "DataProduct",
     "MetadataReviewRecord",
     "MetadataVersionHistory",
+    "IngestionSource",
+    "IngestionJob",
+    "SourceArtifact",
+    "EvidenceFragment",
+    "ExtractionCandidate",
 ]

app/models/data_research.py

@@ -0,0 +1,241 @@
+from __future__ import annotations
+
+from typing import Any
+
+from sqlalchemy.dialects.postgresql import JSONB, UUID
+
+from app import db
+from app.core.common.identifiers import new_governance_uid
+from app.core.common.timezone_utils import now_china_naive
+
+
+JOB_STATUSES = (
+    "created",
+    "queued",
+    "extracting",
+    "normalizing",
+    "matching",
+    "awaiting_review",
+    "published",
+    "partial",
+    "failed",
+    "cancelled",
+)
+
+
+def _iso(value: Any) -> str | None:
+    return value.isoformat() if value is not None else None
+
+
+class IngestionSource(db.Model):
+    __tablename__ = "ingestion_sources"
+    __table_args__ = (
+        db.CheckConstraint(
+            "source_type IN ('database','file','ddl')",
+            name="ck_ingestion_source_type",
+        ),
+        db.CheckConstraint(
+            "status IN ('active','disabled')",
+            name="ck_ingestion_source_status",
+        ),
+        {"schema": "public"},
+    )
+
+    uid = db.Column(UUID(as_uuid=False), primary_key=True, default=new_governance_uid)
+    source_type = db.Column(db.String(20), nullable=False)
+    name = db.Column(db.String(300), nullable=False)
+    config = db.Column(JSONB, nullable=False, default=dict)
+    permission_scope = db.Column(JSONB, nullable=False, default=dict)
+    status = db.Column(db.String(20), nullable=False, default="active")
+    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)
+
+    def to_dict(self) -> dict[str, Any]:
+        return {
+            "uid": str(self.uid),
+            "source_type": self.source_type,
+            "name": self.name,
+            "permission_scope": dict(self.permission_scope or {}),
+            "status": self.status,
+            "created_by": self.created_by,
+            "created_at": _iso(self.created_at),
+            "updated_at": _iso(self.updated_at),
+        }
+
+
+class SourceArtifact(db.Model):
+    __tablename__ = "source_artifacts"
+    __table_args__ = (
+        db.CheckConstraint("size_bytes >= 0", name="ck_source_artifact_size"),
+        db.UniqueConstraint(
+            "source_uid",
+            "content_hash",
+            "parser_version",
+            name="uq_source_artifact_parser_content",
+        ),
+        {"schema": "public"},
+    )
+
+    uid = db.Column(UUID(as_uuid=False), primary_key=True, default=new_governance_uid)
+    source_uid = db.Column(
+        UUID(as_uuid=False),
+        db.ForeignKey("public.ingestion_sources.uid"),
+        nullable=False,
+    )
+    filename = db.Column(db.String(500), nullable=False)
+    media_type = db.Column(db.String(200), nullable=False)
+    size_bytes = db.Column(db.BigInteger, nullable=False)
+    content_hash = db.Column(db.String(64), nullable=False)
+    storage_ref = db.Column(db.Text, nullable=False)
+    parser_version = db.Column(db.String(100), 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),
+            "source_uid": str(self.source_uid),
+            "filename": self.filename,
+            "media_type": self.media_type,
+            "size_bytes": int(self.size_bytes),
+            "content_hash": self.content_hash,
+            "parser_version": self.parser_version,
+            "created_at": _iso(self.created_at),
+        }
+
+
+class IngestionJob(db.Model):
+    __tablename__ = "ingestion_jobs"
+    __table_args__ = (
+        db.CheckConstraint(
+            "status IN (" + ",".join(f"'{value}'" for value in JOB_STATUSES) + ")",
+            name="ck_ingestion_job_status",
+        ),
+        db.UniqueConstraint(
+            "idempotency_key",
+            name="uq_ingestion_job_idempotency_key",
+        ),
+        {"schema": "public"},
+    )
+
+    uid = db.Column(UUID(as_uuid=False), primary_key=True, default=new_governance_uid)
+    source_uid = db.Column(
+        UUID(as_uuid=False),
+        db.ForeignKey("public.ingestion_sources.uid"),
+        nullable=False,
+    )
+    artifact_uid = db.Column(
+        UUID(as_uuid=False),
+        db.ForeignKey("public.source_artifacts.uid"),
+    )
+    job_type = db.Column(db.String(30), nullable=False)
+    status = db.Column(db.String(30), nullable=False, default="created")
+    idempotency_key = db.Column(db.String(64), nullable=False, unique=True)
+    parser_version = db.Column(db.String(100), nullable=False)
+    parameters = db.Column(JSONB, nullable=False, default=dict)
+    statistics = db.Column(JSONB, nullable=False, default=dict)
+    last_error = db.Column(db.String(1000))
+    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)
+    updated_at = db.Column(db.DateTime, nullable=False, default=now_china_naive)
+    started_at = db.Column(db.DateTime)
+    finished_at = db.Column(db.DateTime)
+
+    def to_dict(self) -> dict[str, Any]:
+        return {
+            "uid": str(self.uid),
+            "source_uid": str(self.source_uid),
+            "artifact_uid": str(self.artifact_uid) if self.artifact_uid else None,
+            "job_type": self.job_type,
+            "status": self.status,
+            "parser_version": self.parser_version,
+            "parameters": dict(self.parameters or {}),
+            "statistics": dict(self.statistics or {}),
+            "last_error": self.last_error,
+            "actor_uid": self.actor_uid,
+            "created_at": _iso(self.created_at),
+            "updated_at": _iso(self.updated_at),
+            "started_at": _iso(self.started_at),
+            "finished_at": _iso(self.finished_at),
+        }
+
+
+class EvidenceFragment(db.Model):
+    __tablename__ = "evidence_fragments"
+    __table_args__ = (
+        db.CheckConstraint(
+            "confidence IS NULL OR (confidence >= 0 AND confidence <= 1)",
+            name="ck_evidence_confidence",
+        ),
+        {"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,
+    )
+    artifact_uid = db.Column(
+        UUID(as_uuid=False),
+        db.ForeignKey("public.source_artifacts.uid"),
+    )
+    locator = db.Column(JSONB, nullable=False)
+    excerpt = db.Column(db.Text)
+    confidence = db.Column(db.Float)
+    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),
+            "artifact_uid": str(self.artifact_uid) if self.artifact_uid else None,
+            "locator": dict(self.locator or {}),
+            "excerpt": self.excerpt,
+            "confidence": self.confidence,
+            "created_at": _iso(self.created_at),
+        }
+
+
+class ExtractionCandidate(db.Model):
+    __tablename__ = "extraction_candidates"
+    __table_args__ = (
+        db.CheckConstraint(
+            "confidence >= 0 AND confidence <= 1",
+            name="ck_candidate_confidence",
+        ),
+        db.CheckConstraint(
+            "status IN ('candidate','accepted','rejected','ignored')",
+            name="ck_candidate_status",
+        ),
+        {"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,
+    )
+    candidate_type = db.Column(db.String(40), nullable=False)
+    normalized_data = db.Column(JSONB, nullable=False)
+    evidence_uids = db.Column(JSONB, nullable=False, default=list)
+    confidence = db.Column(db.Float, nullable=False)
+    status = db.Column(db.String(20), nullable=False, default="candidate")
+    created_at = db.Column(db.DateTime, nullable=False, default=now_china_naive)
+    updated_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),
+            "candidate_type": self.candidate_type,
+            "normalized_data": dict(self.normalized_data or {}),
+            "evidence_uids": list(self.evidence_uids or []),
+            "confidence": float(self.confidence),
+            "status": self.status,
+            "created_at": _iso(self.created_at),
+            "updated_at": _iso(self.updated_at),
+        }
+

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: 116
+x-route-count: 121
 servers:
   - url: "http://localhost:15500"
     description: "全本地隔离测试后端"
@@ -18,6 +18,8 @@ tags:
     description: "/api/system"
   - name: data_source
     description: "/api/datasource"
+  - name: data_development
+    description: "/api/development/v1"
   - name: data_flow
     description: "/api/dataflow"
   - name: business_domain
@@ -1640,6 +1642,139 @@ paths:
             application/json:
               schema:
                 $ref: '#/components/schemas/ApiEnvelope'
+  "/api/development/v1/ingestion-jobs":
+    get:
+      tags: [data_development]
+      operationId: data_development_list_ingestion_jobs_get
+      summary: "list ingestion jobs"
+      x-source: "app/api/data_development/routes.py"
+      responses:
+        "200":
+          description: "请求已由当前实现处理"
+          content:
+            application/json:
+              schema:
+                $ref: '#/components/schemas/ApiEnvelope'
+        default:
+          description: "错误响应"
+          content:
+            application/json:
+              schema:
+                $ref: '#/components/schemas/ApiEnvelope'
+    post:
+      tags: [data_development]
+      operationId: data_development_create_ingestion_job_post
+      summary: "create ingestion job"
+      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/ingestion-jobs/{job_uid}":
+    get:
+      tags: [data_development]
+      operationId: data_development_get_ingestion_job_get
+      summary: "get ingestion job"
+      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}/cancel":
+    post:
+      tags: [data_development]
+      operationId: data_development_cancel_ingestion_job_post
+      summary: "cancel 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]
+      operationId: data_development_retry_ingestion_job_post
+      summary: "retry 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/graph/node/create":
     post:
       tags: [graph]

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

@@ -0,0 +1,408 @@
+# Data Research V60-V65 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:** Upgrade DataOps data research into a tested, evidence-backed ingestion, data-element governance, and multi-domain ontology platform through V60-V65.
+
+**Architecture:** PostgreSQL is the control-plane source of truth for jobs, evidence, candidates, reviews, immutable versions, and publication state. MinIO stores source artifacts, while Neo4j stores the active semantic projection; existing Outbox processing provides idempotent cross-store publication. A new `/api/development/v1` boundary orchestrates existing datasource, metadata, business-domain, knowledge, and graph capabilities without removing legacy APIs.
+
+**Tech Stack:** Python 3.11, Flask, Flask-SQLAlchemy, Alembic, PostgreSQL JSONB, Neo4j, MinIO, SQLAlchemy inspector/catalog queries, pandas/openpyxl, python-docx, pdfplumber, pluggable OCR, Vue 2/Vuetify, pytest.
+
+## Global Constraints
+
+- Use TDD for every production behavior: add one failing test, verify the expected failure, implement minimally, then rerun the focused and phase suites.
+- Preserve `/api/meta`, `/api/bd`, `/api/datasource`, `DataMeta`, and `BusinessDomain-[:INCLUDES]->DataMeta` during compatibility migration.
+- Use stable UUID strings for all new cross-store references; Neo4j integer IDs are legacy compatibility only.
+- Database collection is read-only through `DataSourceConnectionManager.connect(uid, "metadata_collection")`; never accept arbitrary metadata SQL from the client.
+- AI/OCR output is always a candidate with confidence and evidence; it cannot directly publish a data element or ontology.
+- Credentials, connection strings, tokens, and unredacted sensitive samples must not appear in API responses, logs, model requests, or evidence previews.
+- `app/` is primary source. Update `deployment/app/` only through `deployment/sync_release.sh` at V65 release preparation.
+- Every phase ends with focused tests, the complete Python suite, frontend production build when frontend changes exist, and `git diff --check`.
+
+---
+
+## V60 — Unified model and ingestion foundation
+
+### Task 1: Add ingestion control-plane schema and models
+
+**Files:**
+- Create: `migrations/versions/20260722_100_data_research_ingestion.py`
+- Create: `app/models/data_research.py`
+- Modify: `app/models/__init__.py`
+- Modify: `tests/test_database_migrations.py`
+- Test: `tests/data_research/test_ingestion_models.py`
+
+**Interfaces:**
+- Produces: `IngestionSource`, `IngestionJob`, `SourceArtifact`, `EvidenceFragment`, `ExtractionCandidate` SQLAlchemy models.
+- Produces: job statuses `created`, `queued`, `extracting`, `normalizing`, `matching`, `awaiting_review`, `published`, `partial`, `failed`, `cancelled`.
+
+- [x] Write migration contract tests asserting the five tables, UUID foreign keys, content-hash/parser-version idempotency key, JSONB evidence locator, status checks, and non-destructive downgrade.
+- [x] Run `PYTHONPATH=. .venv/bin/pytest -q tests/test_database_migrations.py tests/data_research/test_ingestion_models.py` and verify failure because the migration and models do not exist.
+- [x] Implement the additive migration and SQLAlchemy models with `to_dict()` methods that never serialize storage credentials or internal MinIO secrets.
+- [x] Rerun the focused tests and verify they pass.
+
+### Task 2: Implement ingestion state machine and idempotent job service
+
+**Files:**
+- Create: `app/core/data_research/__init__.py`
+- Create: `app/core/data_research/errors.py`
+- Create: `app/core/data_research/models.py`
+- Create: `app/core/data_research/repository.py`
+- Create: `app/core/data_research/ingestion.py`
+- Test: `tests/data_research/test_ingestion_service.py`
+
+**Interfaces:**
+- Produces: `IngestionService.create_job(payload, actor_uid)`, `transition(job_uid, target_status, statistics=None, error=None)`, `retry(job_uid)`, and `cancel(job_uid)`.
+- Produces: `IngestionJobSpec`, `JobTransition`, and typed errors `InvalidJobTransition`, `IngestionJobNotFound`, `IngestionPayloadInvalid`.
+
+- [x] Write tests for deterministic idempotency keys, same-input job reuse, force-rerun behavior, allowed/forbidden transitions, retry from failed, cancellation, and redacted error serialization.
+- [x] Run `PYTHONPATH=. .venv/bin/pytest -q tests/data_research/test_ingestion_service.py` and verify failure because the service is absent.
+- [x] Implement pure transition rules and repository-injected orchestration; keep Flask globals out of the core service.
+- [x] Rerun the focused tests and verify they pass.
+
+### Task 3: Add the development API boundary, permissions, and task contracts
+
+**Files:**
+- Create: `app/api/data_development/__init__.py`
+- Create: `app/api/data_development/routes.py`
+- Modify: `app/__init__.py`
+- Modify: `app/core/system/permissions.py`
+- Create: `frontend/src/api/dataDevelopment.js`
+- Test: `tests/data_research/test_development_api.py`
+- Test: `tests/data_research/test_development_frontend_contract.py`
+- Modify: `tests/test_permission_matrix.py`
+
+**Interfaces:**
+- Produces: `POST/GET /api/development/v1/ingestion-jobs`, `GET /ingestion-jobs/<uid>`, `POST /<uid>/retry`, and `POST /<uid>/cancel`.
+- Produces: permissions `ingestion:run`, `ingestion:admin`, `evidence:download`, `data-elements:edit`, `data-elements:publish`, `ontologies:edit`, `ontologies:publish`.
+
+- [x] Write API and permission tests for 401, 403, editor create/list, owner cancellation, admin retry, stable response envelopes, and secret-free payloads.
+- [x] Run the focused API/permission tests and verify missing blueprint/permissions failures.
+- [x] Register the blueprint, add monotonic role permissions, inject the service through a testable factory, and add frontend API wrappers.
+- [x] Run focused tests, regenerate `docs/architecture/OPENAPI.yaml`, then rerun architecture artifact tests.
+
+### V60 gate
+
+- [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.
+
+## V61 — Database catalog and structured files
+
+### Task 4: Add deterministic SQL, CSV, and Excel extractors
+
+**Files:**
+- Create: `app/core/data_research/extractors/base.py`
+- Create: `app/core/data_research/extractors/sql.py`
+- Create: `app/core/data_research/extractors/csv.py`
+- Create: `app/core/data_research/extractors/excel.py`
+- Create: `app/core/data_research/extractors/registry.py`
+- Test: `tests/data_research/test_structured_extractors.py`
+
+**Interfaces:**
+- 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`.
+
+### Task 5: Add PostgreSQL and MySQL catalog collectors with snapshot diff
+
+**Files:**
+- Create: `app/core/data_research/catalog/base.py`
+- Create: `app/core/data_research/catalog/postgresql.py`
+- Create: `app/core/data_research/catalog/mysql.py`
+- Create: `app/core/data_research/catalog/service.py`
+- Test: `tests/data_research/test_catalog_collectors.py`
+- Test: `tests/integration/test_data_research_catalog.py`
+
+**Interfaces:**
+- 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.
+
+### Task 6: Implement data-element lifecycle and candidate decisions
+
+**Files:**
+- Create: `app/core/data_research/data_elements.py`
+- Create: `app/core/data_research/candidate_decisions.py`
+- Create: `app/core/data_research/graph_projection.py`
+- Modify: `app/api/data_development/routes.py`
+- Modify: `app/models/data_research.py`
+- Create: `tests/data_research/test_data_element_lifecycle.py`
+- Create: `tests/data_research/test_candidate_decisions.py`
+
+**Interfaces:**
+- 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.
+
+### 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.
+
+## V62 — Documents, scanned PDFs, and images
+
+### Task 7: Add DOCX and text-PDF evidence extractors
+
+**Files:**
+- Create: `app/core/data_research/extractors/docx.py`
+- Create: `app/core/data_research/extractors/pdf.py`
+- Modify: `app/core/data_research/extractors/registry.py`
+- Test: `tests/data_research/test_document_extractors.py`
+
+**Interfaces:**
+- Produces evidence locators `docx.paragraph/table/row/cell` and `pdf.page/table/bbox`.
+
+- [ ] Write in-memory fixture tests for paragraphs, tables, repeated headers, empty documents, page ordering, and exact evidence locations.
+- [ ] Run focused tests and verify missing extractors.
+- [ ] Implement structured extraction using python-docx and pdfplumber; return OCR-required classification when a PDF page has no usable text.
+- [ ] Rerun focused tests and existing DDL document parser tests.
+
+### Task 8: Add pluggable OCR and scanned-document processing
+
+**Files:**
+- Create: `app/core/data_research/ocr/base.py`
+- Create: `app/core/data_research/ocr/http_provider.py`
+- Create: `app/core/data_research/ocr/service.py`
+- Create: `app/core/data_research/extractors/image.py`
+- Test: `tests/data_research/test_ocr_service.py`
+- Test: `tests/data_research/test_image_extractor.py`
+
+**Interfaces:**
+- Produces: `OcrProvider.extract(image_bytes) -> list[OcrBlock]` with text, normalized bounding box, page, and confidence.
+- Produces: a fail-closed HTTP provider with configured endpoint, timeout, TLS verification, response-size limit, and no implicit external default.
+
+- [ ] Write tests with a local fake provider for PNG/JPEG, multi-page scanned PDF page ordering, normalized boxes, low-confidence review flags, provider timeout, malformed response, and disabled-provider failure.
+- [ ] Run tests and verify missing OCR interfaces.
+- [ ] Implement provider abstraction, fail-closed configuration, image extractor, and scanned-PDF page handoff.
+- [ ] Rerun focused tests and verify no source bytes or tokens appear in errors/logs.
+
+### Task 9: Unify file policy, artifact storage, and evidence preview
+
+**Files:**
+- Create: `app/core/data_research/file_policy.py`
+- Create: `app/core/data_research/artifacts.py`
+- Modify: `app/api/data_development/routes.py`
+- Modify: `app/config/config.py`
+- Test: `tests/data_research/test_file_policy.py`
+- Test: `tests/data_research/test_evidence_api.py`
+
+**Interfaces:**
+- Produces: one allowlist for SQL, XLS/XLSX, CSV, DOCX, PDF, PNG, JPG/JPEG; legacy DOC returns a deterministic conversion-required error.
+- Produces: `POST /sources/files` and permission-filtered `GET /evidence/<uid>`; full download requires `evidence:download`.
+
+- [ ] Write tests for extension/MIME/magic agreement, size/page limits, legacy DOC rejection, hash deduplication, unsafe filenames, evidence redaction, and download permissions.
+- [ ] Run focused tests and verify failures against current split policies.
+- [ ] Implement the unified policy, MinIO gateway abstraction, artifact hash reuse, and redacted preview endpoint.
+- [ ] Rerun focused tests plus datasource/file security regression tests.
+
+### V62 gate
+
+- [ ] Run all V62-focused tests, the complete Python suite, and `git diff --check`.
+- [ ] Record evidence in `docs/validation/data-research-v62.md` and commit V62.
+
+## V63 — Ontology MVP
+
+### Task 10: Add ontology control-plane schema and repository
+
+**Files:**
+- Create: `migrations/versions/20260722_110_data_research_ontology.py`
+- Extend: `app/models/data_research.py`
+- Create: `app/core/data_research/ontology/models.py`
+- Create: `app/core/data_research/ontology/repository.py`
+- Test: `tests/data_research/test_ontology_schema.py`
+- Test: `tests/data_research/test_ontology_repository.py`
+
+**Interfaces:**
+- Produces: `Ontology`, immutable `OntologyVersion`, `OntologyChangeSet`, and `OntologyPublishRun`.
+- Produces graph document fields `classes`, `properties`, `relations`, `constraints`, `domain_links`, and `element_mappings`.
+
+- [ ] Write schema and repository tests for unique ontology codes, immutable published versions, parent version, content hash, many-domain role links, and optimistic draft revisions.
+- [ ] Run focused tests and verify missing schema/repository failures.
+- [ ] Implement additive migration, dataclasses, canonical graph hashing, and repository operations.
+- [ ] Rerun focused and migration tests.
+
+### Task 11: Implement ontology validation, publication, projection, and rollback
+
+**Files:**
+- Create: `app/core/data_research/ontology/validation.py`
+- Create: `app/core/data_research/ontology/publication.py`
+- Create: `app/core/data_research/ontology/projection.py`
+- Test: `tests/data_research/test_ontology_validation.py`
+- Test: `tests/data_research/test_ontology_publication.py`
+
+**Interfaces:**
+- Produces validation codes for duplicate names, dangling edges, inheritance cycles, invalid cardinality, ownerless ontology, and unmapped required property.
+- Produces events `ontology.version_published` and `ontology.rollback_version_created`.
+
+- [ ] Write tests for every validation code, publish idempotency, outbox atomicity, active-version projection, retryable Neo4j failure, version diff, and rollback-as-new-version behavior.
+- [ ] Run focused tests and verify missing service failures.
+- [ ] Implement pure validation, publish transaction, UID-based Neo4j projection, diff, and rollback version creation.
+- [ ] Rerun focused tests plus cross-store/outbox tests.
+
+### Task 12: Add ontology APIs and Vue 2 ontology center
+
+**Files:**
+- Modify: `app/api/data_development/routes.py`
+- Modify: `frontend/src/api/dataDevelopment.js`
+- Create: `frontend/src/views/dataGovernance/ontology/index.vue`
+- Create: `frontend/src/views/dataGovernance/ontology/workbench.vue`
+- Modify: `frontend/src/router/routes.js`
+- Test: `tests/data_research/test_ontology_api.py`
+- Test: `tests/data_research/test_ontology_frontend_contract.py`
+
+**Interfaces:**
+- Produces ontology CRUD, graph patch, validate, publish, diff, and rollback endpoints under `/api/development/v1`.
+- Produces permission-gated ontology list and workbench routes.
+
+- [ ] Write API tests for editor drafts, publisher release, multi-domain roles, ETag conflicts, validation errors, and viewer read-only behavior; write source contract tests for routes and secret-free rendering.
+- [ ] Run focused tests and verify missing endpoints/pages.
+- [ ] Implement API handlers and Vue pages with graph JSON editing/preview, validation panel, version diff, and publish confirmation.
+- [ ] Run focused tests and frontend production build.
+
+### V63 gate
+
+- [ ] Run all ontology tests, complete Python suite, frontend build, and `git diff --check`.
+- [ ] Record evidence in `docs/validation/data-research-v63.md` and commit V63.
+
+## V64 — Dynamic ontology and knowledge services
+
+### Task 13: Generate and review explainable ontology change sets
+
+**Files:**
+- Create: `app/core/data_research/ontology/suggestions.py`
+- Create: `app/core/data_research/ontology/change_sets.py`
+- Modify: `app/api/data_development/routes.py`
+- Test: `tests/data_research/test_ontology_suggestions.py`
+- Test: `tests/data_research/test_ontology_change_sets.py`
+
+**Interfaces:**
+- Produces deterministic suggestions from business domains, published data elements, foreign-key evidence, and aliases before optional AI suggestions.
+- Produces decisions `accept`, `reject`, `edit` with actor, reason, evidence UIDs, confidence, and model/prompt version.
+
+- [ ] Write tests for deterministic class/property/relation proposals, evidence-required AI proposals, confidence ordering, conflict detection, partial decisions, and the guarantee that undecided suggestions never publish.
+- [ ] Run focused tests and verify missing services.
+- [ ] Implement rule suggestions, injected AI suggestion port, immutable decision audit, and accepted-change application to a draft version.
+- [ ] Rerun focused tests and ontology publication regressions.
+
+### Task 14: Add JSON/RDF/OWL exchange and governance knowledge synchronization
+
+**Files:**
+- Create: `app/core/data_research/ontology/exchange.py`
+- Create: `app/core/data_research/ontology/knowledge_sync.py`
+- Modify: `app/api/data_development/routes.py`
+- Test: `tests/data_research/test_ontology_exchange.py`
+- Test: `tests/data_research/test_ontology_knowledge_sync.py`
+
+**Interfaces:**
+- Produces deterministic JSON export and bounded RDF/OWL XML export/import with stable platform UIDs.
+- Consumes: `app/core/knowledge/document_builder.py`; produces versioned governance documents only for published ontology versions.
+
+- [ ] Write round-trip tests for classes/properties/relations/domain links, unsafe XML rejection, import size limits, deterministic hashes, secret redaction, and knowledge-document version/hash behavior.
+- [ ] Run focused tests and verify missing exchange/sync modules.
+- [ ] Implement bounded import/export, canonical namespaces, publication-triggered knowledge documents, and idempotent chunk sync.
+- [ ] Rerun focused tests plus governance document builder tests.
+
+### Task 15: Expose read-only semantic queries and MCP context
+
+**Files:**
+- Create: `app/core/data_research/ontology/query.py`
+- Modify: `app/api/data_development/routes.py`
+- Modify: `app/core/mcp/context.py`
+- Test: `tests/data_research/test_semantic_query.py`
+- Test: `tests/mcp/test_data_research_context.py`
+
+**Interfaces:**
+- Produces bounded read-only queries from ontology property to data element, physical field, source evidence, and business domain.
+- Produces MCP context tools that return published data only and enforce permission scope.
+
+- [ ] Write tests for bounded traversal, unpublished exclusion, permission filtering, stable pagination, no arbitrary Cypher input, and MCP result redaction.
+- [ ] Run focused tests and verify missing query/tool behavior.
+- [ ] Implement allowlisted query methods and MCP context adapters.
+- [ ] Rerun focused tests and existing MCP security/runtime suites.
+
+### V64 gate
+
+- [ ] Run all V64-focused, governance knowledge, MCP, complete Python, and frontend build tests.
+- [ ] Record evidence in `docs/validation/data-research-v64.md` and commit V64.
+
+## V65 — Production hardening and acceptance
+
+### Task 16: Add limits, metrics, reconciliation, and recovery
+
+**Files:**
+- Create: `app/core/data_research/operations.py`
+- Create: `app/commands/reconcile_data_research.py`
+- Modify: `app/core/system/health.py`
+- Test: `tests/data_research/test_operations.py`
+- Test: `tests/integration/test_data_research_reconciliation.py`
+
+**Interfaces:**
+- Produces metrics for job states/duration, failure stage, candidate acceptance, review time, ontology coverage, and projection lag.
+- Produces reconciliation reports for PostgreSQL versions, Neo4j active projections, MinIO artifacts, and governance documents.
+
+- [ ] Write tests for concurrency/resource limits, stale-job recovery, dead-letter reporting, projection mismatch repair plans, artifact-missing detection, and secret-free health output.
+- [ ] Run focused tests and verify missing operations module.
+- [ ] Implement bounded counters, recovery transitions, read-only audit, explicit repair mode, and health component.
+- [ ] Rerun focused tests plus health/outbox tests.
+
+### Task 17: Complete task center, ingestion, data-element, and ontology UX
+
+**Files:**
+- Create: `frontend/src/views/dataGovernance/development/index.vue`
+- Create: `frontend/src/views/dataGovernance/development/ingestion.vue`
+- Create: `frontend/src/views/dataGovernance/development/tasks.vue`
+- Create: `frontend/src/views/dataGovernance/development/review.vue`
+- Modify: `frontend/src/views/dataGovernance/metadata/index.vue`
+- Modify: `frontend/src/router/routes.js`
+- Test: `tests/data_research/test_data_research_frontend_contract.py`
+
+**Interfaces:**
+- Produces source selection, scope configuration, upload, progress, evidence preview, batch decisions, data-element lifecycle, ontology generation, validation, publish, and rollback navigation.
+
+- [ ] Write frontend source-contract tests for every required route/API/permission, supported format copy, evidence navigation, status labels, no credential fields, and viewer read-only controls.
+- [ ] Run focused tests and verify missing pages/contracts.
+- [ ] Implement Vue 2 pages using existing list/dialog/form patterns and permission metadata.
+- [ ] Run focused tests and `npm --prefix frontend run build` with Node 24.
+
+### Task 18: Release parity, documentation, and end-to-end acceptance
+
+**Files:**
+- Create: `tests/acceptance/test_data_research_v60_v65.py`
+- Create: `docs/validation/data-research-v60-v65-acceptance.md`
+- Modify: `docs/architecture/ARCHITECTURE_OVERVIEW.md`
+- Modify: `docs/architecture/DATA_MODEL.md`
+- Modify: `docs/architecture/NEXT_ITERATION_ROADMAP.md`
+- Modify: `docs/architecture/OPENAPI.yaml`
+- Modify: `deployment/app/` via `deployment/sync_release.sh`
+
+**Interfaces:**
+- Produces one acceptance entry point covering source ingestion, evidence, data-element publication, two-domain ontology, dynamic change set, publish, rollback, knowledge sync, RBAC, reconciliation, and redaction.
+
+- [ ] Write acceptance tests that initially fail on any missing V60-V65 capability and use deterministic fixtures/fake OCR/AI ports while exercising real Flask services and persistence contracts.
+- [ ] Run `PYTHONPATH=. .venv/bin/pytest -q tests/acceptance/test_data_research_v60_v65.py` and verify requirement-specific failures before final wiring.
+- [ ] Complete release wiring, regenerate OpenAPI, run `deployment/sync_release.sh`, and verify `diff -qr app deployment/app` is empty for application source.
+- [ ] Run the final acceptance matrix below and record exact commands, counts, skips, build output, and environment-limited checks in the acceptance document.
+
+### V65 and final acceptance gate
+
+- [ ] `PYTHONPATH=. .venv/bin/pytest -q tests/data_research tests/acceptance/test_data_research_v60_v65.py`
+- [ ] `PYTHONPATH=. .venv/bin/pytest -q`
+- [ ] `npm --prefix frontend run build`
+- [ ] `python scripts/generate_openapi.py --output docs/architecture/OPENAPI.yaml` followed by `PYTHONPATH=. .venv/bin/pytest -q tests/test_architecture_artifacts.py`.
+- [ ] `diff -qr app deployment/app` and `git diff --check`.
+- [ ] When Docker is available, run catalog, reconciliation, RBAC, MCP, and migration integration tests against `deploy/docker/docker-compose.yml`.
+- [ ] Mark V60-V65 complete only if every non-environment-limited requirement has fresh passing evidence and every skipped external check is explicitly listed with its blocker.

docs/validation/data-research-v60.md

@@ -0,0 +1,35 @@
+# Data Research V60 Validation
+
+Date: 2026-07-22  
+Branch: `codex/data-research-v60-v65`
+
+## Delivered
+
+- Additive PostgreSQL control-plane migration for ingestion sources, artifacts, jobs, evidence, and extraction candidates.
+- Secret-free SQLAlchemy model serialization.
+- Deterministic job idempotency, force rerun, validated state transitions, retry, cancellation, and error redaction.
+- `/api/development/v1/ingestion-jobs` create/list/detail/retry/cancel boundary.
+- Server-enforced ingestion, evidence, data-element, and ontology permission primitives.
+- Vue 2 frontend ingestion-job API wrapper.
+- OpenAPI generator support for the new blueprint.
+
+## TDD evidence
+
+- Task 1 RED: 4 expected failures for missing migration and models.
+- Task 1 GREEN: `6 passed, 1 skipped`.
+- Task 2 RED: 5 expected failures for missing ingestion service.
+- Task 2 GREEN: `8 passed`; the cycle also caught and fixed invalid list-valued parameters being normalized as an empty object.
+- Task 3 RED: missing blueprint, frontend wrapper, and permission-policy failures.
+- Task 3 GREEN: `8 passed`.
+- OpenAPI regression RED: generated route count 116 versus source count 121.
+- OpenAPI regression GREEN: generator emitted 121 operations and the combined contract suite passed 15 tests.
+
+## Phase gate
+
+- V60 focused suite: `26 passed, 1 skipped in 0.91s`.
+- Full Python suite: `299 passed, 23 skipped, 59 subtests passed in 3.67s`.
+- Frontend production build: Node `v24.14.0`, build exit code 0. Existing dependency-age, console, CSS-order, and asset-size warnings remain; no build errors.
+- `git diff --check`: exit code 0.
+
+The single focused skip is the existing environment-gated PostgreSQL migration integration test. It will be exercised in final Docker acceptance when the integration database variables are available.
+

frontend/src/api/dataDevelopment.js

@@ -0,0 +1,17 @@
+import http from '@/utils/request'
+
+const BASE = '/development/v1/ingestion-jobs'
+
+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 retryIngestionJob = uid => http.post(`${BASE}/${uid}/retry`)
+export const cancelIngestionJob = uid => http.post(`${BASE}/${uid}/cancel`)
+
+export default {
+  createIngestionJob,
+  getIngestionJobs,
+  getIngestionJob,
+  retryIngestionJob,
+  cancelIngestionJob
+}

migrations/versions/20260722_100_data_research_ingestion.py

@@ -0,0 +1,111 @@
+"""Add the data-research ingestion control plane."""
+
+from alembic import op
+
+
+revision = "20260722_100"
+down_revision = "20260719_90"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    op.execute(
+        """
+        CREATE TABLE IF NOT EXISTS public.ingestion_sources (
+            uid UUID PRIMARY KEY,
+            source_type VARCHAR(20) NOT NULL
+                CHECK (source_type IN ('database','file','ddl')),
+            name VARCHAR(300) NOT NULL,
+            config JSONB NOT NULL DEFAULT '{}'::jsonb,
+            permission_scope JSONB NOT NULL DEFAULT '{}'::jsonb,
+            status VARCHAR(20) NOT NULL DEFAULT 'active'
+                CHECK (status IN ('active','disabled')),
+            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.source_artifacts (
+            uid UUID PRIMARY KEY,
+            source_uid UUID NOT NULL
+                REFERENCES public.ingestion_sources(uid),
+            filename VARCHAR(500) NOT NULL,
+            media_type VARCHAR(200) NOT NULL,
+            size_bytes BIGINT NOT NULL CHECK (size_bytes >= 0),
+            content_hash CHAR(64) NOT NULL,
+            storage_ref TEXT NOT NULL,
+            parser_version VARCHAR(100) NOT NULL,
+            created_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP,
+            UNIQUE (source_uid, content_hash, parser_version)
+        );
+
+        CREATE TABLE IF NOT EXISTS public.ingestion_jobs (
+            uid UUID PRIMARY KEY,
+            source_uid UUID NOT NULL
+                REFERENCES public.ingestion_sources(uid),
+            artifact_uid UUID
+                REFERENCES public.source_artifacts(uid),
+            job_type VARCHAR(30) NOT NULL,
+            status VARCHAR(30) NOT NULL DEFAULT 'created'
+                CHECK (status IN (
+                    'created','queued','extracting','normalizing','matching',
+                    'awaiting_review','published','partial','failed','cancelled'
+                )),
+            idempotency_key CHAR(64) NOT NULL UNIQUE,
+            parser_version VARCHAR(100) NOT NULL,
+            parameters JSONB NOT NULL DEFAULT '{}'::jsonb,
+            statistics JSONB NOT NULL DEFAULT '{}'::jsonb,
+            last_error VARCHAR(1000),
+            actor_uid VARCHAR(100),
+            force_rerun BOOLEAN NOT NULL DEFAULT FALSE,
+            created_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP,
+            updated_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP,
+            started_at TIMESTAMPTZ,
+            finished_at TIMESTAMPTZ
+        );
+
+        CREATE TABLE IF NOT EXISTS public.evidence_fragments (
+            uid UUID PRIMARY KEY,
+            job_uid UUID NOT NULL
+                REFERENCES public.ingestion_jobs(uid) ON DELETE CASCADE,
+            artifact_uid UUID
+                REFERENCES public.source_artifacts(uid),
+            locator JSONB NOT NULL,
+            excerpt TEXT,
+            confidence DOUBLE PRECISION
+                CHECK (confidence IS NULL OR (confidence >= 0 AND confidence <= 1)),
+            created_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP
+        );
+
+        CREATE TABLE IF NOT EXISTS public.extraction_candidates (
+            uid UUID PRIMARY KEY,
+            job_uid UUID NOT NULL
+                REFERENCES public.ingestion_jobs(uid) ON DELETE CASCADE,
+            candidate_type VARCHAR(40) NOT NULL,
+            normalized_data JSONB NOT NULL,
+            evidence_uids JSONB NOT NULL DEFAULT '[]'::jsonb,
+            confidence DOUBLE PRECISION NOT NULL
+                CHECK (confidence >= 0 AND confidence <= 1),
+            status VARCHAR(20) NOT NULL DEFAULT 'candidate'
+                CHECK (status IN ('candidate','accepted','rejected','ignored')),
+            created_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP,
+            updated_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP
+        );
+
+        CREATE INDEX IF NOT EXISTS idx_ingestion_jobs_status_created
+            ON public.ingestion_jobs(status, created_at);
+        CREATE INDEX IF NOT EXISTS idx_ingestion_jobs_source
+            ON public.ingestion_jobs(source_uid, created_at);
+        CREATE INDEX IF NOT EXISTS idx_evidence_fragments_job
+            ON public.evidence_fragments(job_uid);
+        CREATE INDEX IF NOT EXISTS idx_extraction_candidates_job_status
+            ON public.extraction_candidates(job_uid, status);
+        """
+    )
+
+
+def downgrade() -> None:
+    # Additive governance data is intentionally preserved on app rollback.
+    pass
+

scripts/generate_openapi.py

@@ -19,6 +19,7 @@ PREFIXES = {
     "graph": "/api/graph",
     "system": "/api/system",
     "data_source": "/api/datasource",
+    "data_development": "/api/development/v1",
     "data_flow": "/api/dataflow",
     "business_domain": "/api/bd",
     "data_factory": "/api/datafactory",

tests/data_research/__init__.py

@@ -0,0 +1 @@
+

tests/data_research/test_development_api.py

@@ -0,0 +1,141 @@
+from __future__ import annotations
+
+from dataclasses import replace
+
+import pytest
+
+
+class FakeDevelopmentService:
+    def __init__(self, record):
+        self.record = record
+        self.actions = []
+
+    def create_job(self, payload, actor_uid):
+        self.actions.append(("create", payload, actor_uid))
+        return replace(self.record, actor_uid=actor_uid), True
+
+    def list_jobs(self, filters=None):
+        self.actions.append(("list", filters or {}))
+        return [self.record]
+
+    def get_job(self, uid):
+        self.actions.append(("get", uid))
+        return self.record
+
+    def retry(self, uid):
+        self.actions.append(("retry", uid))
+        return replace(self.record, status="queued", last_error=None)
+
+    def cancel(self, uid):
+        self.actions.append(("cancel", uid))
+        return replace(self.record, status="cancelled")
+
+
+@pytest.fixture()
+def development_client(monkeypatch):
+    from flask import request
+
+    from app import create_app
+    from app.api.data_development import routes
+    from app.core.data_research.models import IngestionJobRecord
+    from app.core.system import permissions
+
+    record = IngestionJobRecord(
+        uid="00000000-0000-0000-0000-000000000010",
+        source_uid="00000000-0000-0000-0000-000000000001",
+        artifact_uid="00000000-0000-0000-0000-000000000002",
+        job_type="file_extract",
+        parser_version="csv-v1",
+        idempotency_key="a" * 64,
+        actor_uid="editor-1",
+        parameters={"schema": "public"},
+    )
+    service = FakeDevelopmentService(record)
+
+    def identity():
+        header = request.headers.get("Authorization", "")
+        if header == "Bearer viewer":
+            return {"id": "viewer-1", "roles": ["viewer"]}
+        if header == "Bearer editor":
+            return {"id": "editor-1", "roles": ["editor"]}
+        if header == "Bearer other-editor":
+            return {"id": "editor-2", "roles": ["editor"]}
+        if header == "Bearer admin":
+            return {"id": "admin-1", "roles": ["admin"]}
+        return None
+
+    monkeypatch.setattr(permissions, "authenticate_request", identity)
+    monkeypatch.setattr(routes, "get_ingestion_service", lambda: service)
+    app = create_app()
+    app.config.update(TESTING=True)
+    return app.test_client(), service
+
+
+def job_payload():
+    return {
+        "source_uid": "00000000-0000-0000-0000-000000000001",
+        "artifact_uid": "00000000-0000-0000-0000-000000000002",
+        "job_type": "file_extract",
+        "parser_version": "csv-v1",
+        "parameters": {"schema": "public"},
+        "password": "must-not-return",
+    }
+
+
+def test_ingestion_api_requires_authentication_and_run_permission(development_client):
+    client, _service = development_client
+
+    assert client.post("/api/development/v1/ingestion-jobs", json=job_payload()).status_code == 401
+    assert client.post(
+        "/api/development/v1/ingestion-jobs",
+        json=job_payload(),
+        headers={"Authorization": "Bearer viewer"},
+    ).status_code == 403
+
+
+def test_editor_creates_and_lists_secret_free_jobs(development_client):
+    client, service = development_client
+    created = client.post(
+        "/api/development/v1/ingestion-jobs",
+        json=job_payload(),
+        headers={"Authorization": "Bearer editor"},
+    )
+    listed = client.get(
+        "/api/development/v1/ingestion-jobs?status=created",
+        headers={"Authorization": "Bearer editor"},
+    )
+
+    assert created.status_code == 201
+    assert created.get_json()["data"]["uid"].endswith("0010")
+    assert "must-not-return" not in created.get_data(as_text=True)
+    assert listed.status_code == 200
+    assert listed.get_json()["data"]["total"] == 1
+    assert service.actions[-1] == ("list", {"status": "created"})
+
+
+def test_retry_is_admin_only_and_cancel_is_owner_or_admin(development_client):
+    client, _service = development_client
+    uid = "00000000-0000-0000-0000-000000000010"
+
+    editor_retry = client.post(
+        f"/api/development/v1/ingestion-jobs/{uid}/retry",
+        headers={"Authorization": "Bearer editor"},
+    )
+    admin_retry = client.post(
+        f"/api/development/v1/ingestion-jobs/{uid}/retry",
+        headers={"Authorization": "Bearer admin"},
+    )
+    other_cancel = client.post(
+        f"/api/development/v1/ingestion-jobs/{uid}/cancel",
+        headers={"Authorization": "Bearer other-editor"},
+    )
+    owner_cancel = client.post(
+        f"/api/development/v1/ingestion-jobs/{uid}/cancel",
+        headers={"Authorization": "Bearer editor"},
+    )
+
+    assert editor_retry.status_code == 403
+    assert admin_retry.status_code == 200
+    assert other_cancel.status_code == 403
+    assert owner_cancel.status_code == 200
+

tests/data_research/test_development_frontend_contract.py

@@ -0,0 +1,20 @@
+from pathlib import Path
+
+
+ROOT = Path(__file__).resolve().parents[2]
+
+
+def test_data_development_frontend_api_covers_v60_job_contract():
+    source = (ROOT / "frontend/src/api/dataDevelopment.js").read_text(
+        encoding="utf-8"
+    )
+
+    assert "/development/v1/ingestion-jobs" in source
+    assert "createIngestionJob" in source
+    assert "getIngestionJobs" in source
+    assert "getIngestionJob" in source
+    assert "retryIngestionJob" in source
+    assert "cancelIngestionJob" in source
+    assert "password" not in source.lower()
+    assert "credentials" not in source.lower()
+

tests/data_research/test_ingestion_models.py

@@ -0,0 +1,85 @@
+from __future__ import annotations
+
+from sqlalchemy import CheckConstraint, UniqueConstraint
+
+
+def test_ingestion_models_expose_stable_control_plane_tables():
+    from app.models.data_research import (
+        EvidenceFragment,
+        ExtractionCandidate,
+        IngestionJob,
+        IngestionSource,
+        SourceArtifact,
+    )
+
+    assert IngestionSource.__tablename__ == "ingestion_sources"
+    assert IngestionJob.__tablename__ == "ingestion_jobs"
+    assert SourceArtifact.__tablename__ == "source_artifacts"
+    assert EvidenceFragment.__tablename__ == "evidence_fragments"
+    assert ExtractionCandidate.__tablename__ == "extraction_candidates"
+
+    assert IngestionJob.__table__.c.source_uid.foreign_keys
+    assert IngestionJob.__table__.c.artifact_uid.foreign_keys
+    assert EvidenceFragment.__table__.c.job_uid.foreign_keys
+    assert ExtractionCandidate.__table__.c.job_uid.foreign_keys
+
+
+def test_ingestion_job_has_idempotency_and_status_constraints():
+    from app.models.data_research import IngestionJob
+
+    constraints = list(IngestionJob.__table__.constraints)
+    unique_columns = {
+        tuple(column.name for column in constraint.columns)
+        for constraint in constraints
+        if isinstance(constraint, UniqueConstraint)
+    }
+    check_sql = " ".join(
+        str(constraint.sqltext)
+        for constraint in constraints
+        if isinstance(constraint, CheckConstraint)
+    )
+
+    assert ("idempotency_key",) in unique_columns
+    for status in (
+        "created",
+        "queued",
+        "extracting",
+        "normalizing",
+        "matching",
+        "awaiting_review",
+        "published",
+        "partial",
+        "failed",
+        "cancelled",
+    ):
+        assert status in check_sql
+
+
+def test_model_serialization_redacts_control_plane_secrets():
+    from app.models.data_research import IngestionSource, SourceArtifact
+
+    source = IngestionSource(
+        uid="00000000-0000-0000-0000-000000000001",
+        source_type="database",
+        name="orders",
+        config={"secret_ref": "vault://orders", "schema": "public"},
+        permission_scope={"roles": ["editor"]},
+        status="active",
+    )
+    artifact = SourceArtifact(
+        uid="00000000-0000-0000-0000-000000000002",
+        source_uid=source.uid,
+        filename="orders.csv",
+        media_type="text/csv",
+        size_bytes=12,
+        content_hash="a" * 64,
+        storage_ref="minio://private/orders.csv?token=secret",
+        parser_version="csv-v1",
+    )
+
+    assert "config" not in source.to_dict()
+    assert "vault://orders" not in str(source.to_dict())
+    assert "storage_ref" not in artifact.to_dict()
+    assert "token=secret" not in str(artifact.to_dict())
+    assert artifact.to_dict()["content_hash"] == "a" * 64
+

tests/data_research/test_ingestion_service.py

@@ -0,0 +1,137 @@
+from __future__ import annotations
+
+from dataclasses import replace
+
+import pytest
+
+
+class MemoryJobRepository:
+    def __init__(self):
+        self.records = {}
+
+    def get(self, uid):
+        return self.records.get(uid)
+
+    def get_by_idempotency_key(self, key):
+        return next(
+            (item for item in self.records.values() if item.idempotency_key == key),
+            None,
+        )
+
+    def add(self, record):
+        self.records[record.uid] = record
+        return record
+
+    def save(self, record):
+        self.records[record.uid] = record
+        return record
+
+
+def payload(**overrides):
+    value = {
+        "source_uid": "00000000-0000-0000-0000-000000000001",
+        "artifact_uid": "00000000-0000-0000-0000-000000000002",
+        "job_type": "file_extract",
+        "parser_version": "csv-v1",
+        "parameters": {"delimiter": ",", "schema": "public"},
+    }
+    value.update(overrides)
+    return value
+
+
+def test_same_canonical_input_reuses_existing_job():
+    from app.core.data_research.ingestion import IngestionService
+
+    repository = MemoryJobRepository()
+    service = IngestionService(repository)
+
+    first, first_created = service.create_job(payload(), actor_uid="editor-1")
+    second, second_created = service.create_job(
+        payload(parameters={"schema": "public", "delimiter": ","}),
+        actor_uid="editor-2",
+    )
+
+    assert first_created is True
+    assert second_created is False
+    assert second.uid == first.uid
+    assert second.idempotency_key == first.idempotency_key
+
+
+def test_force_rerun_creates_a_distinct_job_without_changing_canonical_key_rules():
+    from app.core.data_research.ingestion import IngestionService
+
+    repository = MemoryJobRepository()
+    service = IngestionService(repository, nonce_factory=lambda: "rerun-1")
+    first, _ = service.create_job(payload(), actor_uid="editor-1")
+    rerun, created = service.create_job(
+        payload(force_rerun=True),
+        actor_uid="editor-1",
+    )
+
+    assert created is True
+    assert rerun.uid != first.uid
+    assert rerun.idempotency_key != first.idempotency_key
+    assert rerun.force_rerun is True
+
+
+def test_job_transitions_follow_the_declared_state_machine():
+    from app.core.data_research.errors import InvalidJobTransition
+    from app.core.data_research.ingestion import IngestionService
+
+    repository = MemoryJobRepository()
+    service = IngestionService(repository)
+    job, _ = service.create_job(payload(), actor_uid="editor-1")
+
+    for status in (
+        "queued",
+        "extracting",
+        "normalizing",
+        "matching",
+        "awaiting_review",
+        "published",
+    ):
+        job = service.transition(job.uid, status, statistics={"stage": status})
+        assert job.status == status
+
+    with pytest.raises(InvalidJobTransition, match="published -> queued"):
+        service.transition(job.uid, "queued")
+
+
+def test_failed_job_can_retry_and_running_job_can_cancel():
+    from app.core.data_research.ingestion import IngestionService
+
+    repository = MemoryJobRepository()
+    service = IngestionService(repository)
+    job, _ = service.create_job(payload(), actor_uid="editor-1")
+    service.transition(job.uid, "queued")
+    failed = service.transition(
+        job.uid,
+        "failed",
+        error="password=clear-secret token=abc123 connection refused",
+    )
+
+    assert "clear-secret" not in failed.last_error
+    assert "abc123" not in failed.last_error
+    assert "[redacted]" in failed.last_error
+
+    retried = service.retry(job.uid)
+    assert retried.status == "queued"
+    assert retried.last_error is None
+
+    cancelled = service.cancel(job.uid)
+    assert cancelled.status == "cancelled"
+
+
+def test_missing_or_invalid_payload_fails_before_repository_write():
+    from app.core.data_research.errors import IngestionPayloadInvalid
+    from app.core.data_research.ingestion import IngestionService
+
+    repository = MemoryJobRepository()
+    service = IngestionService(repository)
+
+    with pytest.raises(IngestionPayloadInvalid, match="source_uid"):
+        service.create_job(payload(source_uid=""), actor_uid="editor-1")
+    with pytest.raises(IngestionPayloadInvalid, match="parameters"):
+        service.create_job(payload(parameters=[]), actor_uid="editor-1")
+
+    assert repository.records == {}

tests/test_database_migrations.py

@@ -35,9 +35,38 @@ EXPECTED_UPGRADED_TABLES = {
     "workflow_dual_runs",
     "workflow_reconciliation_reports",
     "workflow_cutover_operations",
+    "ingestion_sources",
+    "source_artifacts",
+    "ingestion_jobs",
+    "evidence_fragments",
+    "extraction_candidates",
 }
 
 
+def test_data_research_ingestion_migration_is_additive_and_constrained():
+    migration = (
+        ROOT
+        / "migrations"
+        / "versions"
+        / "20260722_100_data_research_ingestion.py"
+    ).read_text(encoding="utf-8")
+
+    assert 'revision = "20260722_100"' in migration
+    assert 'down_revision = "20260719_90"' in migration
+    for table in (
+        "ingestion_sources",
+        "source_artifacts",
+        "ingestion_jobs",
+        "evidence_fragments",
+        "extraction_candidates",
+    ):
+        assert f"CREATE TABLE IF NOT EXISTS public.{table}" in migration
+    assert "idempotency_key" in migration
+    assert "parser_version" in migration
+    assert "locator JSONB" 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

@@ -10,6 +10,27 @@ def test_fixed_role_permission_matrix_is_monotonic():
     assert "governance:edit" in editor
     assert "users:manage" in admin
     assert "workflow:activate" in admin
+    assert "ingestion:run" in editor
+    assert "data-elements:edit" in editor
+    assert "ontologies:edit" in editor
+    assert "ingestion:admin" in admin
+    assert "evidence:download" in admin
+    assert "data-elements:publish" in admin
+    assert "ontologies:publish" in admin
+
+
+def test_data_development_paths_have_specific_write_policies():
+    from app.core.system.permissions import permission_for_request
+
+    assert permission_for_request(
+        "/api/development/v1/ingestion-jobs", "POST"
+    ) == ("ingestion:run",)
+    assert permission_for_request(
+        "/api/development/v1/ingestion-jobs/job-1/retry", "POST"
+    ) == ("ingestion:admin",)
+    assert permission_for_request(
+        "/api/development/v1/ingestion-jobs/job-1/cancel", "POST"
+    ) == ("ingestion:run",)
 
 
 def test_permissions_are_derived_server_side_from_roles():