feat: add device asset catalog

app/api/data_development/routes.py

@@ -101,6 +101,19 @@ def get_data_element_service():
     )
 
 
+def get_device_asset_service():
+    from app.core.data_research.device_asset_repository import (
+        SqlAlchemyDeviceAssetRepository,
+    )
+    from app.core.data_research.device_assets import DeviceAssetService
+
+    return DeviceAssetService(
+        SqlAlchemyDeviceAssetRepository(db.session),
+        commit=db.session.commit,
+        rollback=db.session.rollback,
+    )
+
+
 def get_candidate_decision_service():
     from app.core.data_research.candidate_decisions import CandidateDecisionService
     from app.core.data_research.data_elements import DataElementService
@@ -316,6 +329,94 @@ def _ontology_version(record):
     }
 
 
+def _iso(value):
+    return value.isoformat() if value else None
+
+
+def _device_asset(record):
+    return {
+        "uid": str(record.uid),
+        "asset_type": record.asset_type,
+        "name": record.name,
+        "status": record.status,
+        "current_version": int(record.current_version),
+        "location": record.location,
+        "organization": record.organization,
+        "responsible_person": record.responsible_person,
+        "attributes": dict(record.attributes or {}),
+        "created_by": record.created_by,
+        "updated_by": record.updated_by,
+        "created_at": _iso(record.created_at),
+        "updated_at": _iso(record.updated_at),
+    }
+
+
+def _device_asset_mapping(record):
+    return {
+        "uid": str(record.uid),
+        "asset_uid": str(record.asset_uid),
+        "source_uid": str(record.source_uid),
+        "source_entity": record.source_entity,
+        "asset_type": record.asset_type,
+        "source_code": record.source_code,
+        "source_updated_at": _iso(record.source_updated_at),
+        "first_seen_at": _iso(record.first_seen_at),
+        "last_seen_at": _iso(record.last_seen_at),
+    }
+
+
+def _device_asset_detail(detail):
+    data = _device_asset(detail.asset)
+    data["source_mappings"] = [
+        _device_asset_mapping(mapping)
+        for mapping in detail.mappings
+    ]
+    return data
+
+
+def _device_asset_version(record):
+    return {
+        "uid": str(record.uid),
+        "asset_uid": str(record.asset_uid),
+        "version": int(record.version),
+        "snapshot": dict(record.snapshot or {}),
+        "source_mapping_uid": str(record.source_mapping_uid),
+        "actor_uid": record.actor_uid,
+        "created_at": _iso(record.created_at),
+    }
+
+
+def _device_asset_import_result(result):
+    return {
+        "records": [
+            {
+                "action": item.action,
+                "asset": _device_asset(item.asset),
+                "source_mapping": _device_asset_mapping(item.mapping),
+            }
+            for item in result.items
+        ],
+        "created_count": int(result.created_count),
+        "updated_count": int(result.updated_count),
+        "unchanged_count": int(result.unchanged_count),
+    }
+
+
+def _device_asset_page(name, *, default, maximum):
+    from app.core.data_research.errors import DeviceAssetInvalid
+
+    raw = request.args.get(name)
+    try:
+        value = default if raw in (None, "") else int(raw)
+    except (TypeError, ValueError) as error:
+        raise DeviceAssetInvalid(f"{name} must be an integer") from error
+    if value < 1 or value > maximum:
+        raise DeviceAssetInvalid(
+            f"{name} must be between 1 and {maximum}"
+        )
+    return value
+
+
 def _error(error):
     if isinstance(error, DataResearchError):
         return (
@@ -468,6 +569,91 @@ def list_data_elements():
         return _error(error)
 
 
+@bp.route("/device-assets", methods=["GET"])
+def list_device_assets():
+    filters = {
+        name: request.args.get(name)
+        for name in ("keyword", "asset_type", "status", "source_uid")
+        if request.args.get(name)
+    }
+    try:
+        page = _device_asset_page(
+            "page",
+            default=1,
+            maximum=1_000_000,
+        )
+        page_size = _device_asset_page(
+            "page_size",
+            default=20,
+            maximum=100,
+        )
+        records, total = get_device_asset_service().search(
+            filters,
+            page=page,
+            page_size=page_size,
+        )
+        return jsonify(
+            success(
+                {
+                    "records": [
+                        _device_asset_detail(record)
+                        for record in records
+                    ],
+                    "total": int(total),
+                    "page": page,
+                    "page_size": page_size,
+                }
+            )
+        ), 200
+    except Exception as error:
+        return _error(error)
+
+
+@bp.route("/device-assets/import", methods=["POST"])
+def import_device_assets():
+    try:
+        result = get_device_asset_service().import_records(
+            request.get_json(silent=True) or {},
+            actor_uid=_identity().get("id") or _identity().get("sub"),
+        )
+        return jsonify(success(_device_asset_import_result(result))), 200
+    except Exception as error:
+        return _error(error)
+
+
+@bp.route("/device-assets/<asset_uid>", methods=["GET"])
+def get_device_asset(asset_uid):
+    try:
+        return jsonify(
+            success(
+                _device_asset_detail(
+                    get_device_asset_service().get(asset_uid)
+                )
+            )
+        ), 200
+    except Exception as error:
+        return _error(error)
+
+
+@bp.route("/device-assets/<asset_uid>/versions", methods=["GET"])
+def list_device_asset_versions(asset_uid):
+    try:
+        records = get_device_asset_service().versions(asset_uid)
+        return jsonify(
+            success(
+                {
+                    "records": [
+                        _device_asset_version(record)
+                        for record in records
+                    ],
+                    "total": len(records),
+                }
+            )
+        ), 200
+    except Exception as error:
+        return _error(error)
+
+
 @bp.route("/data-elements", methods=["POST"])
 def create_data_element():
     try:

app/core/data_research/device_asset_repository.py

@@ -0,0 +1,270 @@
+from __future__ import annotations
+
+from sqlalchemy import distinct, func, or_
+
+from app.core.data_research.device_assets import (
+    DeviceAssetMappingRecord,
+    DeviceAssetRecord,
+    DeviceAssetVersionRecord,
+)
+from app.models.data_research import (
+    DeviceAsset,
+    DeviceAssetSourceMapping,
+    DeviceAssetVersion,
+    IngestionSource,
+)
+
+
+class SqlAlchemyDeviceAssetRepository:
+    def __init__(self, session):
+        self.session = session
+
+    @staticmethod
+    def _asset_record(model):
+        return DeviceAssetRecord(
+            uid=str(model.uid),
+            asset_type=model.asset_type,
+            name=model.name,
+            status=model.status,
+            current_version=int(model.current_version),
+            content_hash=model.content_hash,
+            location=model.location,
+            organization=model.organization,
+            responsible_person=model.responsible_person,
+            attributes=dict(model.attributes or {}),
+            created_by=model.created_by,
+            updated_by=model.updated_by,
+            created_at=model.created_at,
+            updated_at=model.updated_at,
+        )
+
+    @staticmethod
+    def _mapping_record(model):
+        return DeviceAssetMappingRecord(
+            uid=str(model.uid),
+            asset_uid=str(model.asset_uid),
+            source_uid=str(model.source_uid),
+            source_entity=model.source_entity,
+            asset_type=model.asset_type,
+            source_code=model.source_code,
+            source_updated_at=model.source_updated_at,
+            first_seen_at=model.first_seen_at,
+            last_seen_at=model.last_seen_at,
+        )
+
+    @staticmethod
+    def _version_record(model):
+        return DeviceAssetVersionRecord(
+            uid=str(model.uid),
+            asset_uid=str(model.asset_uid),
+            version=int(model.version),
+            content_hash=model.content_hash,
+            snapshot=dict(model.snapshot or {}),
+            source_mapping_uid=str(model.source_mapping_uid),
+            actor_uid=model.actor_uid,
+            created_at=model.created_at,
+        )
+
+    def source_is_active(self, source_uid):
+        return (
+            self.session.query(IngestionSource.uid)
+            .filter_by(uid=str(source_uid), status="active")
+            .first()
+            is not None
+        )
+
+    def find_mapping(
+        self,
+        source_uid,
+        source_entity,
+        asset_type,
+        source_code,
+        *,
+        for_update=False,
+    ):
+        query = self.session.query(DeviceAssetSourceMapping).filter_by(
+            source_uid=str(source_uid),
+            source_entity=str(source_entity),
+            asset_type=str(asset_type),
+            source_code=str(source_code),
+        )
+        if for_update:
+            query = query.with_for_update()
+        model = query.first()
+        return self._mapping_record(model) if model is not None else None
+
+    def get(self, asset_uid):
+        model = self.session.get(DeviceAsset, str(asset_uid))
+        return self._asset_record(model) if model is not None else None
+
+    @staticmethod
+    def _asset_model(record):
+        return DeviceAsset(
+            uid=record.uid,
+            asset_type=record.asset_type,
+            name=record.name,
+            status=record.status,
+            current_version=record.current_version,
+            content_hash=record.content_hash,
+            location=record.location,
+            organization=record.organization,
+            responsible_person=record.responsible_person,
+            attributes=record.attributes,
+            created_by=record.created_by,
+            updated_by=record.updated_by,
+            created_at=record.created_at,
+            updated_at=record.updated_at,
+        )
+
+    @staticmethod
+    def _mapping_model(record):
+        return DeviceAssetSourceMapping(
+            uid=record.uid,
+            asset_uid=record.asset_uid,
+            source_uid=record.source_uid,
+            source_entity=record.source_entity,
+            asset_type=record.asset_type,
+            source_code=record.source_code,
+            source_updated_at=record.source_updated_at,
+            first_seen_at=record.first_seen_at,
+            last_seen_at=record.last_seen_at,
+        )
+
+    @staticmethod
+    def _version_model(record):
+        return DeviceAssetVersion(
+            uid=record.uid,
+            asset_uid=record.asset_uid,
+            version=record.version,
+            content_hash=record.content_hash,
+            snapshot=record.snapshot,
+            source_mapping_uid=record.source_mapping_uid,
+            actor_uid=record.actor_uid,
+            created_at=record.created_at,
+        )
+
+    def create_asset(self, asset, mapping, version):
+        asset_model = self._asset_model(asset)
+        mapping_model = self._mapping_model(mapping)
+        self.session.add(asset_model)
+        self.session.add(mapping_model)
+        self.session.add(self._version_model(version))
+        self.session.flush()
+        return (
+            self._asset_record(asset_model),
+            self._mapping_record(mapping_model),
+        )
+
+    def update_asset(self, asset, mapping, version):
+        asset_model = self.session.get(DeviceAsset, str(asset.uid))
+        mapping_model = self.session.get(
+            DeviceAssetSourceMapping,
+            str(mapping.uid),
+        )
+        for name in (
+            "name",
+            "status",
+            "current_version",
+            "content_hash",
+            "location",
+            "organization",
+            "responsible_person",
+            "attributes",
+            "updated_by",
+            "updated_at",
+        ):
+            setattr(asset_model, name, getattr(asset, name))
+        mapping_model.source_updated_at = mapping.source_updated_at
+        mapping_model.last_seen_at = mapping.last_seen_at
+        self.session.add(self._version_model(version))
+        self.session.flush()
+        return (
+            self._asset_record(asset_model),
+            self._mapping_record(mapping_model),
+        )
+
+    def touch_mapping(
+        self,
+        mapping,
+        *,
+        source_updated_at,
+        last_seen_at,
+    ):
+        model = self.session.get(
+            DeviceAssetSourceMapping,
+            str(mapping.uid),
+        )
+        model.source_updated_at = source_updated_at
+        model.last_seen_at = last_seen_at
+        self.session.flush()
+        return self._mapping_record(model)
+
+    @staticmethod
+    def _filtered(query, filters):
+        keyword = str(filters.get("keyword") or "").strip()
+        if keyword:
+            pattern = f"%{keyword}%"
+            query = query.filter(
+                or_(
+                    DeviceAsset.name.ilike(pattern),
+                    DeviceAsset.location.ilike(pattern),
+                    DeviceAsset.organization.ilike(pattern),
+                    DeviceAsset.responsible_person.ilike(pattern),
+                    DeviceAssetSourceMapping.source_code.ilike(pattern),
+                )
+            )
+        if filters.get("asset_type"):
+            query = query.filter(
+                DeviceAsset.asset_type == str(filters["asset_type"])
+            )
+        if filters.get("status"):
+            query = query.filter(
+                DeviceAsset.status == str(filters["status"])
+            )
+        if filters.get("source_uid"):
+            query = query.filter(
+                DeviceAssetSourceMapping.source_uid
+                == str(filters["source_uid"])
+            )
+        return query
+
+    def search(self, filters, *, page, page_size):
+        base = self.session.query(DeviceAsset).outerjoin(
+            DeviceAssetSourceMapping,
+            DeviceAssetSourceMapping.asset_uid == DeviceAsset.uid,
+        )
+        base = self._filtered(base, filters)
+        total = (
+            base.with_entities(func.count(distinct(DeviceAsset.uid)))
+            .scalar()
+            or 0
+        )
+        rows = (
+            base.distinct()
+            .order_by(DeviceAsset.updated_at.desc(), DeviceAsset.uid.asc())
+            .offset((int(page) - 1) * int(page_size))
+            .limit(int(page_size))
+            .all()
+        )
+        return [self._asset_record(row) for row in rows], int(total)
+
+    def list_mappings(self, asset_uid):
+        return [
+            self._mapping_record(model)
+            for model in self.session.query(DeviceAssetSourceMapping)
+            .filter_by(asset_uid=str(asset_uid))
+            .order_by(
+                DeviceAssetSourceMapping.source_entity.asc(),
+                DeviceAssetSourceMapping.source_code.asc(),
+            )
+            .all()
+        ]
+
+    def list_versions(self, asset_uid):
+        return [
+            self._version_record(model)
+            for model in self.session.query(DeviceAssetVersion)
+            .filter_by(asset_uid=str(asset_uid))
+            .order_by(DeviceAssetVersion.version.desc())
+            .all()
+        ]

app/core/data_research/device_assets.py

@@ -0,0 +1,440 @@
+from __future__ import annotations
+
+import hashlib
+import json
+from dataclasses import dataclass, replace
+from datetime import datetime
+from typing import Any, Callable
+from uuid import UUID
+
+from app.core.common.identifiers import new_governance_uid
+from app.core.common.timezone_utils import now_china
+from app.core.data_research.errors import (
+    DeviceAssetInvalid,
+    DeviceAssetNotFound,
+)
+
+
+ASSET_TYPES = frozenset(
+    {
+        "device",
+        "component",
+        "measurement_point",
+        "alarm",
+        "maintenance_record",
+    }
+)
+ASSET_STATUSES = frozenset({"active", "retired"})
+MAX_IMPORT_RECORDS = 500
+MAX_ATTRIBUTES_BYTES = 64_000
+SECRET_KEYS = frozenset(
+    {
+        "password",
+        "passwd",
+        "credential",
+        "credentials",
+        "encrypted_payload",
+        "api_key",
+        "token",
+        "authorization",
+        "connection_string",
+        "connection_url",
+        "conn_str",
+    }
+)
+
+
+@dataclass(frozen=True)
+class DeviceAssetRecord:
+    uid: str
+    asset_type: str
+    name: str
+    status: str
+    current_version: int
+    content_hash: str
+    location: str | None
+    organization: str | None
+    responsible_person: str | None
+    attributes: dict[str, Any]
+    created_by: str | None
+    updated_by: str | None
+    created_at: datetime | None = None
+    updated_at: datetime | None = None
+
+
+@dataclass(frozen=True)
+class DeviceAssetMappingRecord:
+    uid: str
+    asset_uid: str
+    source_uid: str
+    source_entity: str
+    asset_type: str
+    source_code: str
+    source_updated_at: datetime | None
+    first_seen_at: datetime | None = None
+    last_seen_at: datetime | None = None
+
+
+@dataclass(frozen=True)
+class DeviceAssetVersionRecord:
+    uid: str
+    asset_uid: str
+    version: int
+    content_hash: str
+    snapshot: dict[str, Any]
+    source_mapping_uid: str
+    actor_uid: str | None
+    created_at: datetime | None = None
+
+
+@dataclass(frozen=True)
+class DeviceAssetDetail:
+    asset: DeviceAssetRecord
+    mappings: tuple[DeviceAssetMappingRecord, ...]
+
+
+@dataclass(frozen=True)
+class DeviceAssetImportItem:
+    action: str
+    asset: DeviceAssetRecord
+    mapping: DeviceAssetMappingRecord
+
+
+@dataclass(frozen=True)
+class DeviceAssetImportResult:
+    items: tuple[DeviceAssetImportItem, ...]
+    created_count: int
+    updated_count: int
+    unchanged_count: int
+
+
+def _required_text(payload: dict[str, Any], name: str, *, maximum=300) -> str:
+    value = str(payload.get(name) or "").strip()
+    if not value:
+        raise DeviceAssetInvalid(f"{name} is required")
+    if len(value) > maximum:
+        raise DeviceAssetInvalid(f"{name} exceeds {maximum} characters")
+    return value
+
+
+def _optional_text(payload: dict[str, Any], name: str, *, maximum=300):
+    value = str(payload.get(name) or "").strip()
+    if not value:
+        return None
+    if len(value) > maximum:
+        raise DeviceAssetInvalid(f"{name} exceeds {maximum} characters")
+    return value
+
+
+def _source_time(value: Any) -> datetime | None:
+    if value in (None, ""):
+        return None
+    try:
+        return datetime.fromisoformat(str(value).replace("Z", "+00:00"))
+    except (TypeError, ValueError) as exc:
+        raise DeviceAssetInvalid("source_updated_at must be ISO-8601") from exc
+
+
+def _source_uid(payload: dict[str, Any]) -> str:
+    value = _required_text(payload, "source_uid", maximum=100)
+    try:
+        return str(UUID(value))
+    except (AttributeError, TypeError, ValueError) as exc:
+        raise DeviceAssetInvalid("source_uid must be a UUID") from exc
+
+
+def _search_filters(filters: Any) -> dict[str, str]:
+    if not isinstance(filters, dict):
+        raise DeviceAssetInvalid("filters must be an object")
+    normalized = {}
+    keyword = str(filters.get("keyword") or "").strip()
+    if len(keyword) > 300:
+        raise DeviceAssetInvalid("keyword exceeds 300 characters")
+    if keyword:
+        normalized["keyword"] = keyword
+    asset_type = str(filters.get("asset_type") or "").strip()
+    if asset_type and asset_type not in ASSET_TYPES:
+        raise DeviceAssetInvalid("asset_type is not supported")
+    if asset_type:
+        normalized["asset_type"] = asset_type
+    status = str(filters.get("status") or "").strip()
+    if status and status not in ASSET_STATUSES:
+        raise DeviceAssetInvalid("status is not supported")
+    if status:
+        normalized["status"] = status
+    if filters.get("source_uid"):
+        normalized["source_uid"] = _source_uid(filters)
+    return normalized
+
+
+def _reject_secrets(value: Any) -> None:
+    if isinstance(value, dict):
+        for key, item in value.items():
+            normalized = str(key).strip().lower()
+            if normalized in SECRET_KEYS:
+                raise DeviceAssetInvalid(
+                    "secret attributes are not allowed in device assets"
+                )
+            _reject_secrets(item)
+    elif isinstance(value, (list, tuple)):
+        for item in value:
+            _reject_secrets(item)
+
+
+def _normalize_record(raw: Any) -> tuple[dict[str, Any], datetime | None]:
+    if not isinstance(raw, dict):
+        raise DeviceAssetInvalid("each device asset record must be an object")
+    if "uid" in raw or "asset_uid" in raw:
+        raise DeviceAssetInvalid("uid is assigned by the platform")
+    asset_type = _required_text(raw, "asset_type", maximum=40)
+    if asset_type not in ASSET_TYPES:
+        raise DeviceAssetInvalid("asset_type is not supported")
+    status = str(raw.get("status") or "active").strip()
+    if status not in ASSET_STATUSES:
+        raise DeviceAssetInvalid("status is not supported")
+    attributes = raw.get("attributes") or {}
+    if not isinstance(attributes, dict):
+        raise DeviceAssetInvalid("attributes must be an object")
+    _reject_secrets(attributes)
+    serialized = json.dumps(
+        attributes,
+        ensure_ascii=False,
+        sort_keys=True,
+        separators=(",", ":"),
+    )
+    if len(serialized.encode("utf-8")) > MAX_ATTRIBUTES_BYTES:
+        raise DeviceAssetInvalid("attributes exceed the catalog boundary")
+    normalized_attributes = json.loads(serialized)
+    snapshot = {
+        "asset_type": asset_type,
+        "source_code": _required_text(raw, "source_code"),
+        "name": _required_text(raw, "name"),
+        "status": status,
+        "location": _optional_text(raw, "location"),
+        "organization": _optional_text(raw, "organization"),
+        "responsible_person": _optional_text(raw, "responsible_person"),
+        "attributes": normalized_attributes,
+    }
+    return snapshot, _source_time(raw.get("source_updated_at"))
+
+
+def _content_hash(snapshot: dict[str, Any]) -> str:
+    canonical = json.dumps(
+        snapshot,
+        ensure_ascii=False,
+        sort_keys=True,
+        separators=(",", ":"),
+    )
+    return hashlib.sha256(canonical.encode("utf-8")).hexdigest()
+
+
+class DeviceAssetService:
+    def __init__(
+        self,
+        repository,
+        *,
+        uid_factory: Callable[[], str] = new_governance_uid,
+        now_factory: Callable[[], datetime] = now_china,
+        commit: Callable[[], Any] = lambda: None,
+        rollback: Callable[[], Any] = lambda: None,
+    ):
+        self.repository = repository
+        self.uid_factory = uid_factory
+        self.now_factory = now_factory
+        self.commit = commit
+        self.rollback = rollback
+
+    def import_records(self, payload, actor_uid) -> DeviceAssetImportResult:
+        if not isinstance(payload, dict):
+            raise DeviceAssetInvalid("payload must be an object")
+        source_uid = _source_uid(payload)
+        source_entity = _required_text(
+            payload,
+            "source_entity",
+            maximum=300,
+        )
+        if not self.repository.source_is_active(source_uid):
+            raise DeviceAssetInvalid("source_uid must reference an active source")
+        raw_records = payload.get("records")
+        if not isinstance(raw_records, list) or not raw_records:
+            raise DeviceAssetInvalid("records must be a non-empty array")
+        if len(raw_records) > MAX_IMPORT_RECORDS:
+            raise DeviceAssetInvalid("records cannot exceed 500 items")
+
+        normalized_records = []
+        identities = set()
+        for raw in raw_records:
+            snapshot, source_updated_at = _normalize_record(raw)
+            identity = (
+                source_uid,
+                source_entity,
+                snapshot["asset_type"],
+                snapshot["source_code"],
+            )
+            if identity in identities:
+                raise DeviceAssetInvalid(
+                    "duplicate source identity in import request"
+                )
+            identities.add(identity)
+            normalized_records.append((snapshot, source_updated_at))
+
+        items = []
+        try:
+            for snapshot, source_updated_at in normalized_records:
+                items.append(
+                    self._import_one(
+                        source_uid,
+                        source_entity,
+                        snapshot,
+                        source_updated_at,
+                        actor_uid,
+                    )
+                )
+            self.commit()
+        except Exception:
+            self.rollback()
+            raise
+        actions = [item.action for item in items]
+        return DeviceAssetImportResult(
+            items=tuple(items),
+            created_count=actions.count("created"),
+            updated_count=actions.count("updated"),
+            unchanged_count=actions.count("unchanged"),
+        )
+
+    def _import_one(
+        self,
+        source_uid,
+        source_entity,
+        snapshot,
+        source_updated_at,
+        actor_uid,
+    ):
+        now = self.now_factory()
+        digest = _content_hash(snapshot)
+        mapping = self.repository.find_mapping(
+            source_uid,
+            source_entity,
+            snapshot["asset_type"],
+            snapshot["source_code"],
+            for_update=True,
+        )
+        if mapping is None:
+            asset = DeviceAssetRecord(
+                uid=self.uid_factory(),
+                asset_type=snapshot["asset_type"],
+                name=snapshot["name"],
+                status=snapshot["status"],
+                current_version=1,
+                content_hash=digest,
+                location=snapshot["location"],
+                organization=snapshot["organization"],
+                responsible_person=snapshot["responsible_person"],
+                attributes=dict(snapshot["attributes"]),
+                created_by=actor_uid,
+                updated_by=actor_uid,
+                created_at=now,
+                updated_at=now,
+            )
+            mapping = DeviceAssetMappingRecord(
+                uid=self.uid_factory(),
+                asset_uid=asset.uid,
+                source_uid=source_uid,
+                source_entity=source_entity,
+                asset_type=asset.asset_type,
+                source_code=snapshot["source_code"],
+                source_updated_at=source_updated_at,
+                first_seen_at=now,
+                last_seen_at=now,
+            )
+            version = DeviceAssetVersionRecord(
+                uid=self.uid_factory(),
+                asset_uid=asset.uid,
+                version=1,
+                content_hash=digest,
+                snapshot=dict(snapshot),
+                source_mapping_uid=mapping.uid,
+                actor_uid=actor_uid,
+                created_at=now,
+            )
+            asset, mapping = self.repository.create_asset(
+                asset,
+                mapping,
+                version,
+            )
+            return DeviceAssetImportItem("created", asset, mapping)
+
+        asset = self.repository.get(mapping.asset_uid)
+        if asset is None:
+            raise DeviceAssetInvalid("source mapping references a missing asset")
+        if asset.content_hash == digest:
+            mapping = self.repository.touch_mapping(
+                mapping,
+                source_updated_at=source_updated_at,
+                last_seen_at=now,
+            )
+            return DeviceAssetImportItem("unchanged", asset, mapping)
+
+        asset = replace(
+            asset,
+            name=snapshot["name"],
+            status=snapshot["status"],
+            current_version=asset.current_version + 1,
+            content_hash=digest,
+            location=snapshot["location"],
+            organization=snapshot["organization"],
+            responsible_person=snapshot["responsible_person"],
+            attributes=dict(snapshot["attributes"]),
+            updated_by=actor_uid,
+            updated_at=now,
+        )
+        mapping = replace(
+            mapping,
+            source_updated_at=source_updated_at,
+            last_seen_at=now,
+        )
+        version = DeviceAssetVersionRecord(
+            uid=self.uid_factory(),
+            asset_uid=asset.uid,
+            version=asset.current_version,
+            content_hash=digest,
+            snapshot=dict(snapshot),
+            source_mapping_uid=mapping.uid,
+            actor_uid=actor_uid,
+            created_at=now,
+        )
+        asset, mapping = self.repository.update_asset(
+            asset,
+            mapping,
+            version,
+        )
+        return DeviceAssetImportItem("updated", asset, mapping)
+
+    def search(self, filters, *, page=1, page_size=20):
+        assets, total = self.repository.search(
+            _search_filters(filters or {}),
+            page=int(page),
+            page_size=int(page_size),
+        )
+        return [
+            DeviceAssetDetail(
+                asset=asset,
+                mappings=tuple(self.repository.list_mappings(asset.uid)),
+            )
+            for asset in assets
+        ], total
+
+    def get(self, asset_uid) -> DeviceAssetDetail:
+        asset = self.repository.get(str(asset_uid))
+        if asset is None:
+            raise DeviceAssetNotFound(
+                f"device asset {asset_uid} was not found"
+            )
+        return DeviceAssetDetail(
+            asset=asset,
+            mappings=tuple(self.repository.list_mappings(asset.uid)),
+        )
+
+    def versions(self, asset_uid):
+        self.get(asset_uid)
+        return self.repository.list_versions(str(asset_uid))

app/core/data_research/errors.py

@@ -48,3 +48,13 @@ class CandidateDecisionInvalid(DataResearchError):
 class DataElementProjectionInvalid(DataResearchError):
     code = "DATA_ELEMENT_PROJECTION_INVALID"
     http_status = 422
+
+
+class DeviceAssetInvalid(DataResearchError):
+    code = "DEVICE_ASSET_INVALID"
+    http_status = 422
+
+
+class DeviceAssetNotFound(DataResearchError):
+    code = "DEVICE_ASSET_NOT_FOUND"
+    http_status = 404

app/core/system/permissions.py

@@ -33,6 +33,7 @@ DATA_ELEMENTS_EDIT = "data-elements:edit"
 DATA_ELEMENTS_PUBLISH = "data-elements:publish"
 ONTOLOGIES_EDIT = "ontologies:edit"
 ONTOLOGIES_PUBLISH = "ontologies:publish"
+DEVICE_ASSETS_EDIT = "device-assets:edit"
 
 ROLE_PERMISSIONS = {
     "viewer": frozenset(
@@ -51,6 +52,7 @@ ROLE_PERMISSIONS = {
             INGESTION_RUN,
             DATA_ELEMENTS_EDIT,
             ONTOLOGIES_EDIT,
+            DEVICE_ASSETS_EDIT,
         }
     ),
     "admin": frozenset(
@@ -81,6 +83,7 @@ ROLE_PERMISSIONS = {
             DATA_ELEMENTS_PUBLISH,
             ONTOLOGIES_EDIT,
             ONTOLOGIES_PUBLISH,
+            DEVICE_ASSETS_EDIT,
         }
     ),
 }
@@ -141,6 +144,10 @@ def permission_for_request(path: str, method: str) -> tuple[str, ...]:
         return (KNOWLEDGE_MANAGE,)
     if path.startswith("/api/system/users"):
         return (MANAGE_USERS,)
+    if path.startswith("/api/development/v1/device-assets"):
+        if method == "GET":
+            return (READ_GOVERNANCE,)
+        return (DEVICE_ASSETS_EDIT,)
     if path.startswith("/api/development/v1/ingestion-jobs"):
         if method == "GET":
             return (READ_GOVERNANCE,)

app/models/data_research.py

@@ -6,7 +6,7 @@ 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
+from app.core.common.timezone_utils import now_china, now_china_naive
 
 
 JOB_STATUSES = (
@@ -222,6 +222,150 @@ class CatalogSnapshot(db.Model):
         }
 
 
+class DeviceAsset(db.Model):
+    __tablename__ = "device_assets"
+    __table_args__ = (
+        db.CheckConstraint(
+            "asset_type IN ("
+            "'device','component','measurement_point',"
+            "'alarm','maintenance_record'"
+            ")",
+            name="ck_device_asset_type",
+        ),
+        db.CheckConstraint(
+            "status IN ('active','retired')",
+            name="ck_device_asset_status",
+        ),
+        db.CheckConstraint(
+            "current_version > 0",
+            name="ck_device_asset_current_version",
+        ),
+        {"schema": "public"},
+    )
+
+    uid = db.Column(
+        UUID(as_uuid=False),
+        primary_key=True,
+        default=new_governance_uid,
+    )
+    asset_type = db.Column(db.String(40), nullable=False)
+    name = db.Column(db.String(300), nullable=False)
+    status = db.Column(db.String(20), nullable=False, default="active")
+    current_version = db.Column(db.Integer, nullable=False, default=1)
+    content_hash = db.Column(db.String(64), nullable=False)
+    location = db.Column(db.String(300))
+    organization = db.Column(db.String(300))
+    responsible_person = db.Column(db.String(300))
+    attributes = db.Column(JSONB, nullable=False, default=dict)
+    created_by = db.Column(db.String(100))
+    updated_by = db.Column(db.String(100))
+    created_at = db.Column(
+        db.DateTime(timezone=True),
+        nullable=False,
+        default=now_china,
+    )
+    updated_at = db.Column(
+        db.DateTime(timezone=True),
+        nullable=False,
+        default=now_china,
+    )
+
+
+class DeviceAssetSourceMapping(db.Model):
+    __tablename__ = "device_asset_source_mappings"
+    __table_args__ = (
+        db.CheckConstraint(
+            "asset_type IN ("
+            "'device','component','measurement_point',"
+            "'alarm','maintenance_record'"
+            ")",
+            name="ck_device_asset_mapping_type",
+        ),
+        db.UniqueConstraint(
+            "source_uid",
+            "source_entity",
+            "asset_type",
+            "source_code",
+            name="uq_device_asset_source_identity",
+        ),
+        {"schema": "public"},
+    )
+
+    uid = db.Column(
+        UUID(as_uuid=False),
+        primary_key=True,
+        default=new_governance_uid,
+    )
+    asset_uid = db.Column(
+        UUID(as_uuid=False),
+        db.ForeignKey("public.device_assets.uid", ondelete="CASCADE"),
+        nullable=False,
+    )
+    source_uid = db.Column(
+        UUID(as_uuid=False),
+        db.ForeignKey("public.ingestion_sources.uid"),
+        nullable=False,
+    )
+    source_entity = db.Column(db.String(300), nullable=False)
+    asset_type = db.Column(db.String(40), nullable=False)
+    source_code = db.Column(db.String(300), nullable=False)
+    source_updated_at = db.Column(db.DateTime(timezone=True))
+    first_seen_at = db.Column(
+        db.DateTime(timezone=True),
+        nullable=False,
+        default=now_china,
+    )
+    last_seen_at = db.Column(
+        db.DateTime(timezone=True),
+        nullable=False,
+        default=now_china,
+    )
+
+
+class DeviceAssetVersion(db.Model):
+    __tablename__ = "device_asset_versions"
+    __table_args__ = (
+        db.CheckConstraint(
+            "version > 0",
+            name="ck_device_asset_version_number",
+        ),
+        db.UniqueConstraint(
+            "asset_uid",
+            "version",
+            name="uq_device_asset_version",
+        ),
+        {"schema": "public"},
+    )
+
+    uid = db.Column(
+        UUID(as_uuid=False),
+        primary_key=True,
+        default=new_governance_uid,
+    )
+    asset_uid = db.Column(
+        UUID(as_uuid=False),
+        db.ForeignKey("public.device_assets.uid", ondelete="CASCADE"),
+        nullable=False,
+    )
+    version = db.Column(db.Integer, nullable=False)
+    content_hash = db.Column(db.String(64), nullable=False)
+    snapshot = db.Column(JSONB, nullable=False)
+    source_mapping_uid = db.Column(
+        UUID(as_uuid=False),
+        db.ForeignKey(
+            "public.device_asset_source_mappings.uid",
+            ondelete="RESTRICT",
+        ),
+        nullable=False,
+    )
+    actor_uid = db.Column(db.String(100))
+    created_at = db.Column(
+        db.DateTime(timezone=True),
+        nullable=False,
+        default=now_china,
+    )
+
+
 class EvidenceFragment(db.Model):
     __tablename__ = "evidence_fragments"
     __table_args__ = (

deployment/app/api/data_development/routes.py

@@ -101,6 +101,19 @@ def get_data_element_service():
     )
 
 
+def get_device_asset_service():
+    from app.core.data_research.device_asset_repository import (
+        SqlAlchemyDeviceAssetRepository,
+    )
+    from app.core.data_research.device_assets import DeviceAssetService
+
+    return DeviceAssetService(
+        SqlAlchemyDeviceAssetRepository(db.session),
+        commit=db.session.commit,
+        rollback=db.session.rollback,
+    )
+
+
 def get_candidate_decision_service():
     from app.core.data_research.candidate_decisions import CandidateDecisionService
     from app.core.data_research.data_elements import DataElementService
@@ -316,6 +329,94 @@ def _ontology_version(record):
     }
 
 
+def _iso(value):
+    return value.isoformat() if value else None
+
+
+def _device_asset(record):
+    return {
+        "uid": str(record.uid),
+        "asset_type": record.asset_type,
+        "name": record.name,
+        "status": record.status,
+        "current_version": int(record.current_version),
+        "location": record.location,
+        "organization": record.organization,
+        "responsible_person": record.responsible_person,
+        "attributes": dict(record.attributes or {}),
+        "created_by": record.created_by,
+        "updated_by": record.updated_by,
+        "created_at": _iso(record.created_at),
+        "updated_at": _iso(record.updated_at),
+    }
+
+
+def _device_asset_mapping(record):
+    return {
+        "uid": str(record.uid),
+        "asset_uid": str(record.asset_uid),
+        "source_uid": str(record.source_uid),
+        "source_entity": record.source_entity,
+        "asset_type": record.asset_type,
+        "source_code": record.source_code,
+        "source_updated_at": _iso(record.source_updated_at),
+        "first_seen_at": _iso(record.first_seen_at),
+        "last_seen_at": _iso(record.last_seen_at),
+    }
+
+
+def _device_asset_detail(detail):
+    data = _device_asset(detail.asset)
+    data["source_mappings"] = [
+        _device_asset_mapping(mapping)
+        for mapping in detail.mappings
+    ]
+    return data
+
+
+def _device_asset_version(record):
+    return {
+        "uid": str(record.uid),
+        "asset_uid": str(record.asset_uid),
+        "version": int(record.version),
+        "snapshot": dict(record.snapshot or {}),
+        "source_mapping_uid": str(record.source_mapping_uid),
+        "actor_uid": record.actor_uid,
+        "created_at": _iso(record.created_at),
+    }
+
+
+def _device_asset_import_result(result):
+    return {
+        "records": [
+            {
+                "action": item.action,
+                "asset": _device_asset(item.asset),
+                "source_mapping": _device_asset_mapping(item.mapping),
+            }
+            for item in result.items
+        ],
+        "created_count": int(result.created_count),
+        "updated_count": int(result.updated_count),
+        "unchanged_count": int(result.unchanged_count),
+    }
+
+
+def _device_asset_page(name, *, default, maximum):
+    from app.core.data_research.errors import DeviceAssetInvalid
+
+    raw = request.args.get(name)
+    try:
+        value = default if raw in (None, "") else int(raw)
+    except (TypeError, ValueError) as error:
+        raise DeviceAssetInvalid(f"{name} must be an integer") from error
+    if value < 1 or value > maximum:
+        raise DeviceAssetInvalid(
+            f"{name} must be between 1 and {maximum}"
+        )
+    return value
+
+
 def _error(error):
     if isinstance(error, DataResearchError):
         return (
@@ -468,6 +569,91 @@ def list_data_elements():
         return _error(error)
 
 
+@bp.route("/device-assets", methods=["GET"])
+def list_device_assets():
+    filters = {
+        name: request.args.get(name)
+        for name in ("keyword", "asset_type", "status", "source_uid")
+        if request.args.get(name)
+    }
+    try:
+        page = _device_asset_page(
+            "page",
+            default=1,
+            maximum=1_000_000,
+        )
+        page_size = _device_asset_page(
+            "page_size",
+            default=20,
+            maximum=100,
+        )
+        records, total = get_device_asset_service().search(
+            filters,
+            page=page,
+            page_size=page_size,
+        )
+        return jsonify(
+            success(
+                {
+                    "records": [
+                        _device_asset_detail(record)
+                        for record in records
+                    ],
+                    "total": int(total),
+                    "page": page,
+                    "page_size": page_size,
+                }
+            )
+        ), 200
+    except Exception as error:
+        return _error(error)
+
+
+@bp.route("/device-assets/import", methods=["POST"])
+def import_device_assets():
+    try:
+        result = get_device_asset_service().import_records(
+            request.get_json(silent=True) or {},
+            actor_uid=_identity().get("id") or _identity().get("sub"),
+        )
+        return jsonify(success(_device_asset_import_result(result))), 200
+    except Exception as error:
+        return _error(error)
+
+
+@bp.route("/device-assets/<asset_uid>", methods=["GET"])
+def get_device_asset(asset_uid):
+    try:
+        return jsonify(
+            success(
+                _device_asset_detail(
+                    get_device_asset_service().get(asset_uid)
+                )
+            )
+        ), 200
+    except Exception as error:
+        return _error(error)
+
+
+@bp.route("/device-assets/<asset_uid>/versions", methods=["GET"])
+def list_device_asset_versions(asset_uid):
+    try:
+        records = get_device_asset_service().versions(asset_uid)
+        return jsonify(
+            success(
+                {
+                    "records": [
+                        _device_asset_version(record)
+                        for record in records
+                    ],
+                    "total": len(records),
+                }
+            )
+        ), 200
+    except Exception as error:
+        return _error(error)
+
+
 @bp.route("/data-elements", methods=["POST"])
 def create_data_element():
     try:

deployment/app/core/data_research/device_asset_repository.py

@@ -0,0 +1,270 @@
+from __future__ import annotations
+
+from sqlalchemy import distinct, func, or_
+
+from app.core.data_research.device_assets import (
+    DeviceAssetMappingRecord,
+    DeviceAssetRecord,
+    DeviceAssetVersionRecord,
+)
+from app.models.data_research import (
+    DeviceAsset,
+    DeviceAssetSourceMapping,
+    DeviceAssetVersion,
+    IngestionSource,
+)
+
+
+class SqlAlchemyDeviceAssetRepository:
+    def __init__(self, session):
+        self.session = session
+
+    @staticmethod
+    def _asset_record(model):
+        return DeviceAssetRecord(
+            uid=str(model.uid),
+            asset_type=model.asset_type,
+            name=model.name,
+            status=model.status,
+            current_version=int(model.current_version),
+            content_hash=model.content_hash,
+            location=model.location,
+            organization=model.organization,
+            responsible_person=model.responsible_person,
+            attributes=dict(model.attributes or {}),
+            created_by=model.created_by,
+            updated_by=model.updated_by,
+            created_at=model.created_at,
+            updated_at=model.updated_at,
+        )
+
+    @staticmethod
+    def _mapping_record(model):
+        return DeviceAssetMappingRecord(
+            uid=str(model.uid),
+            asset_uid=str(model.asset_uid),
+            source_uid=str(model.source_uid),
+            source_entity=model.source_entity,
+            asset_type=model.asset_type,
+            source_code=model.source_code,
+            source_updated_at=model.source_updated_at,
+            first_seen_at=model.first_seen_at,
+            last_seen_at=model.last_seen_at,
+        )
+
+    @staticmethod
+    def _version_record(model):
+        return DeviceAssetVersionRecord(
+            uid=str(model.uid),
+            asset_uid=str(model.asset_uid),
+            version=int(model.version),
+            content_hash=model.content_hash,
+            snapshot=dict(model.snapshot or {}),
+            source_mapping_uid=str(model.source_mapping_uid),
+            actor_uid=model.actor_uid,
+            created_at=model.created_at,
+        )
+
+    def source_is_active(self, source_uid):
+        return (
+            self.session.query(IngestionSource.uid)
+            .filter_by(uid=str(source_uid), status="active")
+            .first()
+            is not None
+        )
+
+    def find_mapping(
+        self,
+        source_uid,
+        source_entity,
+        asset_type,
+        source_code,
+        *,
+        for_update=False,
+    ):
+        query = self.session.query(DeviceAssetSourceMapping).filter_by(
+            source_uid=str(source_uid),
+            source_entity=str(source_entity),
+            asset_type=str(asset_type),
+            source_code=str(source_code),
+        )
+        if for_update:
+            query = query.with_for_update()
+        model = query.first()
+        return self._mapping_record(model) if model is not None else None
+
+    def get(self, asset_uid):
+        model = self.session.get(DeviceAsset, str(asset_uid))
+        return self._asset_record(model) if model is not None else None
+
+    @staticmethod
+    def _asset_model(record):
+        return DeviceAsset(
+            uid=record.uid,
+            asset_type=record.asset_type,
+            name=record.name,
+            status=record.status,
+            current_version=record.current_version,
+            content_hash=record.content_hash,
+            location=record.location,
+            organization=record.organization,
+            responsible_person=record.responsible_person,
+            attributes=record.attributes,
+            created_by=record.created_by,
+            updated_by=record.updated_by,
+            created_at=record.created_at,
+            updated_at=record.updated_at,
+        )
+
+    @staticmethod
+    def _mapping_model(record):
+        return DeviceAssetSourceMapping(
+            uid=record.uid,
+            asset_uid=record.asset_uid,
+            source_uid=record.source_uid,
+            source_entity=record.source_entity,
+            asset_type=record.asset_type,
+            source_code=record.source_code,
+            source_updated_at=record.source_updated_at,
+            first_seen_at=record.first_seen_at,
+            last_seen_at=record.last_seen_at,
+        )
+
+    @staticmethod
+    def _version_model(record):
+        return DeviceAssetVersion(
+            uid=record.uid,
+            asset_uid=record.asset_uid,
+            version=record.version,
+            content_hash=record.content_hash,
+            snapshot=record.snapshot,
+            source_mapping_uid=record.source_mapping_uid,
+            actor_uid=record.actor_uid,
+            created_at=record.created_at,
+        )
+
+    def create_asset(self, asset, mapping, version):
+        asset_model = self._asset_model(asset)
+        mapping_model = self._mapping_model(mapping)
+        self.session.add(asset_model)
+        self.session.add(mapping_model)
+        self.session.add(self._version_model(version))
+        self.session.flush()
+        return (
+            self._asset_record(asset_model),
+            self._mapping_record(mapping_model),
+        )
+
+    def update_asset(self, asset, mapping, version):
+        asset_model = self.session.get(DeviceAsset, str(asset.uid))
+        mapping_model = self.session.get(
+            DeviceAssetSourceMapping,
+            str(mapping.uid),
+        )
+        for name in (
+            "name",
+            "status",
+            "current_version",
+            "content_hash",
+            "location",
+            "organization",
+            "responsible_person",
+            "attributes",
+            "updated_by",
+            "updated_at",
+        ):
+            setattr(asset_model, name, getattr(asset, name))
+        mapping_model.source_updated_at = mapping.source_updated_at
+        mapping_model.last_seen_at = mapping.last_seen_at
+        self.session.add(self._version_model(version))
+        self.session.flush()
+        return (
+            self._asset_record(asset_model),
+            self._mapping_record(mapping_model),
+        )
+
+    def touch_mapping(
+        self,
+        mapping,
+        *,
+        source_updated_at,
+        last_seen_at,
+    ):
+        model = self.session.get(
+            DeviceAssetSourceMapping,
+            str(mapping.uid),
+        )
+        model.source_updated_at = source_updated_at
+        model.last_seen_at = last_seen_at
+        self.session.flush()
+        return self._mapping_record(model)
+
+    @staticmethod
+    def _filtered(query, filters):
+        keyword = str(filters.get("keyword") or "").strip()
+        if keyword:
+            pattern = f"%{keyword}%"
+            query = query.filter(
+                or_(
+                    DeviceAsset.name.ilike(pattern),
+                    DeviceAsset.location.ilike(pattern),
+                    DeviceAsset.organization.ilike(pattern),
+                    DeviceAsset.responsible_person.ilike(pattern),
+                    DeviceAssetSourceMapping.source_code.ilike(pattern),
+                )
+            )
+        if filters.get("asset_type"):
+            query = query.filter(
+                DeviceAsset.asset_type == str(filters["asset_type"])
+            )
+        if filters.get("status"):
+            query = query.filter(
+                DeviceAsset.status == str(filters["status"])
+            )
+        if filters.get("source_uid"):
+            query = query.filter(
+                DeviceAssetSourceMapping.source_uid
+                == str(filters["source_uid"])
+            )
+        return query
+
+    def search(self, filters, *, page, page_size):
+        base = self.session.query(DeviceAsset).outerjoin(
+            DeviceAssetSourceMapping,
+            DeviceAssetSourceMapping.asset_uid == DeviceAsset.uid,
+        )
+        base = self._filtered(base, filters)
+        total = (
+            base.with_entities(func.count(distinct(DeviceAsset.uid)))
+            .scalar()
+            or 0
+        )
+        rows = (
+            base.distinct()
+            .order_by(DeviceAsset.updated_at.desc(), DeviceAsset.uid.asc())
+            .offset((int(page) - 1) * int(page_size))
+            .limit(int(page_size))
+            .all()
+        )
+        return [self._asset_record(row) for row in rows], int(total)
+
+    def list_mappings(self, asset_uid):
+        return [
+            self._mapping_record(model)
+            for model in self.session.query(DeviceAssetSourceMapping)
+            .filter_by(asset_uid=str(asset_uid))
+            .order_by(
+                DeviceAssetSourceMapping.source_entity.asc(),
+                DeviceAssetSourceMapping.source_code.asc(),
+            )
+            .all()
+        ]
+
+    def list_versions(self, asset_uid):
+        return [
+            self._version_record(model)
+            for model in self.session.query(DeviceAssetVersion)
+            .filter_by(asset_uid=str(asset_uid))
+            .order_by(DeviceAssetVersion.version.desc())
+            .all()
+        ]

deployment/app/core/data_research/device_assets.py

@@ -0,0 +1,440 @@
+from __future__ import annotations
+
+import hashlib
+import json
+from dataclasses import dataclass, replace
+from datetime import datetime
+from typing import Any, Callable
+from uuid import UUID
+
+from app.core.common.identifiers import new_governance_uid
+from app.core.common.timezone_utils import now_china
+from app.core.data_research.errors import (
+    DeviceAssetInvalid,
+    DeviceAssetNotFound,
+)
+
+
+ASSET_TYPES = frozenset(
+    {
+        "device",
+        "component",
+        "measurement_point",
+        "alarm",
+        "maintenance_record",
+    }
+)
+ASSET_STATUSES = frozenset({"active", "retired"})
+MAX_IMPORT_RECORDS = 500
+MAX_ATTRIBUTES_BYTES = 64_000
+SECRET_KEYS = frozenset(
+    {
+        "password",
+        "passwd",
+        "credential",
+        "credentials",
+        "encrypted_payload",
+        "api_key",
+        "token",
+        "authorization",
+        "connection_string",
+        "connection_url",
+        "conn_str",
+    }
+)
+
+
+@dataclass(frozen=True)
+class DeviceAssetRecord:
+    uid: str
+    asset_type: str
+    name: str
+    status: str
+    current_version: int
+    content_hash: str
+    location: str | None
+    organization: str | None
+    responsible_person: str | None
+    attributes: dict[str, Any]
+    created_by: str | None
+    updated_by: str | None
+    created_at: datetime | None = None
+    updated_at: datetime | None = None
+
+
+@dataclass(frozen=True)
+class DeviceAssetMappingRecord:
+    uid: str
+    asset_uid: str
+    source_uid: str
+    source_entity: str
+    asset_type: str
+    source_code: str
+    source_updated_at: datetime | None
+    first_seen_at: datetime | None = None
+    last_seen_at: datetime | None = None
+
+
+@dataclass(frozen=True)
+class DeviceAssetVersionRecord:
+    uid: str
+    asset_uid: str
+    version: int
+    content_hash: str
+    snapshot: dict[str, Any]
+    source_mapping_uid: str
+    actor_uid: str | None
+    created_at: datetime | None = None
+
+
+@dataclass(frozen=True)
+class DeviceAssetDetail:
+    asset: DeviceAssetRecord
+    mappings: tuple[DeviceAssetMappingRecord, ...]
+
+
+@dataclass(frozen=True)
+class DeviceAssetImportItem:
+    action: str
+    asset: DeviceAssetRecord
+    mapping: DeviceAssetMappingRecord
+
+
+@dataclass(frozen=True)
+class DeviceAssetImportResult:
+    items: tuple[DeviceAssetImportItem, ...]
+    created_count: int
+    updated_count: int
+    unchanged_count: int
+
+
+def _required_text(payload: dict[str, Any], name: str, *, maximum=300) -> str:
+    value = str(payload.get(name) or "").strip()
+    if not value:
+        raise DeviceAssetInvalid(f"{name} is required")
+    if len(value) > maximum:
+        raise DeviceAssetInvalid(f"{name} exceeds {maximum} characters")
+    return value
+
+
+def _optional_text(payload: dict[str, Any], name: str, *, maximum=300):
+    value = str(payload.get(name) or "").strip()
+    if not value:
+        return None
+    if len(value) > maximum:
+        raise DeviceAssetInvalid(f"{name} exceeds {maximum} characters")
+    return value
+
+
+def _source_time(value: Any) -> datetime | None:
+    if value in (None, ""):
+        return None
+    try:
+        return datetime.fromisoformat(str(value).replace("Z", "+00:00"))
+    except (TypeError, ValueError) as exc:
+        raise DeviceAssetInvalid("source_updated_at must be ISO-8601") from exc
+
+
+def _source_uid(payload: dict[str, Any]) -> str:
+    value = _required_text(payload, "source_uid", maximum=100)
+    try:
+        return str(UUID(value))
+    except (AttributeError, TypeError, ValueError) as exc:
+        raise DeviceAssetInvalid("source_uid must be a UUID") from exc
+
+
+def _search_filters(filters: Any) -> dict[str, str]:
+    if not isinstance(filters, dict):
+        raise DeviceAssetInvalid("filters must be an object")
+    normalized = {}
+    keyword = str(filters.get("keyword") or "").strip()
+    if len(keyword) > 300:
+        raise DeviceAssetInvalid("keyword exceeds 300 characters")
+    if keyword:
+        normalized["keyword"] = keyword
+    asset_type = str(filters.get("asset_type") or "").strip()
+    if asset_type and asset_type not in ASSET_TYPES:
+        raise DeviceAssetInvalid("asset_type is not supported")
+    if asset_type:
+        normalized["asset_type"] = asset_type
+    status = str(filters.get("status") or "").strip()
+    if status and status not in ASSET_STATUSES:
+        raise DeviceAssetInvalid("status is not supported")
+    if status:
+        normalized["status"] = status
+    if filters.get("source_uid"):
+        normalized["source_uid"] = _source_uid(filters)
+    return normalized
+
+
+def _reject_secrets(value: Any) -> None:
+    if isinstance(value, dict):
+        for key, item in value.items():
+            normalized = str(key).strip().lower()
+            if normalized in SECRET_KEYS:
+                raise DeviceAssetInvalid(
+                    "secret attributes are not allowed in device assets"
+                )
+            _reject_secrets(item)
+    elif isinstance(value, (list, tuple)):
+        for item in value:
+            _reject_secrets(item)
+
+
+def _normalize_record(raw: Any) -> tuple[dict[str, Any], datetime | None]:
+    if not isinstance(raw, dict):
+        raise DeviceAssetInvalid("each device asset record must be an object")
+    if "uid" in raw or "asset_uid" in raw:
+        raise DeviceAssetInvalid("uid is assigned by the platform")
+    asset_type = _required_text(raw, "asset_type", maximum=40)
+    if asset_type not in ASSET_TYPES:
+        raise DeviceAssetInvalid("asset_type is not supported")
+    status = str(raw.get("status") or "active").strip()
+    if status not in ASSET_STATUSES:
+        raise DeviceAssetInvalid("status is not supported")
+    attributes = raw.get("attributes") or {}
+    if not isinstance(attributes, dict):
+        raise DeviceAssetInvalid("attributes must be an object")
+    _reject_secrets(attributes)
+    serialized = json.dumps(
+        attributes,
+        ensure_ascii=False,
+        sort_keys=True,
+        separators=(",", ":"),
+    )
+    if len(serialized.encode("utf-8")) > MAX_ATTRIBUTES_BYTES:
+        raise DeviceAssetInvalid("attributes exceed the catalog boundary")
+    normalized_attributes = json.loads(serialized)
+    snapshot = {
+        "asset_type": asset_type,
+        "source_code": _required_text(raw, "source_code"),
+        "name": _required_text(raw, "name"),
+        "status": status,
+        "location": _optional_text(raw, "location"),
+        "organization": _optional_text(raw, "organization"),
+        "responsible_person": _optional_text(raw, "responsible_person"),
+        "attributes": normalized_attributes,
+    }
+    return snapshot, _source_time(raw.get("source_updated_at"))
+
+
+def _content_hash(snapshot: dict[str, Any]) -> str:
+    canonical = json.dumps(
+        snapshot,
+        ensure_ascii=False,
+        sort_keys=True,
+        separators=(",", ":"),
+    )
+    return hashlib.sha256(canonical.encode("utf-8")).hexdigest()
+
+
+class DeviceAssetService:
+    def __init__(
+        self,
+        repository,
+        *,
+        uid_factory: Callable[[], str] = new_governance_uid,
+        now_factory: Callable[[], datetime] = now_china,
+        commit: Callable[[], Any] = lambda: None,
+        rollback: Callable[[], Any] = lambda: None,
+    ):
+        self.repository = repository
+        self.uid_factory = uid_factory
+        self.now_factory = now_factory
+        self.commit = commit
+        self.rollback = rollback
+
+    def import_records(self, payload, actor_uid) -> DeviceAssetImportResult:
+        if not isinstance(payload, dict):
+            raise DeviceAssetInvalid("payload must be an object")
+        source_uid = _source_uid(payload)
+        source_entity = _required_text(
+            payload,
+            "source_entity",
+            maximum=300,
+        )
+        if not self.repository.source_is_active(source_uid):
+            raise DeviceAssetInvalid("source_uid must reference an active source")
+        raw_records = payload.get("records")
+        if not isinstance(raw_records, list) or not raw_records:
+            raise DeviceAssetInvalid("records must be a non-empty array")
+        if len(raw_records) > MAX_IMPORT_RECORDS:
+            raise DeviceAssetInvalid("records cannot exceed 500 items")
+
+        normalized_records = []
+        identities = set()
+        for raw in raw_records:
+            snapshot, source_updated_at = _normalize_record(raw)
+            identity = (
+                source_uid,
+                source_entity,
+                snapshot["asset_type"],
+                snapshot["source_code"],
+            )
+            if identity in identities:
+                raise DeviceAssetInvalid(
+                    "duplicate source identity in import request"
+                )
+            identities.add(identity)
+            normalized_records.append((snapshot, source_updated_at))
+
+        items = []
+        try:
+            for snapshot, source_updated_at in normalized_records:
+                items.append(
+                    self._import_one(
+                        source_uid,
+                        source_entity,
+                        snapshot,
+                        source_updated_at,
+                        actor_uid,
+                    )
+                )
+            self.commit()
+        except Exception:
+            self.rollback()
+            raise
+        actions = [item.action for item in items]
+        return DeviceAssetImportResult(
+            items=tuple(items),
+            created_count=actions.count("created"),
+            updated_count=actions.count("updated"),
+            unchanged_count=actions.count("unchanged"),
+        )
+
+    def _import_one(
+        self,
+        source_uid,
+        source_entity,
+        snapshot,
+        source_updated_at,
+        actor_uid,
+    ):
+        now = self.now_factory()
+        digest = _content_hash(snapshot)
+        mapping = self.repository.find_mapping(
+            source_uid,
+            source_entity,
+            snapshot["asset_type"],
+            snapshot["source_code"],
+            for_update=True,
+        )
+        if mapping is None:
+            asset = DeviceAssetRecord(
+                uid=self.uid_factory(),
+                asset_type=snapshot["asset_type"],
+                name=snapshot["name"],
+                status=snapshot["status"],
+                current_version=1,
+                content_hash=digest,
+                location=snapshot["location"],
+                organization=snapshot["organization"],
+                responsible_person=snapshot["responsible_person"],
+                attributes=dict(snapshot["attributes"]),
+                created_by=actor_uid,
+                updated_by=actor_uid,
+                created_at=now,
+                updated_at=now,
+            )
+            mapping = DeviceAssetMappingRecord(
+                uid=self.uid_factory(),
+                asset_uid=asset.uid,
+                source_uid=source_uid,
+                source_entity=source_entity,
+                asset_type=asset.asset_type,
+                source_code=snapshot["source_code"],
+                source_updated_at=source_updated_at,
+                first_seen_at=now,
+                last_seen_at=now,
+            )
+            version = DeviceAssetVersionRecord(
+                uid=self.uid_factory(),
+                asset_uid=asset.uid,
+                version=1,
+                content_hash=digest,
+                snapshot=dict(snapshot),
+                source_mapping_uid=mapping.uid,
+                actor_uid=actor_uid,
+                created_at=now,
+            )
+            asset, mapping = self.repository.create_asset(
+                asset,
+                mapping,
+                version,
+            )
+            return DeviceAssetImportItem("created", asset, mapping)
+
+        asset = self.repository.get(mapping.asset_uid)
+        if asset is None:
+            raise DeviceAssetInvalid("source mapping references a missing asset")
+        if asset.content_hash == digest:
+            mapping = self.repository.touch_mapping(
+                mapping,
+                source_updated_at=source_updated_at,
+                last_seen_at=now,
+            )
+            return DeviceAssetImportItem("unchanged", asset, mapping)
+
+        asset = replace(
+            asset,
+            name=snapshot["name"],
+            status=snapshot["status"],
+            current_version=asset.current_version + 1,
+            content_hash=digest,
+            location=snapshot["location"],
+            organization=snapshot["organization"],
+            responsible_person=snapshot["responsible_person"],
+            attributes=dict(snapshot["attributes"]),
+            updated_by=actor_uid,
+            updated_at=now,
+        )
+        mapping = replace(
+            mapping,
+            source_updated_at=source_updated_at,
+            last_seen_at=now,
+        )
+        version = DeviceAssetVersionRecord(
+            uid=self.uid_factory(),
+            asset_uid=asset.uid,
+            version=asset.current_version,
+            content_hash=digest,
+            snapshot=dict(snapshot),
+            source_mapping_uid=mapping.uid,
+            actor_uid=actor_uid,
+            created_at=now,
+        )
+        asset, mapping = self.repository.update_asset(
+            asset,
+            mapping,
+            version,
+        )
+        return DeviceAssetImportItem("updated", asset, mapping)
+
+    def search(self, filters, *, page=1, page_size=20):
+        assets, total = self.repository.search(
+            _search_filters(filters or {}),
+            page=int(page),
+            page_size=int(page_size),
+        )
+        return [
+            DeviceAssetDetail(
+                asset=asset,
+                mappings=tuple(self.repository.list_mappings(asset.uid)),
+            )
+            for asset in assets
+        ], total
+
+    def get(self, asset_uid) -> DeviceAssetDetail:
+        asset = self.repository.get(str(asset_uid))
+        if asset is None:
+            raise DeviceAssetNotFound(
+                f"device asset {asset_uid} was not found"
+            )
+        return DeviceAssetDetail(
+            asset=asset,
+            mappings=tuple(self.repository.list_mappings(asset.uid)),
+        )
+
+    def versions(self, asset_uid):
+        self.get(asset_uid)
+        return self.repository.list_versions(str(asset_uid))

deployment/app/core/data_research/errors.py

@@ -48,3 +48,13 @@ class CandidateDecisionInvalid(DataResearchError):
 class DataElementProjectionInvalid(DataResearchError):
     code = "DATA_ELEMENT_PROJECTION_INVALID"
     http_status = 422
+
+
+class DeviceAssetInvalid(DataResearchError):
+    code = "DEVICE_ASSET_INVALID"
+    http_status = 422
+
+
+class DeviceAssetNotFound(DataResearchError):
+    code = "DEVICE_ASSET_NOT_FOUND"
+    http_status = 404

deployment/app/core/system/permissions.py

@@ -32,6 +32,7 @@ DATA_ELEMENTS_EDIT = "data-elements:edit"
 DATA_ELEMENTS_PUBLISH = "data-elements:publish"
 ONTOLOGIES_EDIT = "ontologies:edit"
 ONTOLOGIES_PUBLISH = "ontologies:publish"
+DEVICE_ASSETS_EDIT = "device-assets:edit"
 
 ROLE_PERMISSIONS = {
     "viewer": frozenset({READ_GOVERNANCE, RULES_READ}),
@@ -47,6 +48,7 @@ ROLE_PERMISSIONS = {
             INGESTION_RUN,
             DATA_ELEMENTS_EDIT,
             ONTOLOGIES_EDIT,
+            DEVICE_ASSETS_EDIT,
         }
     ),
     "admin": frozenset(
@@ -75,6 +77,7 @@ ROLE_PERMISSIONS = {
             DATA_ELEMENTS_PUBLISH,
             ONTOLOGIES_EDIT,
             ONTOLOGIES_PUBLISH,
+            DEVICE_ASSETS_EDIT,
         }
     ),
 }
@@ -131,6 +134,10 @@ def permission_for_request(path: str, method: str) -> tuple[str, ...]:
         return (KNOWLEDGE_MANAGE,)
     if path.startswith("/api/system/users"):
         return (MANAGE_USERS,)
+    if path.startswith("/api/development/v1/device-assets"):
+        if method == "GET":
+            return (READ_GOVERNANCE,)
+        return (DEVICE_ASSETS_EDIT,)
     if path.startswith("/api/development/v1/ingestion-jobs"):
         if method == "GET":
             return (READ_GOVERNANCE,)

deployment/app/models/data_research.py

@@ -6,7 +6,7 @@ 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
+from app.core.common.timezone_utils import now_china, now_china_naive
 
 
 JOB_STATUSES = (
@@ -222,6 +222,150 @@ class CatalogSnapshot(db.Model):
         }
 
 
+class DeviceAsset(db.Model):
+    __tablename__ = "device_assets"
+    __table_args__ = (
+        db.CheckConstraint(
+            "asset_type IN ("
+            "'device','component','measurement_point',"
+            "'alarm','maintenance_record'"
+            ")",
+            name="ck_device_asset_type",
+        ),
+        db.CheckConstraint(
+            "status IN ('active','retired')",
+            name="ck_device_asset_status",
+        ),
+        db.CheckConstraint(
+            "current_version > 0",
+            name="ck_device_asset_current_version",
+        ),
+        {"schema": "public"},
+    )
+
+    uid = db.Column(
+        UUID(as_uuid=False),
+        primary_key=True,
+        default=new_governance_uid,
+    )
+    asset_type = db.Column(db.String(40), nullable=False)
+    name = db.Column(db.String(300), nullable=False)
+    status = db.Column(db.String(20), nullable=False, default="active")
+    current_version = db.Column(db.Integer, nullable=False, default=1)
+    content_hash = db.Column(db.String(64), nullable=False)
+    location = db.Column(db.String(300))
+    organization = db.Column(db.String(300))
+    responsible_person = db.Column(db.String(300))
+    attributes = db.Column(JSONB, nullable=False, default=dict)
+    created_by = db.Column(db.String(100))
+    updated_by = db.Column(db.String(100))
+    created_at = db.Column(
+        db.DateTime(timezone=True),
+        nullable=False,
+        default=now_china,
+    )
+    updated_at = db.Column(
+        db.DateTime(timezone=True),
+        nullable=False,
+        default=now_china,
+    )
+
+
+class DeviceAssetSourceMapping(db.Model):
+    __tablename__ = "device_asset_source_mappings"
+    __table_args__ = (
+        db.CheckConstraint(
+            "asset_type IN ("
+            "'device','component','measurement_point',"
+            "'alarm','maintenance_record'"
+            ")",
+            name="ck_device_asset_mapping_type",
+        ),
+        db.UniqueConstraint(
+            "source_uid",
+            "source_entity",
+            "asset_type",
+            "source_code",
+            name="uq_device_asset_source_identity",
+        ),
+        {"schema": "public"},
+    )
+
+    uid = db.Column(
+        UUID(as_uuid=False),
+        primary_key=True,
+        default=new_governance_uid,
+    )
+    asset_uid = db.Column(
+        UUID(as_uuid=False),
+        db.ForeignKey("public.device_assets.uid", ondelete="CASCADE"),
+        nullable=False,
+    )
+    source_uid = db.Column(
+        UUID(as_uuid=False),
+        db.ForeignKey("public.ingestion_sources.uid"),
+        nullable=False,
+    )
+    source_entity = db.Column(db.String(300), nullable=False)
+    asset_type = db.Column(db.String(40), nullable=False)
+    source_code = db.Column(db.String(300), nullable=False)
+    source_updated_at = db.Column(db.DateTime(timezone=True))
+    first_seen_at = db.Column(
+        db.DateTime(timezone=True),
+        nullable=False,
+        default=now_china,
+    )
+    last_seen_at = db.Column(
+        db.DateTime(timezone=True),
+        nullable=False,
+        default=now_china,
+    )
+
+
+class DeviceAssetVersion(db.Model):
+    __tablename__ = "device_asset_versions"
+    __table_args__ = (
+        db.CheckConstraint(
+            "version > 0",
+            name="ck_device_asset_version_number",
+        ),
+        db.UniqueConstraint(
+            "asset_uid",
+            "version",
+            name="uq_device_asset_version",
+        ),
+        {"schema": "public"},
+    )
+
+    uid = db.Column(
+        UUID(as_uuid=False),
+        primary_key=True,
+        default=new_governance_uid,
+    )
+    asset_uid = db.Column(
+        UUID(as_uuid=False),
+        db.ForeignKey("public.device_assets.uid", ondelete="CASCADE"),
+        nullable=False,
+    )
+    version = db.Column(db.Integer, nullable=False)
+    content_hash = db.Column(db.String(64), nullable=False)
+    snapshot = db.Column(JSONB, nullable=False)
+    source_mapping_uid = db.Column(
+        UUID(as_uuid=False),
+        db.ForeignKey(
+            "public.device_asset_source_mappings.uid",
+            ondelete="RESTRICT",
+        ),
+        nullable=False,
+    )
+    actor_uid = db.Column(db.String(100))
+    created_at = db.Column(
+        db.DateTime(timezone=True),
+        nullable=False,
+        default=now_china,
+    )
+
+
 class EvidenceFragment(db.Model):
     __tablename__ = "evidence_fragments"
     __table_args__ = (

docs/DATAOPS_PHASE1_3_MONTH_WORK_PLAN_20260729.md

@@ -156,6 +156,7 @@ P2 不阻塞第一阶段验收。没有完成的 P2 功能必须保留接口和
 | WP-01 | 工程完成 | 合并提交 `cab9c1f`；后端 839 项回归、前端构建、OpenAPI 142 项和 Docker 本体链路通过 | `deployment/app` 历史发布副本仍有基线差异，正式交付前按 WP-13 统一收口 |
 | WP-02 | 工程完成，待企业配置 | 本地三级 RBAC；设备责任矩阵；唯一最终设备资产管理员；修订、并发冲突和审计快照；OpenAPI 144 项 | 需要企业提供实际设备资产管理员、治理人员和查看者名单后配置并完成 UAT |
 | WP-03 | 工程完成，待企业接入 | 数据源自动登记；只读目录执行；幂等与主动重采；尝试次数、失败阶段和脱敏诊断；每次执行不可变目录快照；字段级来源证据；本地隔离 PostgreSQL/MySQL 双源实测；OpenAPI 147 项 | 需要企业设备台账库、维修库只读账号、网络、采集范围和数据字典；当前同步执行满足示范版，异步 Worker 与断点续跑保留为后续增强 |
+| WP-04 | 工程完成，待企业数据验收 | 五类设备对象规范化导入；稳定平台 UID 与来源身份唯一映射；无变化不增版、变化生成不可变版本；关键词和类型/状态/来源筛选；来源与责任信息详情；查看者只读、编辑者受控导入；OpenAPI 151 项；本地 PostgreSQL 与页面链路定向验证 | 需要企业提供设备台账与维修样本、字段映射和业务确认；跨来源自动匹配、合并与回滚归 WP-06，质量与血缘汇聚不在 WP-04 |
 
 ## 7. 12 周执行计划
 

docs/FUNCTION_MODULE_CENSUS_20260726.md

@@ -541,6 +541,8 @@ DataOps Platform 当前已经具备较完整的“治理对象 → 知识服务
 | CAT-25 | 资产导入 / 批量导入 | HOPMS 数据集导入、dry-run 和结果报告 | 已建设 |
 | CAT-26 | 资产纠错 / 反馈 | 用户纠错、补充说明、问题上报和责任人处置 | 规划中 |
 
+WP-04 已补齐设备垂直切片的工程拼图：设备、部件、测点、告警和维护记录可通过受控接口导入，按名称、源编码、位置、组织和负责人检索，并展示稳定平台 UID、来源映射和不可变版本。CAT-13、CAT-14、CAT-21 仍保留“部分建设”，因为跨来源实体匹配、合并/回滚、质量与血缘汇聚分别属于 WP-06、WP-07 和 WP-09；企业真实台账与维修数据尚待接入验收。
+
 ### 12.5 数据标准、语义与本体
 
 | 模块编号 | 模块分级 | 功能项 | 成熟度 |
@@ -779,8 +781,8 @@ DataOps Platform 当前已经具备较完整的“治理对象 → 知识服务
 | 3 | CON-01～05 | 关系数据库接入 | 复用 PostgreSQL/MySQL 安全连接池接入设备台账和维修库 |
 | 4 | CON-09 | MES/SCADA/IoT 元数据接入 | 首期通过其数据库或既有接口采集资产、测点和 Schema，不开发全量工业协议 |
 | 5 | CON-17～19 | 企业内数据边界 | 全部部署在企业内网；保存聚合运行信息和近期明细，不向外传原始数据 |
-| 6 | CAT-01、CAT-04、CAT-13 | 设备资产目录 | 展示设备、部件、测点、告警、维修记录及来源信息 |
-| 7 | CAT-14、SEM-19 | 平台设备 UID | 为跨系统设备生成稳定 UID，保留所有源系统编码 |
+| 6 | CAT-01、CAT-04、CAT-13 | 设备资产目录 | 工程能力已形成；展示设备、部件、测点、告警、维修记录、责任信息及来源，待企业数据验收 |
+| 7 | CAT-14、SEM-19 | 平台设备 UID | 单一来源身份的稳定 UID 和源编码映射已形成；跨系统同实体合并顺延至 WP-06 |
 | 8 | SEM-15～17 | 设备实体匹配 | 规则 + AI 生成候选；高置信度自动合并，其他进入审核并可回滚 |
 | 9 | SEM-08～13 | 设备本体工作台 | 从隔离分支合入必要本体能力，建模设备、部件、位置、组织和责任人 |
 | 10 | SEM-20、SEM-21 | 故障代码统一 | 汇总多系统故障/原因/措施代码，AI 辅助聚类，设备资产管理员审批 |

docs/architecture/DATA_MODEL.md

@@ -146,6 +146,9 @@ flowchart LR
 | `data_elements` | `code`, `current_version`, `status`, `business_domain_uids` | 稳定数据元素身份与生命周期 |
 | `data_element_versions` | `data_element_uid`, `version`, `snapshot`, `evidence_uids` | 不可变数据元素版本 |
 | `candidate_decisions` | `candidate_uid`, `action`, `data_element_uid`, `actor_uid` | `reuse/create/map/ignore` 决策审计 |
+| `device_assets` | `uid`, `asset_type`, `name`, `current_version`, `content_hash`, `location`, `organization`, `responsible_person`, `attributes` | 设备、部件、测点、告警和维护记录的稳定平台档案 |
+| `device_asset_source_mappings` | `asset_uid`, `source_uid`, `source_entity`, `asset_type`, `source_code`, `source_updated_at` | 源系统身份映射；同一来源身份唯一，不在 WP-04 自动跨源合并 |
+| `device_asset_versions` | `asset_uid`, `version`, `content_hash`, `snapshot`, `source_mapping_uid`, `actor_uid` | 设备资产不可变版本和变更来源追溯 |
 | `ontologies` | `code`, `owner_uid`, `draft_revision`, `active_version_uid` | 本体稳定身份和生效版本 |
 | `ontology_versions` | `ontology_uid`, `version`, `parent_version_uid`, `graph_document`, `content_hash` | 不可变本体版本 |
 | `ontology_domain_links` | `ontology_uid`, `domain_uid`, `role` | 多业务域 owner/contributor/consumer 关系 |
@@ -173,4 +176,5 @@ flowchart LR
 - MinIO 是附件原件的源真相；PostgreSQL 只保存对象键和元数据。
 - n8n 是 Workflow 定义与执行记录的源真相；平台保存治理映射和生效状态。
 - 本体与数据元素的发布版本以 PostgreSQL 为源真相；Neo4j 是可重建的已发布语义投影。
+- 设备资产、源编码映射和不可变版本以 PostgreSQL 为源真相；跨来源匹配、合并与回滚在 WP-06 经审核后实施。
 - 本轮只清理代码和建库脚本。生产表必须在数据核查、备份和依赖确认后以独立变更单下线。

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: 147
+x-route-count: 151
 servers:
   - url: "http://localhost:15500"
     description: "全本地隔离测试后端"
@@ -1744,6 +1744,101 @@ paths:
             application/json:
               schema:
                 $ref: '#/components/schemas/ApiEnvelope'
+  "/api/development/v1/device-assets":
+    get:
+      tags: [data_development]
+      operationId: data_development_list_device_assets_get
+      summary: "list device assets"
+      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'
+  "/api/development/v1/device-assets/import":
+    post:
+      tags: [data_development]
+      operationId: data_development_import_device_assets_post
+      summary: "import device assets"
+      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/device-assets/{asset_uid}":
+    get:
+      tags: [data_development]
+      operationId: data_development_get_device_asset_get
+      summary: "get device asset"
+      x-source: "app/api/data_development/routes.py"
+      parameters:
+        - name: asset_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/device-assets/{asset_uid}/versions":
+    get:
+      tags: [data_development]
+      operationId: data_development_list_device_asset_versions_get
+      summary: "list device asset versions"
+      x-source: "app/api/data_development/routes.py"
+      parameters:
+        - name: asset_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/evidence/{evidence_uid}":
     get:
       tags: [data_development]

docs/superpowers/plans/2026-07-29-wp04-device-asset-catalog.md

@@ -0,0 +1,200 @@
+# WP-04 Device Asset Catalog 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:** Build a canonical device asset catalog with stable platform UIDs, immutable versions, source-system code mappings, source timestamps, search, and end-to-end traceability for devices, components, measurement points, alarms, and maintenance records.
+
+**Architecture:** PostgreSQL remains authoritative for canonical device assets, source mappings, and immutable versions. A normalized import service accepts bounded records from the existing ingestion boundary, never reads credentials, and never merges two source identities automatically; later WP-06 matching may explicitly link identities after review. The Vue 2 catalog is a read and trace interface, while write access is exposed through a dedicated permission-controlled API for adapters and data engineers.
+
+**Tech Stack:** Flask, SQLAlchemy, PostgreSQL JSONB, Alembic, Vue 2, Vuetify, pytest.
+
+## Global Constraints
+
+- Supported asset types are `device`, `component`, `measurement_point`, `alarm`, and `maintenance_record`.
+- Stable asset identity is preserved by the unique tuple `(source_uid, source_entity, asset_type, source_code)`.
+- Reimporting an unchanged record updates source observation time but does not create a new asset version.
+- A changed normalized record creates exactly one immutable next version.
+- WP-04 does not automatically merge identities across sources; matching, review, merge, and rollback belong to WP-06.
+- Credentials, connection strings, passwords, API keys, and tokens are rejected from asset attributes and never returned.
+- Imports contain at most 500 records and each normalized attributes object is bounded.
+- Viewer can read; editor and admin can import through `device-assets:edit`.
+- Real enterprise device and maintenance data remains an external acceptance gate.
+- Validation follows the user-approved rule: only changed functions, routes, migration, frontend files, and the local WP-04 browser flow are tested; no full repository regression is run.
+- Work continues on `codex/dataops-phase1-equipment-governance`; no push or remote deployment is authorized.
+
+---
+
+### Task 1: Canonical Asset Identity and Version Semantics
+
+**Files:**
+- Create: `app/core/data_research/device_assets.py`
+- Modify: `app/core/data_research/errors.py`
+- Test: `tests/data_research/test_device_assets.py`
+
+**Interfaces:**
+- Consumes: `repository.source_is_active(source_uid)`, `repository.find_mapping(...)`, `repository.create_asset(...)`, `repository.update_asset(...)`, and `repository.touch_mapping(...)`.
+- Produces: `DeviceAssetService.import_records(payload, actor_uid)`, `DeviceAssetRecord`, `DeviceAssetMappingRecord`, `DeviceAssetVersionRecord`, and `DeviceAssetImportResult`.
+
+- [x] **Step 1: Write failing domain tests**
+
+Cover stable UID reuse, unchanged reimport without a new version, changed record with exactly one new version, five allowed asset types, inactive source rejection, duplicate source code rejection inside one request, bounded batch size, required source entity/code/name, attributes size, secret-key rejection, and no caller-supplied asset UID.
+
+- [x] **Step 2: Verify RED**
+
+Run:
+
+```bash
+PYTHONPATH=. .venv/bin/pytest -q tests/data_research/test_device_assets.py
+```
+
+Expected: failure because the device asset service does not exist.
+
+- [x] **Step 3: Implement minimal domain service**
+
+Normalize fields, compute a deterministic SHA-256 content hash, create a UUIDv7 only for a new source identity, preserve the existing UID on reimport, and classify every input as `created`, `updated`, or `unchanged`.
+
+- [x] **Step 4: Verify GREEN**
+
+Run the Task 1 test command and expect all tests to pass.
+
+### Task 2: PostgreSQL Persistence and Migration
+
+**Files:**
+- Create: `app/core/data_research/device_asset_repository.py`
+- Modify: `app/models/data_research.py`
+- Create: `migrations/versions/20260729_290_device_asset_catalog.py`
+- Modify: `tests/test_database_migrations.py`
+- Create: `tests/integration/test_device_asset_postgres.py`
+
+**Interfaces:**
+- Produces: `SqlAlchemyDeviceAssetRepository`, tables `device_assets`, `device_asset_source_mappings`, and `device_asset_versions`.
+
+- [x] **Step 1: Write failing persistence tests**
+
+Add a migration contract test and a real PostgreSQL integration test covering source validation, one canonical asset, one unique mapping, immutable sequential versions, source timestamp refresh, search by source code/name/location/responsible person, type/status/source filters, and detail traceability.
+
+- [x] **Step 2: Verify RED**
+
+Run:
+
+```bash
+PYTHONPATH=. .venv/bin/pytest -q \
+  tests/test_database_migrations.py::test_device_asset_catalog_migration_is_versioned_and_traceable \
+  tests/integration/test_device_asset_postgres.py
+```
+
+Expected: failure because the migration, models, and SQL repository do not exist.
+
+- [x] **Step 3: Implement schema and SQL repository**
+
+Use unique constraints for source identity and asset version, row locking for updates, indexed search fields, source and actor timestamps, JSONB snapshots, and data-preserving downgrade behavior.
+
+- [x] **Step 4: Upgrade local database and verify GREEN**
+
+Run:
+
+```bash
+docker exec dataops-test-backend-1 alembic upgrade head
+PYTHONPATH=. .venv/bin/pytest -q \
+  tests/test_database_migrations.py::test_device_asset_catalog_migration_is_versioned_and_traceable \
+  tests/integration/test_device_asset_postgres.py
+```
+
+Expected: migration head `20260729_290`; tests pass.
+
+### Task 3: Permission-Controlled Catalog API
+
+**Files:**
+- Modify: `app/api/data_development/routes.py`
+- Modify: `app/core/system/permissions.py`
+- Modify: `tests/data_research/test_development_api.py`
+- Modify: `tests/test_permission_matrix.py`
+
+**Interfaces:**
+- Produces: `POST /api/development/v1/device-assets/import`, `GET /api/development/v1/device-assets`, `GET /api/development/v1/device-assets/{asset_uid}`, and `GET /api/development/v1/device-assets/{asset_uid}/versions`.
+
+- [x] **Step 1: Write failing API and permission tests**
+
+Cover viewer read, viewer import denial, editor import, secret-free responses, search/filter pagination, detail mappings, version history, malformed pagination, and not-found errors.
+
+- [x] **Step 2: Verify RED**
+
+Run:
+
+```bash
+PYTHONPATH=. .venv/bin/pytest -q \
+  tests/data_research/test_development_api.py \
+  tests/test_permission_matrix.py::test_data_development_paths_have_specific_write_policies
+```
+
+Expected: failure because the routes and permission do not exist.
+
+- [x] **Step 3: Implement API and permission boundary**
+
+Add `device-assets:edit` to editor/admin, keep reads on `governance:read`, bound page to at least 1 and page size to 1–100, serialize source mappings and versions without secret material, and return domain errors through the existing data-research error envelope.
+
+- [x] **Step 4: Verify GREEN**
+
+Run the Task 3 test command and expect all tests to pass.
+
+### Task 4: Device Asset Catalog UI and Delivery Evidence
+
+**Files:**
+- Modify: `frontend/src/api/dataDevelopment.js`
+- Modify: `frontend/src/router/routes.js`
+- Modify: `frontend/src/views/dataGovernance/development/index.vue`
+- Create: `frontend/src/views/dataGovernance/development/deviceAssets.vue`
+- Modify: `docs/architecture/OPENAPI.yaml`
+- Modify: `docs/architecture/DATA_MODEL.md`
+- Modify: `docs/FUNCTION_MODULE_CENSUS_20260726.md`
+- Modify: `docs/DATAOPS_PHASE1_3_MONTH_WORK_PLAN_20260729.md`
+- Modify: `deployment/app/` only for the WP-04 backend subset.
+
+**Interfaces:**
+- Produces: `/data-governance/development/device-assets` with keyword/type/status filters, stable UID display, source-code chips, source timestamps, responsibility fields, detail traceability, and immutable version history.
+
+- [x] **Step 1: Implement the catalog UI**
+
+Add API functions, hidden child route, research-center entry, responsive table, detail dialog, empty/error/loading states, source mappings, and version timeline. Do not add a generic analysis-development or raw-SQL interface.
+
+- [x] **Step 2: Regenerate contracts and update ledgers**
+
+Run:
+
+```bash
+.venv/bin/python scripts/generate_openapi.py
+```
+
+Record engineering completion separately from enterprise data acceptance and keep WP-06 automatic matching outside WP-04.
+
+- [x] **Step 3: Run targeted verification**
+
+Run only:
+
+```bash
+PYTHONPATH=. .venv/bin/pytest -q \
+  tests/data_research/test_device_assets.py \
+  tests/integration/test_device_asset_postgres.py \
+  tests/data_research/test_development_api.py \
+  tests/test_permission_matrix.py::test_data_development_paths_have_specific_write_policies \
+  tests/test_database_migrations.py::test_device_asset_catalog_migration_is_versioned_and_traceable \
+  tests/test_architecture_artifacts.py
+npx eslint src/api/dataDevelopment.js src/router/routes.js \
+  src/views/dataGovernance/development/index.vue \
+  src/views/dataGovernance/development/deviceAssets.vue
+```
+
+Then rebuild only the affected local backend/frontend services and exercise import, search, detail, source mapping, and version history in the browser. Run `git diff --check`.
+
+- [x] **Step 4: Commit**
+
+Create one independently reversible WP-04 engineering commit. Do not push.
+
+## Implementation Receipt
+
+- Domain RED/GREEN covered stable UID reuse, unchanged import, immutable next version, five object types, secret rejection, bounded inputs, source/filter validation, source mappings, and timezone-aware catalog timestamps.
+- PostgreSQL migration `20260729_290` is the local head; the real PostgreSQL integration test covers source identity, search, filters, detail, and version history.
+- Four device asset endpoints use `governance:read` for reads and `device-assets:edit` for import. OpenAPI now inventories 151 operations.
+- The Vue catalog, hidden child route, research-center entry, and API client passed changed-file ESLint; affected local backend/frontend images built successfully.
+- Browser acceptance retained `EQ-WP04-DEMO`: source-code search returned one stable asset at V2, detail showed source mapping plus V1/V2 locations and correct China timestamps, with zero browser console errors.
+- Enterprise device and maintenance records remain the only WP-04 acceptance gate. Cross-source matching, merge, and rollback remain WP-06.

frontend/src/api/dataDevelopment.js

@@ -2,6 +2,7 @@ import http from '@/utils/request'
 
 const BASE = '/development/v1/ingestion-jobs'
 const ONTOLOGY_BASE = '/development/v1/ontologies'
+const DEVICE_ASSET_BASE = '/development/v1/device-assets'
 
 export const createIngestionJob = params => http.post(BASE, params)
 export const getIngestionJobs = params => http.get(BASE, params)
@@ -65,6 +66,15 @@ export const importOntology = (formData, format) => http.upload(
   `${ONTOLOGY_BASE}/import?format=${format}`,
   formData
 )
+export const getDeviceAssets = params => http.get(DEVICE_ASSET_BASE, params)
+export const getDeviceAsset = uid => http.get(`${DEVICE_ASSET_BASE}/${uid}`)
+export const getDeviceAssetVersions = uid => http.get(
+  `${DEVICE_ASSET_BASE}/${uid}/versions`
+)
+export const importDeviceAssets = params => http.post(
+  `${DEVICE_ASSET_BASE}/import`,
+  params
+)
 
 export default {
   createIngestionJob,
@@ -94,5 +104,9 @@ export default {
   generateOntologySuggestions,
   reviewOntologyChangeSet,
   exportOntology,
-  importOntology
+  importOntology,
+  getDeviceAssets,
+  getDeviceAsset,
+  getDeviceAssetVersions,
+  importDeviceAssets
 }

frontend/src/router/routes.js

@@ -216,6 +216,18 @@ export default {
           name: 'dataResearchTasks',
           alwaysShow: 0
         },
+        {
+          hidden: 1,
+          type: 1,
+          title: '设备台账',
+          path: '/data-governance/development/device-assets',
+          children: [],
+          label: '设备台账',
+          component: 'dataGovernance/development/deviceAssets',
+          meta: { roles: ['viewer', 'editor', 'admin'], title: '设备台账', readOnly: 'viewer' },
+          name: 'dataResearchDeviceAssets',
+          alwaysShow: 0
+        },
         {
           hidden: 1,
           type: 1,

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

@@ -0,0 +1,381 @@
+<template>
+  <div class="pa-6 device-assets">
+    <div class="d-flex flex-wrap align-center mb-5">
+      <div>
+        <h1 class="text-h4 mb-1">设备台账</h1>
+        <div class="text--secondary">
+          统一管理设备、部件、测点、告警与维护记录，保留来源映射和变更版本。
+        </div>
+      </div>
+      <v-spacer />
+      <v-chip color="primary" outlined>
+        共 {{ total }} 项
+      </v-chip>
+    </div>
+
+    <v-card outlined class="mb-5">
+      <v-card-text>
+        <v-row dense align="center">
+          <v-col cols="12" md="4">
+            <v-text-field
+              v-model.trim="filters.keyword"
+              label="搜索名称、位置、组织、负责人或来源编码"
+              prepend-inner-icon="mdi-magnify"
+              clearable
+              hide-details
+              @keyup.enter="applyFilters"
+            />
+          </v-col>
+          <v-col cols="12" sm="6" md="2">
+            <v-select
+              v-model="filters.asset_type"
+              :items="typeOptions"
+              item-text="text"
+              item-value="value"
+              label="对象类型"
+              clearable
+              hide-details
+            />
+          </v-col>
+          <v-col cols="12" sm="6" md="2">
+            <v-select
+              v-model="filters.status"
+              :items="statusOptions"
+              item-text="text"
+              item-value="value"
+              label="状态"
+              clearable
+              hide-details
+            />
+          </v-col>
+          <v-col cols="12" sm="6" md="2">
+            <v-text-field
+              v-model.trim="filters.source_uid"
+              label="来源 UID"
+              clearable
+              hide-details
+            />
+          </v-col>
+          <v-col cols="12" sm="6" md="2" class="d-flex">
+            <v-btn color="primary" class="mr-2" @click="applyFilters">
+              查询
+            </v-btn>
+            <v-btn text @click="resetFilters">重置</v-btn>
+          </v-col>
+        </v-row>
+      </v-card-text>
+    </v-card>
+
+    <v-card outlined>
+      <v-data-table
+        :headers="headers"
+        :items="items"
+        :loading="loading"
+        :items-per-page="pageSize"
+        hide-default-footer
+      >
+        <template v-slot:[`item.name`]="{ item }">
+          <div class="py-2">
+            <div class="font-weight-medium">{{ item.name }}</div>
+            <div class="text-caption text--secondary">
+              {{ sourceCode(item) || item.uid }}
+            </div>
+          </div>
+        </template>
+        <template v-slot:[`item.asset_type`]="{ item }">
+          <v-chip small outlined color="primary">
+            {{ typeLabel(item.asset_type) }}
+          </v-chip>
+        </template>
+        <template v-slot:[`item.location`]="{ item }">
+          <div>{{ item.location || '—' }}</div>
+          <div class="text-caption text--secondary">
+            {{ item.organization || '未归属组织' }}
+          </div>
+        </template>
+        <template v-slot:[`item.status`]="{ item }">
+          <v-chip
+            small
+            :color="item.status === 'active' ? 'success' : 'grey'"
+            text-color="white"
+          >
+            {{ item.status === 'active' ? '在用' : '停用' }}
+          </v-chip>
+        </template>
+        <template v-slot:[`item.current_version`]="{ item }">
+          V{{ item.current_version }}
+        </template>
+        <template v-slot:[`item.actions`]="{ item }">
+          <v-btn text color="primary" @click="openDetail(item)">
+            查看追溯
+          </v-btn>
+        </template>
+        <template v-slot:no-data>
+          <div class="py-10 text--secondary">
+            暂无符合条件的设备台账数据
+          </div>
+        </template>
+      </v-data-table>
+
+      <v-divider />
+      <div class="d-flex flex-wrap align-center pa-4">
+        <span class="text-caption text--secondary">
+          第 {{ page }} 页，每页 {{ pageSize }} 项
+        </span>
+        <v-spacer />
+        <v-pagination
+          v-model="page"
+          :length="pageCount"
+          :total-visible="7"
+          @input="load"
+        />
+      </div>
+    </v-card>
+
+    <v-dialog v-model="detailDialog" max-width="980" scrollable>
+      <v-card>
+        <v-card-title class="d-flex align-center">
+          <div>
+            <div>{{ detail ? detail.name : '设备台账详情' }}</div>
+            <div v-if="detail" class="text-caption text--secondary">
+              {{ typeLabel(detail.asset_type) }} · V{{ detail.current_version }}
+            </div>
+          </div>
+          <v-spacer />
+          <v-btn icon @click="detailDialog = false">
+            <v-icon>mdi-close</v-icon>
+          </v-btn>
+        </v-card-title>
+        <v-divider />
+        <v-card-text class="pt-5">
+          <v-progress-linear v-if="detailLoading" indeterminate />
+          <template v-else-if="detail">
+            <v-row>
+              <v-col cols="12" md="4">
+                <div class="field-label">位置</div>
+                <div>{{ detail.location || '—' }}</div>
+              </v-col>
+              <v-col cols="12" md="4">
+                <div class="field-label">归属组织</div>
+                <div>{{ detail.organization || '—' }}</div>
+              </v-col>
+              <v-col cols="12" md="4">
+                <div class="field-label">负责人</div>
+                <div>{{ detail.responsible_person || '—' }}</div>
+              </v-col>
+            </v-row>
+
+            <h3 class="text-subtitle-1 font-weight-bold mt-5 mb-2">
+              来源映射
+            </h3>
+            <v-simple-table dense>
+              <thead>
+                <tr>
+                  <th>来源 UID</th>
+                  <th>来源实体</th>
+                  <th>来源编码</th>
+                  <th>最近发现</th>
+                </tr>
+              </thead>
+              <tbody>
+                <tr
+                  v-for="mapping in detail.source_mappings"
+                  :key="mapping.uid"
+                >
+                  <td class="mono">{{ mapping.source_uid }}</td>
+                  <td>{{ mapping.source_entity }}</td>
+                  <td>{{ mapping.source_code }}</td>
+                  <td>{{ displayTime(mapping.last_seen_at) }}</td>
+                </tr>
+              </tbody>
+            </v-simple-table>
+
+            <h3 class="text-subtitle-1 font-weight-bold mt-6 mb-2">
+              版本历史
+            </h3>
+            <v-timeline dense align-top>
+              <v-timeline-item
+                v-for="version in versions"
+                :key="version.uid"
+                small
+                color="primary"
+              >
+                <div class="d-flex flex-wrap align-center">
+                  <strong class="mr-3">V{{ version.version }}</strong>
+                  <span class="text-caption text--secondary">
+                    {{ displayTime(version.created_at) }}
+                  </span>
+                </div>
+                <div class="text-body-2 mt-1">
+                  {{ version.snapshot.name }}
+                  <span v-if="version.snapshot.location">
+                    · {{ version.snapshot.location }}
+                  </span>
+                </div>
+              </v-timeline-item>
+            </v-timeline>
+
+            <h3 class="text-subtitle-1 font-weight-bold mt-5 mb-2">
+              扩展属性
+            </h3>
+            <pre class="attributes">{{ prettyAttributes }}</pre>
+          </template>
+        </v-card-text>
+      </v-card>
+    </v-dialog>
+  </div>
+</template>
+
+<script>
+import {
+  getDeviceAsset,
+  getDeviceAssets,
+  getDeviceAssetVersions
+} from '@/api/dataDevelopment'
+
+export default {
+  name: 'DataResearchDeviceAssets',
+  data: () => ({
+    loading: false,
+    detailLoading: false,
+    detailDialog: false,
+    detail: null,
+    versions: [],
+    items: [],
+    total: 0,
+    page: 1,
+    pageSize: 20,
+    filters: {
+      keyword: '',
+      asset_type: null,
+      status: null,
+      source_uid: ''
+    },
+    typeOptions: [
+      { text: '设备', value: 'device' },
+      { text: '部件', value: 'component' },
+      { text: '测点', value: 'measurement_point' },
+      { text: '告警', value: 'alarm' },
+      { text: '维护记录', value: 'maintenance_record' }
+    ],
+    statusOptions: [
+      { text: '在用', value: 'active' },
+      { text: '停用', value: 'retired' }
+    ],
+    headers: [
+      { text: '台账名称 / 来源编码', value: 'name' },
+      { text: '对象类型', value: 'asset_type' },
+      { text: '位置 / 归属组织', value: 'location' },
+      { text: '负责人', value: 'responsible_person' },
+      { text: '状态', value: 'status' },
+      { text: '版本', value: 'current_version' },
+      { text: '操作', value: 'actions', sortable: false }
+    ]
+  }),
+  computed: {
+    pageCount () {
+      return Math.max(1, Math.ceil(this.total / this.pageSize))
+    },
+    prettyAttributes () {
+      return JSON.stringify(
+        (this.detail && this.detail.attributes) || {},
+        null,
+        2
+      )
+    }
+  },
+  created () {
+    this.load()
+  },
+  methods: {
+    async load () {
+      this.loading = true
+      try {
+        const response = await getDeviceAssets({
+          ...this.filters,
+          page: this.page,
+          page_size: this.pageSize
+        })
+        this.items = (response.data && response.data.records) || []
+        this.total = (response.data && response.data.total) || 0
+      } catch (error) {
+        this.$snackbar.error(error)
+      } finally {
+        this.loading = false
+      }
+    },
+    applyFilters () {
+      this.page = 1
+      this.load()
+    },
+    resetFilters () {
+      this.filters = {
+        keyword: '',
+        asset_type: null,
+        status: null,
+        source_uid: ''
+      }
+      this.applyFilters()
+    },
+    async openDetail (item) {
+      this.detailDialog = true
+      this.detailLoading = true
+      this.detail = null
+      this.versions = []
+      try {
+        const [detail, versions] = await Promise.all([
+          getDeviceAsset(item.uid),
+          getDeviceAssetVersions(item.uid)
+        ])
+        this.detail = detail.data
+        this.versions = (versions.data && versions.data.records) || []
+      } catch (error) {
+        this.$snackbar.error(error)
+        this.detailDialog = false
+      } finally {
+        this.detailLoading = false
+      }
+    },
+    sourceCode (item) {
+      const mapping = (item.source_mappings || [])[0]
+      return mapping && mapping.source_code
+    },
+    typeLabel (value) {
+      const option = this.typeOptions.find(item => item.value === value)
+      return option ? option.text : value
+    },
+    displayTime (value) {
+      if (!value) return '—'
+      return new Date(value).toLocaleString('zh-CN', { hour12: false })
+    }
+  }
+}
+</script>
+
+<style lang="scss" scoped>
+.device-assets {
+  max-width: 1500px;
+  margin: 0 auto;
+}
+
+.field-label {
+  color: rgba(0, 0, 0, 0.6);
+  font-size: 12px;
+  margin-bottom: 4px;
+}
+
+.mono {
+  font-family: ui-monospace, SFMono-Regular, Menlo, monospace;
+  font-size: 12px;
+}
+
+.attributes {
+  max-height: 260px;
+  overflow: auto;
+  padding: 16px;
+  border-radius: 6px;
+  background: #f5f7fa;
+  font-size: 12px;
+  white-space: pre-wrap;
+}
+</style>

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

@@ -24,6 +24,7 @@ export default {
       { title: '新建采集', description: '数据库目录、DDL 与文件解析', icon: 'mdi-database-import-outline', path: '/data-governance/development/ingestion' },
       { title: '研发任务', description: '查看解析进度、失败阶段与重试状态', icon: 'mdi-progress-clock', path: '/data-governance/development/tasks' },
       { title: '治理评审', description: '证据预览、批量决策与数据元素生命周期', icon: 'mdi-clipboard-check-outline', path: '/data-governance/development/review' },
+      { title: '设备台账', description: '设备、部件、测点、告警与维护记录统一追溯', icon: 'mdi-factory', path: '/data-governance/development/device-assets' },
       { title: '本体中心', description: '跨业务域本体定义、校验、发布与回滚', icon: 'mdi-graph-outline', path: '/data-governance/ontology' }
     ]
   })

migrations/versions/20260729_290_device_asset_catalog.py

@@ -0,0 +1,91 @@
+"""Add canonical device assets, source mappings, and immutable versions."""
+
+from alembic import op
+
+
+revision = "20260729_290"
+down_revision = "20260729_280"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    op.execute(
+        """
+        CREATE TABLE public.device_assets (
+            uid UUID PRIMARY KEY,
+            asset_type VARCHAR(40) NOT NULL
+                CHECK (
+                    asset_type IN (
+                        'device','component','measurement_point',
+                        'alarm','maintenance_record'
+                    )
+                ),
+            name VARCHAR(300) NOT NULL,
+            status VARCHAR(20) NOT NULL DEFAULT 'active'
+                CHECK (status IN ('active','retired')),
+            current_version INTEGER NOT NULL DEFAULT 1
+                CHECK (current_version > 0),
+            content_hash CHAR(64) NOT NULL,
+            location VARCHAR(300),
+            organization VARCHAR(300),
+            responsible_person VARCHAR(300),
+            attributes JSONB NOT NULL DEFAULT '{}'::jsonb,
+            created_by VARCHAR(100),
+            updated_by VARCHAR(100),
+            created_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP,
+            updated_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP
+        );
+        CREATE INDEX idx_device_assets_type_status
+            ON public.device_assets(asset_type, status);
+        CREATE INDEX idx_device_assets_updated
+            ON public.device_assets(updated_at DESC);
+
+        CREATE TABLE public.device_asset_source_mappings (
+            uid UUID PRIMARY KEY,
+            asset_uid UUID NOT NULL
+                REFERENCES public.device_assets(uid) ON DELETE CASCADE,
+            source_uid UUID NOT NULL
+                REFERENCES public.ingestion_sources(uid),
+            source_entity VARCHAR(300) NOT NULL,
+            asset_type VARCHAR(40) NOT NULL
+                CHECK (
+                    asset_type IN (
+                        'device','component','measurement_point',
+                        'alarm','maintenance_record'
+                    )
+                ),
+            source_code VARCHAR(300) NOT NULL,
+            source_updated_at TIMESTAMPTZ,
+            first_seen_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP,
+            last_seen_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP,
+            UNIQUE (source_uid, source_entity, asset_type, source_code)
+        );
+        CREATE INDEX idx_device_asset_mapping_asset
+            ON public.device_asset_source_mappings(asset_uid);
+        CREATE INDEX idx_device_asset_mapping_source_code
+            ON public.device_asset_source_mappings(source_code);
+
+        CREATE TABLE public.device_asset_versions (
+            uid UUID PRIMARY KEY,
+            asset_uid UUID NOT NULL
+                REFERENCES public.device_assets(uid) ON DELETE CASCADE,
+            version INTEGER NOT NULL CHECK (version > 0),
+            content_hash CHAR(64) NOT NULL,
+            snapshot JSONB NOT NULL,
+            source_mapping_uid UUID NOT NULL
+                REFERENCES public.device_asset_source_mappings(uid)
+                ON DELETE RESTRICT,
+            actor_uid VARCHAR(100),
+            created_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP,
+            UNIQUE (asset_uid, version)
+        );
+        CREATE INDEX idx_device_asset_versions_created
+            ON public.device_asset_versions(asset_uid, created_at DESC);
+        """
+    )
+
+
+def downgrade() -> None:
+    # Device identity and immutable history are retained on application rollback.
+    pass

tests/data_research/test_development_api.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 from dataclasses import replace
+from datetime import datetime
 
 import pytest
 
@@ -100,6 +101,99 @@ class FakeEvidenceService:
         ]
 
 
+class FakeDeviceAssetService:
+    def __init__(self):
+        from app.core.data_research.device_assets import (
+            DeviceAssetDetail,
+            DeviceAssetImportItem,
+            DeviceAssetImportResult,
+            DeviceAssetMappingRecord,
+            DeviceAssetRecord,
+            DeviceAssetVersionRecord,
+        )
+
+        now = datetime(2026, 7, 29, 11, 0)
+        self.asset = DeviceAssetRecord(
+            uid="00000000-0000-7000-8000-000000000301",
+            asset_type="device",
+            name="一号循环泵",
+            status="active",
+            current_version=2,
+            content_hash="c" * 64,
+            location="动力车间",
+            organization="设备动力部",
+            responsible_person="张工",
+            attributes={"model": "P-100"},
+            created_by="editor-1",
+            updated_by="editor-1",
+            created_at=now,
+            updated_at=now,
+        )
+        self.mapping = DeviceAssetMappingRecord(
+            uid="00000000-0000-7000-8000-000000000302",
+            asset_uid=self.asset.uid,
+            source_uid="00000000-0000-0000-0000-000000000101",
+            source_entity="asset.equipment",
+            asset_type="device",
+            source_code="EQ-001",
+            source_updated_at=now,
+            first_seen_at=now,
+            last_seen_at=now,
+        )
+        self.version = DeviceAssetVersionRecord(
+            uid="00000000-0000-7000-8000-000000000303",
+            asset_uid=self.asset.uid,
+            version=2,
+            content_hash=self.asset.content_hash,
+            snapshot={
+                "asset_type": "device",
+                "source_code": "EQ-001",
+                "name": "一号循环泵",
+                "status": "active",
+                "location": "动力车间",
+                "organization": "设备动力部",
+                "responsible_person": "张工",
+                "attributes": {"model": "P-100"},
+            },
+            source_mapping_uid=self.mapping.uid,
+            actor_uid="editor-1",
+            created_at=now,
+        )
+        self.detail = DeviceAssetDetail(
+            asset=self.asset,
+            mappings=(self.mapping,),
+        )
+        self.import_result = DeviceAssetImportResult(
+            items=(
+                DeviceAssetImportItem(
+                    action="updated",
+                    asset=self.asset,
+                    mapping=self.mapping,
+                ),
+            ),
+            created_count=0,
+            updated_count=1,
+            unchanged_count=0,
+        )
+        self.actions = []
+
+    def import_records(self, payload, actor_uid):
+        self.actions.append(("import", payload, actor_uid))
+        return self.import_result
+
+    def search(self, filters, *, page, page_size):
+        self.actions.append(("search", filters, page, page_size))
+        return [self.detail], 1
+
+    def get(self, asset_uid):
+        self.actions.append(("get", asset_uid))
+        return self.detail
+
+    def versions(self, asset_uid):
+        self.actions.append(("versions", asset_uid))
+        return [self.version]
+
+
 @pytest.fixture()
 def development_client(monkeypatch):
     from flask import request
@@ -122,6 +216,7 @@ def development_client(monkeypatch):
     service = FakeDevelopmentService(record)
     registrar = FakeSourceRegistrar()
     executor = FakeCatalogExecutor(record)
+    asset_service = FakeDeviceAssetService()
 
     def identity():
         header = request.headers.get("Authorization", "")
@@ -157,10 +252,17 @@ def development_client(monkeypatch):
         "get_evidence_service",
         lambda: FakeEvidenceService(),
     )
+    monkeypatch.setattr(
+        routes,
+        "get_device_asset_service",
+        lambda: asset_service,
+        raising=False,
+    )
     app = create_app()
     app.config.update(TESTING=True)
     service.registrar = registrar
     service.executor = executor
+    service.asset_service = asset_service
     return app.test_client(), service
 
 
@@ -288,3 +390,109 @@ def test_catalog_snapshots_and_evidence_are_readable_without_secrets(
     text = evidence.get_data(as_text=True)
     assert "clear-secret" not in text
     assert "equipment_code" in text
+
+
+def test_device_asset_catalog_is_readable_and_import_requires_edit_permission(
+    development_client,
+):
+    client, service = development_client
+    endpoint = "/api/development/v1/device-assets"
+    payload = {
+        "source_uid": "00000000-0000-0000-0000-000000000101",
+        "source_entity": "asset.equipment",
+        "records": [
+            {
+                "asset_type": "device",
+                "source_code": "EQ-001",
+                "name": "一号循环泵",
+                "attributes": {"model": "P-100"},
+            }
+        ],
+    }
+
+    viewed = client.get(
+        f"{endpoint}?keyword=循环泵&asset_type=device&status=active"
+        "&source_uid=00000000-0000-0000-0000-000000000101"
+        "&page=2&page_size=10",
+        headers={"Authorization": "Bearer viewer"},
+    )
+    denied = client.post(
+        f"{endpoint}/import",
+        json=payload,
+        headers={"Authorization": "Bearer viewer"},
+    )
+    imported = client.post(
+        f"{endpoint}/import",
+        json=payload,
+        headers={"Authorization": "Bearer editor"},
+    )
+
+    assert viewed.status_code == 200
+    data = viewed.get_json()["data"]
+    assert data["page"] == 2
+    assert data["page_size"] == 10
+    assert data["total"] == 1
+    assert data["records"][0]["uid"].endswith("0301")
+    assert data["records"][0]["source_mappings"][0]["source_code"] == "EQ-001"
+    assert service.asset_service.actions[0] == (
+        "search",
+        {
+            "keyword": "循环泵",
+            "asset_type": "device",
+            "status": "active",
+            "source_uid": "00000000-0000-0000-0000-000000000101",
+        },
+        2,
+        10,
+    )
+    assert denied.status_code == 403
+    assert imported.status_code == 200
+    assert imported.get_json()["data"]["updated_count"] == 1
+    assert service.asset_service.actions[-1][2] == "editor-1"
+
+
+def test_device_asset_detail_and_versions_return_traceability_without_secrets(
+    development_client,
+):
+    client, _service = development_client
+    uid = "00000000-0000-7000-8000-000000000301"
+
+    detail = client.get(
+        f"/api/development/v1/device-assets/{uid}",
+        headers={"Authorization": "Bearer viewer"},
+    )
+    versions = client.get(
+        f"/api/development/v1/device-assets/{uid}/versions",
+        headers={"Authorization": "Bearer viewer"},
+    )
+
+    assert detail.status_code == 200
+    assert versions.status_code == 200
+    assert detail.get_json()["data"]["current_version"] == 2
+    assert detail.get_json()["data"]["source_mappings"][0][
+        "source_entity"
+    ] == "asset.equipment"
+    assert versions.get_json()["data"]["records"][0]["version"] == 2
+    text = detail.get_data(as_text=True) + versions.get_data(as_text=True)
+    assert "password" not in text.lower()
+    assert "credential" not in text.lower()
+
+
+def test_device_asset_catalog_rejects_unbounded_or_malformed_pagination(
+    development_client,
+):
+    client, _service = development_client
+    endpoint = "/api/development/v1/device-assets"
+
+    malformed = client.get(
+        f"{endpoint}?page=not-a-number",
+        headers={"Authorization": "Bearer viewer"},
+    )
+    oversized = client.get(
+        f"{endpoint}?page_size=101",
+        headers={"Authorization": "Bearer viewer"},
+    )
+
+    assert malformed.status_code == 422
+    assert oversized.status_code == 422
+    assert malformed.get_json()["error"]["code"] == "DEVICE_ASSET_INVALID"

tests/data_research/test_device_assets.py

@@ -0,0 +1,319 @@
+from __future__ import annotations
+
+from dataclasses import replace
+from datetime import datetime, timedelta
+
+import pytest
+
+
+SOURCE_UID = "00000000-0000-0000-0000-000000000101"
+
+
+class MemoryDeviceAssetRepository:
+    def __init__(self, *, active_sources=(SOURCE_UID,)):
+        self.active_sources = set(active_sources)
+        self.assets = {}
+        self.mappings = {}
+        self.versions = {}
+
+    def source_is_active(self, source_uid):
+        return source_uid in self.active_sources
+
+    @staticmethod
+    def _mapping_key(source_uid, source_entity, asset_type, source_code):
+        return source_uid, source_entity, asset_type, source_code
+
+    def find_mapping(
+        self,
+        source_uid,
+        source_entity,
+        asset_type,
+        source_code,
+        *,
+        for_update=False,
+    ):
+        del for_update
+        return self.mappings.get(
+            self._mapping_key(
+                source_uid,
+                source_entity,
+                asset_type,
+                source_code,
+            )
+        )
+
+    def get(self, asset_uid):
+        return self.assets.get(asset_uid)
+
+    def create_asset(self, asset, mapping, version):
+        self.assets[asset.uid] = asset
+        self.mappings[
+            self._mapping_key(
+                mapping.source_uid,
+                mapping.source_entity,
+                mapping.asset_type,
+                mapping.source_code,
+            )
+        ] = mapping
+        self.versions[asset.uid] = [version]
+        return asset, mapping
+
+    def update_asset(self, asset, mapping, version):
+        self.assets[asset.uid] = asset
+        self.mappings[
+            self._mapping_key(
+                mapping.source_uid,
+                mapping.source_entity,
+                mapping.asset_type,
+                mapping.source_code,
+            )
+        ] = mapping
+        self.versions[asset.uid].append(version)
+        return asset, mapping
+
+    def touch_mapping(self, mapping, *, source_updated_at, last_seen_at):
+        updated = replace(
+            mapping,
+            source_updated_at=source_updated_at,
+            last_seen_at=last_seen_at,
+        )
+        self.mappings[
+            self._mapping_key(
+                mapping.source_uid,
+                mapping.source_entity,
+                mapping.asset_type,
+                mapping.source_code,
+            )
+        ] = updated
+        return updated
+
+    def search(self, filters, *, page, page_size):
+        del filters
+        records = list(self.assets.values())
+        start = (page - 1) * page_size
+        return records[start : start + page_size], len(records)
+
+    def list_mappings(self, asset_uid):
+        return [
+            mapping
+            for mapping in self.mappings.values()
+            if mapping.asset_uid == asset_uid
+        ]
+
+    def list_versions(self, asset_uid):
+        return list(self.versions.get(asset_uid, ()))
+
+
+def payload(**record_overrides):
+    record = {
+        "asset_type": "device",
+        "source_code": "EQ-001",
+        "name": "一号循环泵",
+        "location": "一号车间",
+        "organization": "动力部",
+        "responsible_person": "设备管理员",
+        "status": "active",
+        "source_updated_at": "2026-07-29T08:00:00+08:00",
+        "attributes": {"model": "P-100", "criticality": "A"},
+    }
+    record.update(record_overrides)
+    return {
+        "source_uid": SOURCE_UID,
+        "source_entity": "asset.equipment",
+        "records": [record],
+    }
+
+
+def service(repository=None):
+    from app.core.data_research.device_assets import DeviceAssetService
+
+    ids = iter(
+        (
+            "00000000-0000-7000-8000-000000000201",
+            "00000000-0000-7000-8000-000000000202",
+            "00000000-0000-7000-8000-000000000203",
+            "00000000-0000-7000-8000-000000000204",
+            "00000000-0000-7000-8000-000000000205",
+            "00000000-0000-7000-8000-000000000206",
+        )
+    )
+    clock = iter(
+        (
+            datetime(2026, 7, 29, 8, 1),
+            datetime(2026, 7, 29, 8, 2),
+            datetime(2026, 7, 29, 8, 3),
+        )
+    )
+    return DeviceAssetService(
+        repository or MemoryDeviceAssetRepository(),
+        uid_factory=ids.__next__,
+        now_factory=clock.__next__,
+    )
+
+
+def test_reimport_preserves_stable_uid_and_does_not_version_unchanged_data():
+    repository = MemoryDeviceAssetRepository()
+    assets = service(repository)
+
+    first = assets.import_records(payload(), actor_uid="editor-1")
+    second = assets.import_records(
+        payload(source_updated_at="2026-07-29T09:00:00+08:00"),
+        actor_uid="editor-1",
+    )
+
+    assert first.created_count == 1
+    assert first.updated_count == 0
+    assert second.unchanged_count == 1
+    assert second.items[0].asset.uid == first.items[0].asset.uid
+    assert second.items[0].asset.current_version == 1
+    assert len(repository.versions[first.items[0].asset.uid]) == 1
+    assert second.items[0].mapping.source_updated_at.isoformat() == (
+        "2026-07-29T09:00:00+08:00"
+    )
+
+
+def test_changed_normalized_record_creates_one_next_immutable_version():
+    repository = MemoryDeviceAssetRepository()
+    assets = service(repository)
+
+    first = assets.import_records(payload(), actor_uid="editor-1")
+    changed = assets.import_records(
+        payload(location="二号车间"),
+        actor_uid="editor-2",
+    )
+
+    asset_uid = first.items[0].asset.uid
+    assert changed.updated_count == 1
+    assert changed.items[0].asset.uid == asset_uid
+    assert changed.items[0].asset.current_version == 2
+    assert [item.version for item in repository.versions[asset_uid]] == [1, 2]
+    assert repository.versions[asset_uid][0].snapshot["location"] == "一号车间"
+    assert repository.versions[asset_uid][1].snapshot["location"] == "二号车间"
+    assert repository.versions[asset_uid][1].actor_uid == "editor-2"
+
+
+@pytest.mark.parametrize(
+    "asset_type",
+    (
+        "device",
+        "component",
+        "measurement_point",
+        "alarm",
+        "maintenance_record",
+    ),
+)
+def test_import_accepts_each_wp04_asset_type(asset_type):
+    result = service().import_records(
+        payload(asset_type=asset_type),
+        actor_uid="editor-1",
+    )
+
+    assert result.items[0].asset.asset_type == asset_type
+
+
+@pytest.mark.parametrize(
+    ("bad_payload", "message"),
+    (
+        ({"source_uid": "missing", "source_entity": "asset.equipment", "records": []}, "source"),
+        (payload(asset_type="report"), "asset_type"),
+        (payload(source_code=""), "source_code"),
+        (payload(name=""), "name"),
+        (payload(uid="caller-selected"), "uid"),
+        (payload(attributes={"password": "secret"}), "secret"),
+        (payload(attributes={"nested": {"api_key": "secret"}}), "secret"),
+    ),
+)
+def test_import_rejects_invalid_or_secret_bearing_records(
+    bad_payload,
+    message,
+):
+    from app.core.data_research.errors import DeviceAssetInvalid
+
+    with pytest.raises(DeviceAssetInvalid, match=message):
+        service().import_records(bad_payload, actor_uid="editor-1")
+
+
+def test_import_rejects_inactive_source_duplicate_identity_and_oversized_batch():
+    from app.core.data_research.errors import DeviceAssetInvalid
+
+    inactive = service(MemoryDeviceAssetRepository(active_sources=()))
+    with pytest.raises(DeviceAssetInvalid, match="active"):
+        inactive.import_records(payload(), actor_uid="editor-1")
+
+    duplicate = payload()
+    duplicate["records"].append(dict(duplicate["records"][0]))
+    with pytest.raises(DeviceAssetInvalid, match="duplicate"):
+        service().import_records(duplicate, actor_uid="editor-1")
+
+    oversized = payload()
+    oversized["records"] = [
+        {
+            **oversized["records"][0],
+            "source_code": f"EQ-{index:04d}",
+        }
+        for index in range(501)
+    ]
+    with pytest.raises(DeviceAssetInvalid, match="500"):
+        service().import_records(oversized, actor_uid="editor-1")
+
+
+def test_import_rejects_attributes_larger_than_the_catalog_boundary():
+    from app.core.data_research.errors import DeviceAssetInvalid
+
+    with pytest.raises(DeviceAssetInvalid, match="attributes"):
+        service().import_records(
+            payload(attributes={"notes": "x" * 65536}),
+            actor_uid="editor-1",
+        )
+
+
+def test_search_returns_asset_details_with_source_mappings():
+    repository = MemoryDeviceAssetRepository()
+    assets = service(repository)
+    imported = assets.import_records(payload(), actor_uid="editor-1")
+
+    records, total = assets.search(
+        {"keyword": "循环泵"},
+        page=1,
+        page_size=20,
+    )
+
+    assert total == 1
+    assert records[0].asset.uid == imported.items[0].asset.uid
+    assert records[0].mappings == (imported.items[0].mapping,)
+
+
+def test_default_catalog_timestamps_keep_the_china_timezone():
+    from app.core.data_research.device_assets import DeviceAssetService
+
+    ids = iter(
+        (
+            "00000000-0000-7000-8000-000000000401",
+            "00000000-0000-7000-8000-000000000402",
+            "00000000-0000-7000-8000-000000000403",
+        )
+    )
+    result = DeviceAssetService(
+        MemoryDeviceAssetRepository(),
+        uid_factory=ids.__next__,
+    ).import_records(payload(), actor_uid="editor-1")
+
+    created_at = result.items[0].asset.created_at
+    assert created_at.tzinfo is not None
+    assert created_at.utcoffset() == timedelta(hours=8)
+
+
+@pytest.mark.parametrize(
+    "filters",
+    (
+        {"source_uid": "not-a-uuid"},
+        {"asset_type": "report"},
+        {"status": "deleted"},
+        {"keyword": "x" * 301},
+    ),
+)
+def test_search_rejects_invalid_catalog_filters(filters):
+    from app.core.data_research.errors import DeviceAssetInvalid
+
+    with pytest.raises(DeviceAssetInvalid):
+        service().search(filters, page=1, page_size=20)

tests/integration/test_device_asset_postgres.py

@@ -0,0 +1,195 @@
+from __future__ import annotations
+
+import os
+import uuid
+from datetime import datetime
+
+import pytest
+
+
+pytestmark = pytest.mark.integration
+
+
+def test_device_assets_keep_source_identity_versions_and_search(
+    monkeypatch,
+):
+    platform_url = os.environ.get("TEST_DATABASE_URL")
+    if not platform_url:
+        pytest.skip("TEST_DATABASE_URL is required")
+
+    monkeypatch.setenv("DATABASE_URL", platform_url)
+    from app import create_app, db
+    from app.core.data_research.device_asset_repository import (
+        SqlAlchemyDeviceAssetRepository,
+    )
+    from app.core.data_research.device_assets import DeviceAssetService
+    from app.models.data_research import (
+        DeviceAsset,
+        DeviceAssetSourceMapping,
+        DeviceAssetVersion,
+        IngestionSource,
+    )
+
+    app = create_app()
+    app.config.update(TESTING=True)
+    source_uid = str(uuid.uuid4())
+    asset_uids = []
+    try:
+        with app.app_context():
+            db.session.add(
+                IngestionSource(
+                    uid=source_uid,
+                    source_type="database",
+                    name="设备资产集成验收源",
+                    config={
+                        "database_type": "postgresql",
+                        "database": "acceptance",
+                        "schema": "asset",
+                    },
+                    permission_scope={},
+                    status="active",
+                    created_by="integration-test",
+                )
+            )
+            db.session.commit()
+            repository = SqlAlchemyDeviceAssetRepository(db.session)
+            service = DeviceAssetService(
+                repository,
+                commit=db.session.commit,
+                rollback=db.session.rollback,
+            )
+
+            created = service.import_records(
+                {
+                    "source_uid": source_uid,
+                    "source_entity": "asset.equipment",
+                    "records": [
+                        {
+                            "asset_type": "device",
+                            "source_code": "EQ-PG-001",
+                            "name": "循环水泵 A",
+                            "location": "动力车间",
+                            "organization": "设备动力部",
+                            "responsible_person": "张工",
+                            "source_updated_at": (
+                                "2026-07-29T08:00:00+08:00"
+                            ),
+                            "attributes": {"model": "P-100"},
+                        },
+                        {
+                            "asset_type": "component",
+                            "source_code": "PART-PG-001",
+                            "name": "循环水泵 A 轴承",
+                            "location": "动力车间",
+                            "organization": "设备动力部",
+                            "responsible_person": "李工",
+                            "attributes": {"parent_source_code": "EQ-PG-001"},
+                        },
+                    ],
+                },
+                actor_uid="integration-test",
+            )
+            asset_uids = [item.asset.uid for item in created.items]
+            stable_uid = created.items[0].asset.uid
+
+            unchanged = service.import_records(
+                {
+                    "source_uid": source_uid,
+                    "source_entity": "asset.equipment",
+                    "records": [
+                        {
+                            "asset_type": "device",
+                            "source_code": "EQ-PG-001",
+                            "name": "循环水泵 A",
+                            "location": "动力车间",
+                            "organization": "设备动力部",
+                            "responsible_person": "张工",
+                            "source_updated_at": (
+                                "2026-07-29T09:00:00+08:00"
+                            ),
+                            "attributes": {"model": "P-100"},
+                        }
+                    ],
+                },
+                actor_uid="integration-test",
+            )
+            updated = service.import_records(
+                {
+                    "source_uid": source_uid,
+                    "source_entity": "asset.equipment",
+                    "records": [
+                        {
+                            "asset_type": "device",
+                            "source_code": "EQ-PG-001",
+                            "name": "循环水泵 A",
+                            "location": "二号动力车间",
+                            "organization": "设备动力部",
+                            "responsible_person": "张工",
+                            "source_updated_at": (
+                                "2026-07-29T10:00:00+08:00"
+                            ),
+                            "attributes": {"model": "P-100"},
+                        }
+                    ],
+                },
+                actor_uid="integration-test-2",
+            )
+
+            assert unchanged.unchanged_count == 1
+            assert updated.updated_count == 1
+            assert updated.items[0].asset.uid == stable_uid
+            assert updated.items[0].asset.current_version == 2
+
+            for keyword in (
+                "循环水泵",
+                "EQ-PG-001",
+                "二号动力车间",
+                "张工",
+            ):
+                rows, total = repository.search(
+                    {"keyword": keyword},
+                    page=1,
+                    page_size=20,
+                )
+                assert total >= 1
+                assert stable_uid in {row.uid for row in rows}
+
+            filtered, total = repository.search(
+                {
+                    "asset_type": "device",
+                    "status": "active",
+                    "source_uid": source_uid,
+                },
+                page=1,
+                page_size=20,
+            )
+            assert total == 1
+            assert filtered[0].uid == stable_uid
+
+            detail = service.get(stable_uid)
+            versions = service.versions(stable_uid)
+            assert detail.asset.current_version == 2
+            assert len(detail.mappings) == 1
+            assert detail.mappings[0].source_code == "EQ-PG-001"
+            assert detail.mappings[0].source_updated_at == (
+                datetime.fromisoformat("2026-07-29T10:00:00+08:00")
+            )
+            assert [item.version for item in versions] == [2, 1]
+            assert versions[0].snapshot["location"] == "二号动力车间"
+            assert versions[1].snapshot["location"] == "动力车间"
+    finally:
+        with app.app_context():
+            if asset_uids:
+                db.session.query(DeviceAssetVersion).filter(
+                    DeviceAssetVersion.asset_uid.in_(asset_uids)
+                ).delete(synchronize_session=False)
+                db.session.query(DeviceAssetSourceMapping).filter(
+                    DeviceAssetSourceMapping.asset_uid.in_(asset_uids)
+                ).delete(synchronize_session=False)
+                db.session.query(DeviceAsset).filter(
+                    DeviceAsset.uid.in_(asset_uids)
+                ).delete(synchronize_session=False)
+            db.session.query(IngestionSource).filter_by(
+                uid=source_uid
+            ).delete(synchronize_session=False)
+            db.session.commit()

tests/test_database_migrations.py

@@ -52,6 +52,9 @@ EXPECTED_UPGRADED_TABLES = {
     "governance_responsibility_assignments",
     "governance_responsibility_audit_events",
     "catalog_snapshots",
+    "device_assets",
+    "device_asset_source_mappings",
+    "device_asset_versions",
 }
 
 
@@ -96,6 +99,32 @@ def test_catalog_execution_migration_adds_attempts_and_snapshots():
     assert "DROP TABLE" not in migration.upper()
 
 
+def test_device_asset_catalog_migration_is_versioned_and_traceable():
+    migration = (
+        ROOT
+        / "migrations"
+        / "versions"
+        / "20260729_290_device_asset_catalog.py"
+    ).read_text(encoding="utf-8")
+
+    assert 'revision = "20260729_290"' in migration
+    assert 'down_revision = "20260729_280"' in migration
+    for table in (
+        "device_assets",
+        "device_asset_source_mappings",
+        "device_asset_versions",
+    ):
+        assert f"CREATE TABLE public.{table}" in migration
+    assert (
+        "UNIQUE (source_uid, source_entity, asset_type, source_code)"
+        in migration
+    )
+    assert "UNIQUE (asset_uid, version)" in migration
+    assert "source_updated_at" in migration
+    assert "snapshot JSONB NOT NULL" in migration
+    assert "DROP TABLE" not in migration.upper()
+
+
 def test_data_element_migration_adds_versioned_governance_tables():
     migration = (
         ROOT

tests/test_permission_matrix.py

@@ -19,6 +19,7 @@ def test_fixed_role_permission_matrix_is_monotonic():
     assert "data-elements:publish" in admin
     assert "ontologies:publish" in admin
     assert "governance:responsibilities:manage" in admin
+    assert "device-assets:edit" in editor
 
 
 def test_data_development_paths_have_specific_write_policies():
@@ -51,6 +52,15 @@ def test_data_development_paths_have_specific_write_policies():
     assert permission_for_request(
         "/api/development/v1/candidate-decisions", "POST"
     ) == ("data-elements:edit",)
+    assert permission_for_request(
+        "/api/development/v1/device-assets", "GET"
+    ) == ("governance:read",)
+    assert permission_for_request(
+        "/api/development/v1/device-assets/asset-1", "GET"
+    ) == ("governance:read",)
+    assert permission_for_request(
+        "/api/development/v1/device-assets/import", "POST"
+    ) == ("device-assets:edit",)
 
 
 def test_business_domain_read_endpoints_are_available_to_viewers():