diff --git a/netbox_custom_objects/__init__.py b/netbox_custom_objects/__init__.py
index af41097c..d0a58ddf 100644
--- a/netbox_custom_objects/__init__.py
+++ b/netbox_custom_objects/__init__.py
@@ -19,6 +19,14 @@
 _migrations_checked = None
 _checking_migrations = False
 
+# Set to True once ready() has completed and _model_cache is fully populated.
+# get_models() checks this flag and skips dynamic model generation until it's True,
+# preventing ContentType lookups from firing during other apps' ready() calls (e.g.
+# dcim.ready() triggers Device._meta._relation_tree → apps.get_models()).  After
+# ready() sets this flag it calls apps.clear_cache(), so the next _relation_tree
+# access recomputes with the full set of COT models.
+_app_ready = False
+
 
 def _migration_started(sender, **kwargs):
     """Signal handler for pre_migrate - sets the migration flag."""
@@ -209,6 +217,17 @@ def ready(self):
                 super().ready()
                 return
 
+        # Signal that ready() has fully completed.  get_models() checks this flag
+        # before attempting dynamic model generation so that early calls triggered
+        # by other apps' ready() (e.g. dcim.ready() → Device._meta._relation_tree
+        # → apps.get_models()) return only static models rather than crashing on
+        # ContentType lookups.  We call apps.clear_cache() so the next
+        # _relation_tree access recomputes with the full COT model set.
+        global _app_ready
+        _app_ready = True
+        from django.apps import apps as django_apps
+        django_apps.clear_cache()
+
         super().ready()
 
     def get_model(self, model_name, require_ready=True):
@@ -266,6 +285,16 @@ def get_models(self, include_auto_created=False, include_swapped=False):
                 "ignore", category=UserWarning, message=".*database.*"
             )
 
+            # Skip dynamic model generation until ready() has completed.
+            # Other apps' ready() calls (e.g. dcim) trigger _relation_tree →
+            # apps.get_models() before our ready() runs.  At that point _model_cache
+            # is empty, so get_model() would regenerate every COT from scratch —
+            # including ContentType DB lookups that may fail.  After our ready()
+            # finishes, _app_ready is True and get_model() returns cached models
+            # without any ContentType lookups.
+            if not _app_ready:
+                return
+
             # Skip custom object type model loading if dynamic models can't be created yet
             if self.should_skip_dynamic_model_creation():
                 return
diff --git a/netbox_custom_objects/api/serializers.py b/netbox_custom_objects/api/serializers.py
index 132fde4a..4cb3fc0d 100644
--- a/netbox_custom_objects/api/serializers.py
+++ b/netbox_custom_objects/api/serializers.py
@@ -264,6 +264,14 @@ def get_field_data(self, obj):
 def get_serializer_class(model, skip_object_fields=False):
     model_fields = model.custom_object_type.fields.all()
 
+    # Fields skipped during model generation (e.g. broken/null related_object_type_id)
+    # won't be present on the model.  Build a name set from safe list attributes so
+    # we can skip absent fields consistently in both loops below.
+    model_field_names = {
+        f.name
+        for f in list(model._meta.local_fields) + list(model._meta.local_many_to_many)
+    }
+
     # Create field list including all necessary fields
     base_fields = ["id", "url", "display", "created", "last_updated", "tags"]
 
@@ -275,6 +283,8 @@ def get_serializer_class(model, skip_object_fields=False):
     # Only include custom field names that will actually be added to the serializer
     custom_field_names = []
     for field in model_fields:
+        if field.name not in model_field_names:
+            continue  # excluded during model generation (e.g. broken FK)
         if skip_object_fields and field.type in [
             CustomFieldTypeChoices.TYPE_OBJECT, CustomFieldTypeChoices.TYPE_MULTIOBJECT
         ]:
@@ -383,6 +393,8 @@ def update(self, instance, validated_data):
         attrs["get__context"] = get__context
 
     for field in model_fields:
+        if field.name not in model_field_names:
+            continue  # excluded during model generation (e.g. broken FK)
         if skip_object_fields and field.type in [
             CustomFieldTypeChoices.TYPE_OBJECT, CustomFieldTypeChoices.TYPE_MULTIOBJECT
         ]:
diff --git a/netbox_custom_objects/api/urls.py b/netbox_custom_objects/api/urls.py
index 374473e5..44c3bde6 100644
--- a/netbox_custom_objects/api/urls.py
+++ b/netbox_custom_objects/api/urls.py
@@ -90,6 +90,8 @@ def get(self, request, *args, **kwargs):
 urlpatterns = [
     path("", include(router.urls)),
     path("linked-objects/", views.LinkedObjectsView.as_view(), name="linked-objects"),
+    path("schema/preview/", views.SchemaPreviewView.as_view(), name="schema-preview"),
+    path("schema/apply/", views.SchemaApplyView.as_view(), name="schema-apply"),
     path("<str:custom_object_type>/", custom_object_list, name="customobject-list"),
     path(
         "<str:custom_object_type>/<int:pk>/",
diff --git a/netbox_custom_objects/api/views.py b/netbox_custom_objects/api/views.py
index 3ce1107d..c046fae8 100644
--- a/netbox_custom_objects/api/views.py
+++ b/netbox_custom_objects/api/views.py
@@ -1,8 +1,13 @@
+import json
+import logging
+from pathlib import Path
+
 from django.contrib.contenttypes.models import ContentType
 from django.http import Http404
 from django.utils.translation import gettext_lazy as _
 from drf_spectacular.utils import extend_schema_view, extend_schema
 from extras.choices import CustomFieldTypeChoices
+from rest_framework import status
 try:
     from netbox.api.viewsets import ETagMixin  # NetBox 4.6+
 except ImportError:
@@ -13,7 +18,9 @@ class ETagMixin:  # pragma: no cover – NetBox < 4.6 shim
 from rest_framework.routers import APIRootView
 from rest_framework.views import APIView
 from rest_framework.viewsets import ModelViewSet
-from rest_framework.exceptions import ValidationError
+from rest_framework.exceptions import PermissionDenied, ValidationError
+
+from netbox.api.authentication import IsAuthenticatedOrLoginNotRequired, TokenWritePermission
 
 from netbox_custom_objects.filtersets import get_filterset_class
 from netbox_custom_objects.models import CustomObjectType, CustomObjectTypeField
@@ -21,6 +28,73 @@ class ETagMixin:  # pragma: no cover – NetBox < 4.6 shim
 
 from . import serializers
 
+logger = logging.getLogger(__name__)
+
+# ---------------------------------------------------------------------------
+# Schema document helpers
+# ---------------------------------------------------------------------------
+
+_SCHEMA_FILE = Path(__file__).parent.parent / "schema" / "cot_schema_v1.json"
+
+try:
+    import jsonschema as _jsonschema
+    with open(_SCHEMA_FILE) as _f:
+        _COT_SCHEMA = json.load(_f)
+    _VALIDATOR = _jsonschema.Draft202012Validator(_COT_SCHEMA)
+    _HAS_JSONSCHEMA = True
+except (ImportError, OSError, json.JSONDecodeError) as exc:
+    logger.warning("COT schema validation unavailable: %s", exc)
+    _HAS_JSONSCHEMA = False
+    _VALIDATOR = None
+
+
+def _validate_schema_doc(schema_doc: dict) -> None:
+    """
+    Validate *schema_doc* against the COT schema v1 JSON Schema.
+
+    Raises ``ValidationError`` (DRF 400) if validation fails or if
+    ``jsonschema`` is not installed.
+    """
+    if not _HAS_JSONSCHEMA:
+        # Can't validate without jsonschema; allow the request through and
+        # let the comparator / executor surface any structural problems.
+        return
+    errors = sorted(_VALIDATOR.iter_errors(schema_doc), key=lambda e: list(e.path))
+    if errors:
+        raise ValidationError({
+            "schema_errors": [
+                {"path": list(e.path), "message": e.message}
+                for e in errors[:10]  # cap at 10 to avoid overwhelming responses
+            ]
+        })
+
+
+def _serialize_field_change(fc) -> dict:
+    result = {
+        "op": fc.op.value,
+        "schema_id": fc.schema_id,
+        "db_name": fc.db_name,
+        "schema_def": fc.schema_def,
+    }
+    if fc.changed_attrs:
+        # Tuples are not JSON-serialisable; convert to lists.
+        result["changed_attrs"] = {k: list(v) for k, v in fc.changed_attrs.items()}
+    return result
+
+
+def _serialize_diff(diff) -> dict:
+    return {
+        "slug": diff.slug,
+        "name": diff.name,
+        "is_new": diff.is_new,
+        "has_changes": diff.has_changes,
+        "has_destructive_changes": diff.has_destructive_changes,
+        "cot_changes": {k: list(v) for k, v in diff.cot_changes.items()},
+        "field_changes": [_serialize_field_change(fc) for fc in diff.field_changes],
+        "warnings": diff.warnings,
+    }
+
+
 # Constants
 BRANCH_ACTIVE_ERROR_MESSAGE = _("Please switch to the main branch to perform this operation.")
 
@@ -190,7 +264,7 @@ def get(self, request, *args, **kwargs):
 
         results = []
         for field in fields:
-            custom_object_model = field.custom_object_type.get_model(no_cache=True)
+            custom_object_model = field.custom_object_type.get_model()
 
             if field.type == CustomFieldTypeChoices.TYPE_MULTIOBJECT:
                 m2m_field = custom_object_model._meta.get_field(field.name)
@@ -213,3 +287,170 @@ def get(self, request, *args, **kwargs):
                 })
 
         return Response({'count': len(results), 'results': results})
+
+
+class SchemaPreviewView(APIView):
+    """
+    Preview the diff that would result from applying a COT schema document.
+
+    Accepts a ``POST`` request whose body is a schema document conforming to
+    ``cot_schema_v1.json``.  Returns a structured diff for every COT in the
+    document without making any DB changes.
+
+    ## Request body
+
+        {
+            "schema_version": "1",
+            "types": [ ... ]
+        }
+
+    ## Response (200)
+
+        {
+            "diffs": [
+                {
+                    "slug":                   "my-cot",
+                    "name":                   "my_cot",
+                    "is_new":                 false,
+                    "has_changes":            true,
+                    "has_destructive_changes": false,
+                    "cot_changes":            {"description": ["old", "new"]},
+                    "field_changes": [
+                        {
+                            "op":         "add",
+                            "schema_id":  5,
+                            "db_name":    null,
+                            "schema_def": { ... }
+                        }
+                    ],
+                    "warnings": []
+                }
+            ]
+        }
+    """
+
+    permission_classes = [IsAuthenticatedOrLoginNotRequired]
+
+    def post(self, request, *args, **kwargs):
+        # Deferred import: the schema package imports Django models at module level, which
+        # triggers app-registry access before it is ready if imported at the top of views.py.
+        from netbox_custom_objects.schema.comparator import diff_document  # noqa: PLC0415
+
+        schema_doc = request.data
+        _validate_schema_doc(schema_doc)
+        diffs = diff_document(schema_doc)
+        return Response({"diffs": [_serialize_diff(d) for d in diffs]})
+
+
+class SchemaApplyView(APIView):
+    """
+    Apply a COT schema document to the live DB.
+
+    Accepts a ``POST`` request whose body wraps a schema document with an
+    optional ``allow_destructive`` flag.  The document is diffed against the
+    current DB state and all changes are applied atomically.  The applied
+    diffs are returned in the response.
+
+    ## Request body
+
+        {
+            "allow_destructive": false,
+            "schema": {
+                "schema_version": "1",
+                "types": [ ... ]
+            }
+        }
+
+    ``allow_destructive`` defaults to ``false``.  Set it to ``true`` to
+    permit ``REMOVE`` field operations (which drop DB columns).
+
+    ## Response (200)
+
+        {
+            "applied": true,
+            "diffs": [ ... ]
+        }
+
+    ## Error responses
+
+    **409 Conflict** — the document contains ``REMOVE`` operations and
+    ``allow_destructive`` was not set:
+
+        {
+            "error":             "destructive_changes",
+            "detail":            "Schema contains destructive ...",
+            "destructive_slugs": ["my-cot"]
+        }
+
+    **400 Bad Request** — circular COT dependency, unresolvable FK target,
+    or invalid schema document structure.
+
+    Unexpected DB errors (e.g. ``IntegrityError`` from a constraint violation
+    unrelated to the COT schema logic) are not caught and will surface as
+    **500 Internal Server Error**.  The entire apply is wrapped in
+    ``transaction.atomic()``, so any such failure leaves the DB unchanged.
+    """
+
+    permission_classes = [IsAuthenticatedOrLoginNotRequired, TokenWritePermission]
+
+    def post(self, request, *args, **kwargs):
+        # TODO: Schema apply is blocked while in a branch context because the executor
+        # performs direct DDL (ALTER/DROP TABLE) that is not branch-aware.  When branching
+        # is extended to support schema operations, remove this guard and wire up the
+        # appropriate branch-scoped apply path.
+        if is_in_branch():
+            raise ValidationError(BRANCH_ACTIVE_ERROR_MESSAGE)
+
+        if not (
+            request.user.has_perm('netbox_custom_objects.add_customobjecttype') and
+            request.user.has_perm('netbox_custom_objects.change_customobjecttype')
+        ):
+            raise PermissionDenied(
+                "You do not have permission to apply a schema document. "
+                "Both add and change permissions on CustomObjectType are required."
+            )
+
+        # Deferred import: same app-registry concern as the comparator import above.
+        from netbox_custom_objects.schema.executor import (  # noqa: PLC0415
+            apply_document,
+            CircularDependencyError,
+            DestructiveChangesError,
+            UnknownChoiceSetError,
+            UnknownFieldTypeError,
+            UnknownObjectTypeError,
+        )
+
+        allow_destructive = request.data.get("allow_destructive", False)
+        if not isinstance(allow_destructive, bool):
+            raise ValidationError({"allow_destructive": _("'allow_destructive' must be a boolean.")})
+        schema_doc = request.data.get("schema")
+        if not isinstance(schema_doc, dict):
+            raise ValidationError(
+                {"schema": _("A 'schema' object containing the COT schema document is required.")}
+            )
+
+        _validate_schema_doc(schema_doc)
+
+        try:
+            diffs = apply_document(schema_doc, allow_destructive=allow_destructive)
+        except DestructiveChangesError as exc:
+            return Response(
+                {
+                    "error": "destructive_changes",
+                    "detail": str(exc),
+                    "destructive_slugs": [d.slug for d in exc.diffs],
+                },
+                status=status.HTTP_409_CONFLICT,
+            )
+        except CircularDependencyError as exc:
+            return Response(
+                {"error": "circular_dependency", "detail": str(exc)},
+                status=status.HTTP_400_BAD_REQUEST,
+            )
+        except (UnknownChoiceSetError, UnknownFieldTypeError, UnknownObjectTypeError) as exc:
+            return Response(
+                {"error": "unresolvable_reference", "detail": str(exc)},
+                status=status.HTTP_400_BAD_REQUEST,
+            )
+
+        return Response({"applied": True, "diffs": [_serialize_diff(d) for d in diffs]})
diff --git a/netbox_custom_objects/comparator.py b/netbox_custom_objects/comparator.py
index 1c36cf3d..31713c12 100644
--- a/netbox_custom_objects/comparator.py
+++ b/netbox_custom_objects/comparator.py
@@ -51,7 +51,7 @@
 
 if TYPE_CHECKING:
     from django.contrib.contenttypes.models import ContentType
-from netbox_custom_objects.schema_format import (
+from netbox_custom_objects.schema.format import (
     CUSTOM_OBJECTS_APP_LABEL_SLUG,
     FIELD_DEFAULTS,
     FIELD_TYPE_ATTRS,
diff --git a/netbox_custom_objects/exporter.py b/netbox_custom_objects/exporter.py
deleted file mode 100644
index e9ec4d71..00000000
--- a/netbox_custom_objects/exporter.py
+++ /dev/null
@@ -1,224 +0,0 @@
-"""
-Exporter for the COT portable schema format (issue #388).
-
-Converts live CustomObjectType DB state into a schema document dict that
-conforms to cot_schema_v1.json.  The returned dict can be serialised to YAML
-or JSON by the caller.
-
-Public API
-----------
-    export_cot(cot)     → dict   # single COT definition (no top-level wrapper)
-    export_cots(cots)   → dict   # full schema document  { schema_version, types }
-
-Notes
------
-- Fields without a schema_id (created before the schema-format feature) are
-  skipped with a WARNING log entry.  They cannot be tracked across installs.
-- Attribute values that equal FIELD_DEFAULTS are omitted to keep the output
-  minimal (round-trip safe: the importer re-applies the same defaults).
-- Tombstones (removed_fields) are read from the COT's schema_document.  Until
-  the apply endpoint (#390) is implemented this will always be empty; once
-  apply is wired up, deletions will be persisted there automatically.
-"""
-
-import logging
-import re
-
-from netbox_custom_objects import constants
-from netbox_custom_objects.schema_format import (
-    CHOICES_TO_SCHEMA_TYPE,
-    CUSTOM_OBJECTS_APP_LABEL_SLUG,
-    FIELD_DEFAULTS,
-    FIELD_TYPE_ATTRS,
-    SCHEMA_FORMAT_VERSION,
-)
-
-logger = logging.getLogger(__name__)
-
-# Matches the generated model name produced by CustomObjectType.get_table_model_name().
-# Capturing group 1 is the numeric COT id.
-_TABLE_MODEL_RE = re.compile(r'^table(\d+)model$', re.IGNORECASE)
-
-# Ordered list of field_base attributes to check for non-default values.
-# Type-specific attributes (validation_*, choice_set, related_*) are handled
-# separately via FIELD_TYPE_ATTRS.
-_BASE_ATTRS = (
-    "label",
-    "description",
-    "group_name",
-    "primary",
-    "required",
-    "unique",
-    "default",
-    "weight",
-    "search_weight",
-    "filter_logic",
-    "ui_visible",
-    "ui_editable",
-    "is_cloneable",
-    "deprecated",
-    "deprecated_since",
-    "scheduled_removal",
-)
-
-
-# ---------------------------------------------------------------------------
-# Internal helpers
-# ---------------------------------------------------------------------------
-
-def _encode_related_object_type(rot) -> str:
-    """
-    Encode an ObjectType FK as a schema ``related_object_type`` string.
-
-    Built-in NetBox objects → ``"<app_label>/<model>"``  (e.g. ``"dcim/device"``)
-    Custom Object Types     → ``"custom-objects/<slug>"``
-    """
-    if rot.app_label == constants.APP_LABEL:
-        m = _TABLE_MODEL_RE.match(rot.model)
-        if m:
-            # Avoid a circular import — import here so the module can be loaded
-            # independently of the full Django app stack in unit tests.
-            from netbox_custom_objects.models import CustomObjectType  # noqa: PLC0415
-            cot_id = int(m.group(1))
-            slug = CustomObjectType.objects.values_list('slug', flat=True).get(pk=cot_id)
-            return f"{CUSTOM_OBJECTS_APP_LABEL_SLUG}/{slug}"
-    return f"{rot.app_label}/{rot.model}"
-
-
-def _export_field(field) -> dict:
-    """
-    Serialise a single ``CustomObjectTypeField`` instance to a schema field dict.
-
-    Raises ``ValueError`` if ``field.schema_id`` is ``None``; callers should
-    pre-filter or handle this case before calling this function.
-    """
-    if field.schema_id is None:
-        raise ValueError(
-            f"Field {field.name!r} on COT {field.custom_object_type_id!r} "
-            "has no schema_id and cannot be exported."
-        )
-
-    schema_type = CHOICES_TO_SCHEMA_TYPE[field.type]
-
-    result = {
-        "id": field.schema_id,
-        "name": field.name,
-        "type": schema_type,
-    }
-
-    # ── Base attributes (omit when equal to documented defaults) ────────────
-    for attr in _BASE_ATTRS:
-        value = getattr(field, attr)
-        if value != FIELD_DEFAULTS.get(attr):
-            result[attr] = value
-
-    # ── Type-specific attributes ─────────────────────────────────────────────
-    for attr in sorted(FIELD_TYPE_ATTRS[schema_type]):
-        if attr == "choice_set":
-            # Required for select/multiselect; validate.
-            if field.choice_set is None:
-                raise ValueError(
-                    f"Field {field.name!r} is type {schema_type!r} but has no choice_set assigned."
-                )
-            result["choice_set"] = field.choice_set.name
-        elif attr == "related_object_type":
-            # Required for object/multiobject; always present.
-            result["related_object_type"] = _encode_related_object_type(
-                field.related_object_type
-            )
-        elif attr == "related_object_filter":
-            value = field.related_object_filter
-            if value != FIELD_DEFAULTS.get("related_object_filter"):
-                result["related_object_filter"] = value
-        elif attr in ("validation_regex", "validation_minimum", "validation_maximum"):
-            value = getattr(field, attr)
-            if value != FIELD_DEFAULTS.get(attr):
-                result[attr] = value
-
-    return result
-
-
-def _removed_fields_from_document(cot) -> list:
-    """
-    Extract the ``removed_fields`` tombstone list for *cot* from its stored
-    ``schema_document``.  Returns an empty list if the document is absent or
-    does not reference this COT.
-    """
-    if not cot.schema_document:
-        return []
-    # NOTE: matches by COT name. If the COT is renamed after tombstones
-    # are persisted, they will not be found. This will be addressed when
-    # #390 (apply) is implemented and tombstones are managed more explicitly.
-    for type_def in cot.schema_document.get("types", []):
-        if type_def.get("name") == cot.name:
-            return list(type_def.get("removed_fields", []))
-    return []
-
-
-# ---------------------------------------------------------------------------
-# Public API
-# ---------------------------------------------------------------------------
-
-def export_cot(cot) -> dict:
-    """
-    Serialise a single ``CustomObjectType`` to its schema definition dict
-    (the inner object that goes inside the ``types`` list).
-
-    Fields without a ``schema_id`` are skipped; a WARNING is logged for each.
-    """
-    result: dict = {
-        "name": cot.name,
-        "slug": cot.slug,
-    }
-
-    # Optional COT-level attributes — omit when blank/unset.
-    if cot.version:
-        result["version"] = cot.version
-    if cot.verbose_name:
-        result["verbose_name"] = cot.verbose_name
-    if cot.verbose_name_plural:
-        result["verbose_name_plural"] = cot.verbose_name_plural
-    if cot.description:
-        result["description"] = cot.description
-    if cot.group_name:
-        result["group_name"] = cot.group_name
-
-    # Active + deprecated fields, ordered by schema_id for stable output.
-    exported_fields = []
-    for field in cot.fields.order_by("schema_id"):
-        if field.schema_id is None:
-            logger.warning(
-                "Skipping field %r on COT %r during export: no schema_id assigned. "
-                "This field was likely created before the schema-format feature was "
-                "introduced and cannot be tracked portably.",
-                field.name,
-                cot.name,
-            )
-            continue
-        exported_fields.append(_export_field(field))
-
-    if exported_fields:
-        result["fields"] = exported_fields
-
-    # Tombstones from previous apply operations.
-    removed = _removed_fields_from_document(cot)
-    if removed:
-        result["removed_fields"] = removed
-
-    return result
-
-
-def export_cots(cots) -> dict:
-    """
-    Serialise one or more ``CustomObjectType`` instances to a complete schema
-    document dict (``{ schema_version, types }``) that validates against
-    ``cot_schema_v1.json``.
-
-    *cots* may be any iterable of ``CustomObjectType`` instances.
-    """
-    if not cots:
-        raise ValueError("Minimum 1 Custom Object Type required.")
-    return {
-        "schema_version": SCHEMA_FORMAT_VERSION,
-        "types": [export_cot(cot) for cot in cots],
-    }
diff --git a/netbox_custom_objects/field_types.py b/netbox_custom_objects/field_types.py
index 3b881f2c..d4632d57 100644
--- a/netbox_custom_objects/field_types.py
+++ b/netbox_custom_objects/field_types.py
@@ -3,7 +3,6 @@
 import django_tables2 as tables
 from django import forms
 from django.apps import apps
-from django.contrib.contenttypes.models import ContentType
 from django.contrib.postgres.fields import ArrayField
 from django.core.validators import RegexValidator
 from django.db import models
@@ -113,6 +112,27 @@ def get_table_column_field(self, field, **kwargs):
     def render_table_column_linkified(self, record):
         return linkify(record)
 
+    def _get_related_content_type(self, field):
+        """
+        Return the ContentType for field.related_object_type_id.
+
+        Raises NotImplementedError (rather than ContentType.DoesNotExist) so that
+        all callers — which already guard against NotImplementedError — skip the
+        field gracefully when the FK is null or its ContentType row is missing.
+        """
+        from django.contrib.contenttypes.models import ContentType as CT
+        if not field.related_object_type_id:
+            raise NotImplementedError(
+                f"Field {field.name!r} has no related_object_type set"
+            )
+        try:
+            return CT.objects.get(pk=field.related_object_type_id)
+        except CT.DoesNotExist:
+            raise NotImplementedError(
+                f"Field {field.name!r}: related_object_type_id="
+                f"{field.related_object_type_id} references a missing ContentType"
+            )
+
     def after_model_generation(self, instance, model, field_name): ...
 
     def create_m2m_table(self, instance, model, field_name): ...
@@ -395,7 +415,7 @@ def render_table_column(self, value):
 
 class ObjectFieldType(FieldType):
     def get_model_field(self, field, **kwargs):
-        content_type = ContentType.objects.get(pk=field.related_object_type_id)
+        content_type = self._get_related_content_type(field)
         to_model = content_type.model
 
         # Extract our custom parameters and keep only Django field parameters
@@ -460,7 +480,7 @@ def get_form_field(self, field, for_csv_import=False, **kwargs):
         For custom objects, uses CustomObjectDynamicModelChoiceField.
         For regular NetBox objects, uses DynamicModelChoiceField.
         """
-        content_type = ContentType.objects.get(pk=field.related_object_type_id)
+        content_type = self._get_related_content_type(field)
 
         has_context = False
         if content_type.app_label == APP_LABEL:
@@ -508,7 +528,7 @@ def get_form_field(self, field, for_csv_import=False, **kwargs):
             return form_field
 
     def get_filterform_field(self, field, **kwargs):
-        content_type = ContentType.objects.get(pk=field.related_object_type_id)
+        content_type = self._get_related_content_type(field)
         if content_type.app_label == APP_LABEL:
             from netbox_custom_objects.models import CustomObjectType
             custom_object_type_id = extract_cot_id_from_model_name(content_type.model)
@@ -531,6 +551,7 @@ def render_table_column(self, value):
         return linkify(value)
 
     def get_serializer_field(self, field, **kwargs):
+        self._get_related_content_type(field)  # validates FK; raises NotImplementedError if null/missing
         related_model_class = field.related_object_type.model_class()
         if related_model_class._meta.app_label == APP_LABEL:
             from netbox_custom_objects.api.serializers import get_serializer_class
@@ -747,7 +768,7 @@ def get_through_model(self, field, model_string):
         )
 
         # Check if this is a self-referential M2M
-        content_type = ContentType.objects.get(pk=field.related_object_type_id)
+        content_type = self._get_related_content_type(field)
         custom_object_type_id = extract_cot_id_from_model_name(content_type.model)
         if content_type.app_label == APP_LABEL:
             if custom_object_type_id is None:
@@ -784,7 +805,7 @@ def get_model_field(self, field, **kwargs):
         Creates the M2M field with appropriate model references
         """
         # Check if this is a self-referential M2M
-        content_type = ContentType.objects.get(pk=field.related_object_type_id)
+        content_type = self._get_related_content_type(field)
         custom_object_type_id = extract_cot_id_from_model_name(content_type.model)
         if content_type.app_label == APP_LABEL:
             if custom_object_type_id is None:
@@ -838,7 +859,7 @@ def get_form_field(self, field, for_csv_import=False, **kwargs):
         Returns a form field for multi-object relationships.
         Uses DynamicModelMultipleChoiceField for both custom objects and regular NetBox objects.
         """
-        content_type = ContentType.objects.get(pk=field.related_object_type_id)
+        content_type = self._get_related_content_type(field)
 
         has_context = False
         if content_type.app_label == APP_LABEL:
@@ -884,7 +905,7 @@ def get_form_field(self, field, for_csv_import=False, **kwargs):
             return form_field
 
     def get_filterform_field(self, field, **kwargs):
-        content_type = ContentType.objects.get(pk=field.related_object_type_id)
+        content_type = self._get_related_content_type(field)
         if content_type.app_label == APP_LABEL:
             from netbox_custom_objects.models import CustomObjectType
             custom_object_type_id = extract_cot_id_from_model_name(content_type.model)
@@ -911,6 +932,7 @@ def get_table_column_field(self, field, **kwargs):
         return tables.ManyToManyColumn(linkify_item=True, orderable=False)
 
     def get_serializer_field(self, field, **kwargs):
+        self._get_related_content_type(field)  # validates FK; raises NotImplementedError if null/missing
         related_model_class = field.related_object_type.model_class()
         if related_model_class._meta.app_label == APP_LABEL:
             from netbox_custom_objects.api.serializers import get_serializer_class
@@ -947,7 +969,7 @@ def after_model_generation(self, instance, model, field_name):
 
         # For non-self-referential fields, we need to resolve the target model
         # Use the instance parameter which contains the field definition
-        content_type = ContentType.objects.get(pk=instance.related_object_type_id)
+        content_type = self._get_related_content_type(instance)
 
         # Now we can safely resolve the target model
         if content_type.app_label == APP_LABEL:
@@ -1000,7 +1022,7 @@ def create_m2m_table(self, instance, model, field_name):
         if getattr(field, "_is_self_referential", False):
             to_model = model
         else:
-            content_type = ContentType.objects.get(pk=instance.related_object_type_id)
+            content_type = self._get_related_content_type(instance)
             if content_type.app_label == APP_LABEL:
                 from netbox_custom_objects.models import CustomObjectType
 
diff --git a/netbox_custom_objects/migrations/0009_alter_customobjecttype_version.py b/netbox_custom_objects/migrations/0009_alter_customobjecttype_version.py
new file mode 100644
index 00000000..5eff7535
--- /dev/null
+++ b/netbox_custom_objects/migrations/0009_alter_customobjecttype_version.py
@@ -0,0 +1,41 @@
+# Generated by Django 5.2.12 on 2026-04-08 01:04
+
+import netbox_custom_objects.models
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ('netbox_custom_objects', '0008_backfill_schema_ids'),
+    ]
+
+    operations = [
+        migrations.AlterField(
+            model_name='customobjecttype',
+            name='version',
+            field=models.CharField(
+                blank=True,
+                max_length=50,
+                validators=[netbox_custom_objects.models.validate_pep440],
+            ),
+        ),
+        migrations.AlterField(
+            model_name='customobjecttypefield',
+            name='deprecated_since',
+            field=models.CharField(
+                blank=True,
+                max_length=50,
+                validators=[netbox_custom_objects.models.validate_pep440],
+            ),
+        ),
+        migrations.AlterField(
+            model_name='customobjecttypefield',
+            name='scheduled_removal',
+            field=models.CharField(
+                blank=True,
+                max_length=50,
+                validators=[netbox_custom_objects.models.validate_pep440],
+            ),
+        ),
+    ]
diff --git a/netbox_custom_objects/models.py b/netbox_custom_objects/models.py
index 0412a1e8..88181491 100644
--- a/netbox_custom_objects/models.py
+++ b/netbox_custom_objects/models.py
@@ -1,8 +1,11 @@
 import decimal
+import logging
 import re
 import threading
 from datetime import date, datetime
 
+from packaging.version import Version, InvalidVersion
+
 import django_filters
 from core.models import ObjectType, ObjectChange
 from core.models.object_types import ObjectTypeManager
@@ -11,7 +14,6 @@
 
 # from django.contrib.contenttypes.management import create_contenttypes
 from django.contrib.contenttypes.models import ContentType
-from django.core.exceptions import FieldDoesNotExist
 from django.core.validators import RegexValidator, ValidationError
 from django.db import connection, IntegrityError, models, transaction
 from django.db.models import Q
@@ -57,6 +59,8 @@
 from netbox_custom_objects.jobs import ReindexCustomObjectTypeJob
 from netbox_custom_objects.utilities import _suppress_clear_cache, generate_model
 
+logger = logging.getLogger(__name__)
+
 
 class UniquenessConstraintTestError(Exception):
     """Custom exception used to signal successful uniqueness constraint test."""
@@ -163,6 +167,19 @@ def _get_action_url(cls, action=None, rest_api=False, kwargs=None):
         return reverse(cls._get_viewname(action, rest_api), kwargs=kwargs)
 
 
+def validate_pep440(value):
+    """Validate that *value* is a valid PEP 440 version string."""
+    if not value:
+        return
+    try:
+        Version(value)
+    except InvalidVersion:
+        raise ValidationError(
+            _("'%(value)s' is not a valid version string (expected e.g. '1.0.0')."),
+            params={"value": value},
+        )
+
+
 class CustomObjectType(NetBoxModel):
     # Class-level cache for generated models
     _model_cache = {}
@@ -193,7 +210,7 @@ class CustomObjectType(NetBoxModel):
         verbose_name=_('comments'),
         blank=True
     )
-    version = models.CharField(max_length=50, blank=True)
+    version = models.CharField(max_length=50, blank=True, validators=[validate_pep440])
     verbose_name = models.CharField(max_length=100, blank=True)
     verbose_name_plural = models.CharField(max_length=100, blank=True)
     slug = models.SlugField(max_length=100, unique=True, db_index=True, blank=False)
@@ -389,9 +406,24 @@ def _fetch_and_generate_field_attrs(
             field_type = FIELD_TYPE_CLASS[field.type]()
             field_name = field.name
 
-            field_attrs[field.name] = field_type.get_model_field(
-                field,
-            )
+            try:
+                field_attrs[field.name] = field_type.get_model_field(
+                    field,
+                )
+            except NotImplementedError:
+                if field.related_object_type_id is None:
+                    logger.debug(
+                        "Skipping field %r (pk=%s) on COT %r: "
+                        "related_object_type_id is NULL — field has no related type set.",
+                        field.name, field.pk, self.slug,
+                    )
+                else:
+                    logger.debug(
+                        "Skipping field %r (pk=%s) on COT %r: related_object_type_id=%s "
+                        "references a ContentType that no longer exists.",
+                        field.name, field.pk, self.slug, field.related_object_type_id,
+                    )
+                continue
 
             # Add to field objects only if the field was successfully generated
             field_attrs["_field_objects"][field.id] = {
@@ -416,6 +448,16 @@ def _after_model_generation(self, attrs, model):
         # Get the set of fields that were skipped due to recursion
         skipped_fields = attrs.get("_skipped_fields", set())
 
+        # Build a lookup from field name → Django field object using plain lists that
+        # don't trigger _relation_tree.  _meta.get_field() for a name that isn't in
+        # _forward_fields_map (e.g. tombstoned fields in _trashed_field_objects) falls
+        # through to fields_map → _relation_tree → apps.get_models() → our get_models()
+        # override → get_model() for every COT → infinite recursion.
+        present_fields = {
+            f.name: f
+            for f in list(model._meta.local_fields) + list(model._meta.local_many_to_many)
+        }
+
         # Collect through models during after_model_generation
         through_models = []
 
@@ -426,28 +468,24 @@ def _after_model_generation(self, attrs, model):
             if field_name in skipped_fields:
                 continue
 
-            # Only process fields that actually exist on the model
-            # Fields might be skipped due to recursion prevention
-            if hasattr(model._meta, 'get_field'):
-                try:
-                    field = model._meta.get_field(field_name)
-                    # Field exists, process it
-                    field_object["type"].after_model_generation(
-                        field_object["field"], model, field_name
-                    )
+            # Only process fields that actually exist on the model.
+            # Tombstoned fields (in _trashed_field_objects) won't be in present_fields.
+            field = present_fields.get(field_name)
+            if field is None:
+                continue
 
-                    # Collect through models from M2M fields
-                    if hasattr(field, 'remote_field') and hasattr(field.remote_field, 'through'):
-                        through_model = field.remote_field.through
-                        # Only collect custom through models, not auto-created Django ones
-                        if (through_model and through_model not in through_models and
-                            hasattr(through_model._meta, 'app_label') and
-                            through_model._meta.app_label == APP_LABEL):
-                            through_models.append(through_model)
-
-                except Exception:
-                    # Field doesn't exist (likely skipped due to recursion), skip processing
-                    continue
+            field_object["type"].after_model_generation(
+                field_object["field"], model, field_name
+            )
+
+            # Collect through models from M2M fields
+            if hasattr(field, 'remote_field') and hasattr(field.remote_field, 'through'):
+                through_model = field.remote_field.through
+                # Only collect custom through models, not auto-created Django ones
+                if (through_model and through_model not in through_models and
+                    hasattr(through_model._meta, 'app_label') and
+                    through_model._meta.app_label == APP_LABEL):
+                    through_models.append(through_model)
 
         # Store through models on the model for yielding in get_models()
         model._through_models = through_models
@@ -483,16 +521,17 @@ def get_content_type_label(custom_object_type_id):
 
     def register_custom_object_search_index(self, model):
         # model must be an instance of this CustomObjectType's get_model() generated class
+        # Use local_fields / local_many_to_many — plain lists populated at class-creation
+        # time — instead of _meta.get_field(), which triggers Django's lazy _relation_tree
+        # computation.  _relation_tree calls apps.get_models(), which re-enters our
+        # get_models() override, which calls get_model() for every COT → infinite recursion.
+        present = (
+            {f.name for f in model._meta.local_fields}
+            | {f.name for f in model._meta.local_many_to_many}
+        )
         fields = []
         for field in self.fields.filter(search_weight__gt=0):
-            # Only index fields that are actually present on the generated model.
-            # When the model was built with skip_object_fields=True (a lightweight
-            # stub used to break cross-COT FK recursion), object-type fields are
-            # intentionally absent.  Including them in the index causes
-            # FieldDoesNotExist / AttributeError in the post_save search handler.
-            try:
-                model._meta.get_field(field.name)
-            except FieldDoesNotExist:
+            if field.name not in present:
                 continue
             fields.append((field.name, field.search_weight))
 
@@ -1006,12 +1045,14 @@ class CustomObjectTypeField(CloningMixin, ExportTemplatesMixin, ChangeLoggedMode
         blank=True,
         verbose_name=_("deprecated since"),
         help_text=_("Schema version in which this field was marked deprecated (e.g. '2.0.0')."),
+        validators=[validate_pep440],
     )
     scheduled_removal = models.CharField(
         max_length=50,
         blank=True,
         verbose_name=_("scheduled removal"),
         help_text=_("Schema version in which this field is planned to be removed (e.g. '3.0.0')."),
+        validators=[validate_pep440],
     )
 
     clone_fields = ("custom_object_type",)
diff --git a/netbox_custom_objects/schema_format.py b/netbox_custom_objects/schema_format.py
deleted file mode 100644
index c3e38e4e..00000000
--- a/netbox_custom_objects/schema_format.py
+++ /dev/null
@@ -1,110 +0,0 @@
-"""
-Constants and helpers for the COT portable schema format.
-
-The schema format is a YAML (or JSON) document describing one or more Custom
-Object Type definitions in a portable, versionable way.  A multi-type export
-always uses a top-level ``types:`` list; the importer also accepts a bare
-single-type document for convenience.
-
-Format version history
-----------------------
-"1"  Initial version (introduced alongside schema_id / deprecated field support).
-"""
-
-from extras.choices import CustomFieldTypeChoices
-
-# ── Format version ──────────────────────────────────────────────────────────
-# Bump this only when the format itself changes in a breaking way.
-SCHEMA_FORMAT_VERSION = "1"
-
-# ── Field type names (value → schema string) ────────────────────────────────
-# These are the canonical type names used in schema documents.
-# They happen to match CustomFieldTypeChoices values, but are redefined here
-# explicitly so the schema format is not silently broken by upstream changes.
-FIELD_TYPE_TEXT = "text"
-FIELD_TYPE_LONGTEXT = "longtext"
-FIELD_TYPE_INTEGER = "integer"
-FIELD_TYPE_DECIMAL = "decimal"
-FIELD_TYPE_BOOLEAN = "boolean"
-FIELD_TYPE_DATE = "date"
-FIELD_TYPE_DATETIME = "datetime"
-FIELD_TYPE_URL = "url"
-FIELD_TYPE_JSON = "json"
-FIELD_TYPE_SELECT = "select"
-FIELD_TYPE_MULTISELECT = "multiselect"
-FIELD_TYPE_OBJECT = "object"
-FIELD_TYPE_MULTIOBJECT = "multiobject"
-
-# Mapping from CustomFieldTypeChoices values to schema type names.
-# Used by the exporter; the importer uses the inverse.
-CHOICES_TO_SCHEMA_TYPE = {
-    CustomFieldTypeChoices.TYPE_TEXT: FIELD_TYPE_TEXT,
-    CustomFieldTypeChoices.TYPE_LONGTEXT: FIELD_TYPE_LONGTEXT,
-    CustomFieldTypeChoices.TYPE_INTEGER: FIELD_TYPE_INTEGER,
-    CustomFieldTypeChoices.TYPE_DECIMAL: FIELD_TYPE_DECIMAL,
-    CustomFieldTypeChoices.TYPE_BOOLEAN: FIELD_TYPE_BOOLEAN,
-    CustomFieldTypeChoices.TYPE_DATE: FIELD_TYPE_DATE,
-    CustomFieldTypeChoices.TYPE_DATETIME: FIELD_TYPE_DATETIME,
-    CustomFieldTypeChoices.TYPE_URL: FIELD_TYPE_URL,
-    CustomFieldTypeChoices.TYPE_JSON: FIELD_TYPE_JSON,
-    CustomFieldTypeChoices.TYPE_SELECT: FIELD_TYPE_SELECT,
-    CustomFieldTypeChoices.TYPE_MULTISELECT: FIELD_TYPE_MULTISELECT,
-    CustomFieldTypeChoices.TYPE_OBJECT: FIELD_TYPE_OBJECT,
-    CustomFieldTypeChoices.TYPE_MULTIOBJECT: FIELD_TYPE_MULTIOBJECT,
-}
-
-SCHEMA_TYPE_TO_CHOICES = {v: k for k, v in CHOICES_TO_SCHEMA_TYPE.items()}
-
-# ── related_object_type encoding ─────────────────────────────────────────────
-# Built-in NetBox objects:  "dcim/device"  (app_label/model)
-# Custom Object Types:      "custom-objects/circuit"  (using the COT slug)
-CUSTOM_OBJECTS_APP_LABEL_SLUG = "custom-objects"
-
-# ── Field attribute defaults ─────────────────────────────────────────────────
-# Attributes that match these defaults MAY be omitted from the schema document.
-# The importer applies them when a key is absent.
-FIELD_DEFAULTS = {
-    # label resolves to name.replace("_", " ").capitalize() at runtime. Importer must implement this same logic.
-    # An empty or absent label means "derive from name".
-    "label": "",
-    "description": "",
-    "group_name": "",
-    "primary": False,
-    "required": False,
-    "unique": False,
-    "default": None,
-    "weight": 100,
-    "search_weight": 500,
-    "filter_logic": "loose",
-    "ui_visible": "always",
-    "ui_editable": "yes",
-    "is_cloneable": False,
-    "deprecated": False,
-    "deprecated_since": "",
-    "scheduled_removal": "",
-    # type-specific defaults
-    "validation_regex": "",
-    "validation_minimum": None,
-    "validation_maximum": None,
-    "related_object_filter": None,
-}
-
-# ── Field groups by type ─────────────────────────────────────────────────────
-# Which type-specific attributes are valid for each field type.
-# Used by the exporter to omit irrelevant keys and by the JSON Schema.
-# Note: The exporter should use FIELD_TYPE_ATTRS to drop irrelevant keys inherited from field_base.
-FIELD_TYPE_ATTRS = {
-    FIELD_TYPE_TEXT: {"validation_regex"},
-    FIELD_TYPE_LONGTEXT: {"validation_regex"},
-    FIELD_TYPE_INTEGER: {"validation_minimum", "validation_maximum"},
-    FIELD_TYPE_DECIMAL: {"validation_minimum", "validation_maximum"},
-    FIELD_TYPE_BOOLEAN: set(),
-    FIELD_TYPE_DATE: set(),
-    FIELD_TYPE_DATETIME: set(),
-    FIELD_TYPE_URL: set(),
-    FIELD_TYPE_JSON: set(),
-    FIELD_TYPE_SELECT: {"choice_set"},
-    FIELD_TYPE_MULTISELECT: {"choice_set"},
-    FIELD_TYPE_OBJECT: {"related_object_type", "related_object_filter"},
-    FIELD_TYPE_MULTIOBJECT: {"related_object_type", "related_object_filter"},
-}
diff --git a/netbox_custom_objects/tests/base.py b/netbox_custom_objects/tests/base.py
index ae48fc57..8f4b1770 100644
--- a/netbox_custom_objects/tests/base.py
+++ b/netbox_custom_objects/tests/base.py
@@ -2,11 +2,23 @@
 from django.test import Client
 from core.models import ObjectType
 from extras.models import CustomFieldChoiceSet
+from users.models import Token
 from utilities.testing import create_test_user
 
 from netbox_custom_objects.models import CustomObjectType, CustomObjectTypeField
 
 
+def create_api_token(user):
+    """Create an API token for *user*, handling the NetBox ≥ 4.5 version field."""
+    try:
+        from users.choices import TokenVersionChoices  # noqa: PLC0415
+        token = Token(version=TokenVersionChoices.V1, user=user)
+    except ImportError:
+        token = Token(user=user)
+    token.save()
+    return token
+
+
 class TransactionCleanupMixin:
     """Mixin for TransactionTestCase subclasses that create CustomObjectType instances.
 
diff --git a/netbox_custom_objects/tests/schema/test_schema_api.py b/netbox_custom_objects/tests/schema/test_schema_api.py
new file mode 100644
index 00000000..f4a50388
--- /dev/null
+++ b/netbox_custom_objects/tests/schema/test_schema_api.py
@@ -0,0 +1,395 @@
+"""
+Tests for the schema preview and apply API endpoints (issue #390).
+
+Covers:
+- POST /schema/preview/: returns diffs without modifying the DB
+- POST /schema/apply/: applies the schema document and returns diffs
+- allow_destructive flag behaviour (409 without it, 200 with it)
+- Schema document validation (400 for invalid input)
+- Circular dependency error (400)
+- Unresolvable FK reference error (400)
+- Missing / malformed 'schema' key (400)
+- Authentication enforced (401 for unauthenticated requests)
+"""
+
+
+import unittest
+
+from django.urls import reverse
+from django.test import TransactionTestCase
+from rest_framework import status
+from rest_framework.test import APIClient
+
+from netbox_custom_objects.api.views import _HAS_JSONSCHEMA
+
+from core.models import ObjectType
+from users.models import ObjectPermission
+from utilities.testing import create_test_user
+
+from netbox_custom_objects.schema.exporter import export_cot
+from netbox_custom_objects.models import CustomObjectType
+
+from ..base import CustomObjectsTestCase, TransactionCleanupMixin, create_api_token
+
+
+# ---------------------------------------------------------------------------
+# Base for schema API tests
+# ---------------------------------------------------------------------------
+
+class _SchemaAPIBase(TransactionCleanupMixin, CustomObjectsTestCase, TransactionTestCase):
+    """Base class providing an authenticated API client and helper shortcuts."""
+
+    def setUp(self):
+        super().setUp()
+        self.user = create_test_user('schema_api_user')
+        self.token = create_api_token(self.user)
+        try:
+            token_key = self.token.token  # NetBox ≥ 4.5
+        except AttributeError:
+            token_key = self.token.key
+        self.client = APIClient()
+        self.client.credentials(HTTP_AUTHORIZATION=f"Token {token_key}")
+
+    @property
+    def preview_url(self):
+        return reverse("plugins-api:netbox_custom_objects-api:schema-preview")
+
+    @property
+    def apply_url(self):
+        return reverse("plugins-api:netbox_custom_objects-api:schema-apply")
+
+    def _apply_body(self, schema_doc, allow_destructive=False):
+        return {"schema": schema_doc, "allow_destructive": allow_destructive}
+
+    @staticmethod
+    def _next_field_id(cot):
+        # next_schema_id stores the *last assigned* ID; +1 is the next available one,
+        # mirroring the auto-assign logic in CustomObjectTypeField.save().
+        return cot.next_schema_id + 1
+
+
+# ---------------------------------------------------------------------------
+# Preview endpoint
+# ---------------------------------------------------------------------------
+
+class SchemaPreviewTestCase(_SchemaAPIBase):
+    """POST /schema/preview/ returns a diff without touching the DB."""
+
+    def setUp(self):
+        super().setUp()
+        self.cot = self.create_custom_object_type(name='previewcot', slug='preview-cot')
+        self.field = self.create_custom_object_type_field(
+            self.cot, name='alpha', type='text',
+        )
+
+    def test_preview_returns_200(self):
+        type_def = export_cot(self.cot)
+        resp = self.client.post(
+            self.preview_url,
+            data={"schema_version": "1", "types": [type_def]},
+            format="json",
+        )
+        self.assertEqual(resp.status_code, status.HTTP_200_OK)
+
+    def test_preview_response_contains_diffs_key(self):
+        type_def = export_cot(self.cot)
+        resp = self.client.post(
+            self.preview_url,
+            data={"schema_version": "1", "types": [type_def]},
+            format="json",
+        )
+        self.assertIn("diffs", resp.data)
+
+    def test_preview_noop_has_no_changes(self):
+        type_def = export_cot(self.cot)
+        resp = self.client.post(
+            self.preview_url,
+            data={"schema_version": "1", "types": [type_def]},
+            format="json",
+        )
+        self.assertEqual(len(resp.data["diffs"]), 1)
+        self.assertFalse(resp.data["diffs"][0]["has_changes"])
+
+    def test_preview_detects_field_add(self):
+        self.cot.refresh_from_db()
+        type_def = export_cot(self.cot)
+        next_id = self._next_field_id(self.cot)
+        type_def["fields"].append({"id": next_id, "name": "beta", "type": "text"})
+        resp = self.client.post(
+            self.preview_url,
+            data={"schema_version": "1", "types": [type_def]},
+            format="json",
+        )
+        self.assertEqual(resp.status_code, status.HTTP_200_OK)
+        diff = resp.data["diffs"][0]
+        self.assertTrue(diff["has_changes"])
+        ops = [fc["op"] for fc in diff["field_changes"]]
+        self.assertIn("add", ops)
+
+    def test_preview_detects_field_alter(self):
+        type_def = export_cot(self.cot)
+        for f in type_def["fields"]:
+            if f["name"] == "alpha":
+                f["description"] = "Changed"
+        resp = self.client.post(
+            self.preview_url,
+            data={"schema_version": "1", "types": [type_def]},
+            format="json",
+        )
+        diff = resp.data["diffs"][0]
+        self.assertTrue(diff["has_changes"])
+        alter_ops = [fc for fc in diff["field_changes"] if fc["op"] == "alter"]
+        self.assertEqual(len(alter_ops), 1)
+        self.assertIn("description", alter_ops[0]["changed_attrs"])
+
+    def test_preview_does_not_modify_db(self):
+        self.cot.refresh_from_db()
+        type_def = export_cot(self.cot)
+        next_id = self._next_field_id(self.cot)
+        type_def["fields"].append({"id": next_id, "name": "ghost", "type": "text"})
+        self.client.post(
+            self.preview_url,
+            data={"schema_version": "1", "types": [type_def]},
+            format="json",
+        )
+        # Field must NOT have been created.
+        self.assertFalse(self.cot.fields.filter(name="ghost").exists())
+
+    def test_preview_new_cot_reports_is_new(self):
+        schema_doc = {
+            "schema_version": "1",
+            "types": [{"name": "brandnew", "slug": "brand-new"}],
+        }
+        resp = self.client.post(self.preview_url, data=schema_doc, format="json")
+        self.assertTrue(resp.data["diffs"][0]["is_new"])
+
+    def test_preview_unauthenticated_returns_403_or_401(self):
+        anon = APIClient()
+        type_def = export_cot(self.cot)
+        resp = anon.post(
+            self.preview_url,
+            data={"schema_version": "1", "types": [type_def]},
+            format="json",
+        )
+        self.assertIn(resp.status_code, (status.HTTP_401_UNAUTHORIZED, status.HTTP_403_FORBIDDEN))
+
+    @unittest.skipUnless(_HAS_JSONSCHEMA, "jsonschema not installed")
+    def test_preview_invalid_schema_doc_returns_400(self):
+        # schema_version must be "1" (const in cot_schema_v1.json)
+        resp = self.client.post(
+            self.preview_url,
+            data={"schema_version": "99", "types": [{"name": "x", "slug": "x"}]},
+            format="json",
+        )
+        self.assertEqual(resp.status_code, status.HTTP_400_BAD_REQUEST)
+
+    def test_preview_destructive_change_does_not_raise_error(self):
+        """Preview reports has_destructive_changes=True but does NOT raise 409."""
+        schema_id = self.field.schema_id
+        type_def = export_cot(self.cot)
+        type_def["fields"] = []
+        type_def.setdefault("removed_fields", []).append(
+            {"id": schema_id, "name": "alpha", "type": "text"}
+        )
+        resp = self.client.post(
+            self.preview_url,
+            data={"schema_version": "1", "types": [type_def]},
+            format="json",
+        )
+        self.assertEqual(resp.status_code, status.HTTP_200_OK)
+        self.assertTrue(resp.data["diffs"][0]["has_destructive_changes"])
+
+
+# ---------------------------------------------------------------------------
+# Apply endpoint
+# ---------------------------------------------------------------------------
+
+class SchemaApplyTestCase(_SchemaAPIBase):
+    """POST /schema/apply/ applies the schema document atomically."""
+
+    def setUp(self):
+        super().setUp()
+        perm = ObjectPermission(name='schema_apply_cot_perm', actions=['add', 'change'])
+        perm.save()
+        perm.users.add(self.user)
+        perm.object_types.add(ObjectType.objects.get_for_model(CustomObjectType))
+
+    def test_apply_new_cot_returns_200(self):
+        schema_doc = {
+            "schema_version": "1",
+            "types": [{"name": "applynew", "slug": "apply-new"}],
+        }
+        resp = self.client.post(
+            self.apply_url,
+            data=self._apply_body(schema_doc),
+            format="json",
+        )
+        self.assertEqual(resp.status_code, status.HTTP_200_OK)
+        self.assertTrue(resp.data["applied"])
+
+    def test_apply_new_cot_creates_cot_in_db(self):
+        schema_doc = {
+            "schema_version": "1",
+            "types": [{"name": "applynew2", "slug": "apply-new-2"}],
+        }
+        self.client.post(self.apply_url, data=self._apply_body(schema_doc), format="json")
+        self.assertTrue(CustomObjectType.objects.filter(slug="apply-new-2").exists())
+
+    def test_apply_response_contains_diffs(self):
+        cot = self.create_custom_object_type(name='applydiff', slug='apply-diff')
+        type_def = export_cot(cot)
+        resp = self.client.post(
+            self.apply_url,
+            data=self._apply_body({"schema_version": "1", "types": [type_def]}),
+            format="json",
+        )
+        self.assertIn("diffs", resp.data)
+        self.assertEqual(len(resp.data["diffs"]), 1)
+        self.assertEqual(resp.data["diffs"][0]["slug"], "apply-diff")
+
+    def test_apply_adds_field(self):
+        cot = self.create_custom_object_type(name='applyfield', slug='apply-field')
+        self.create_custom_object_type_field(cot, name='exists', type='text')
+        cot.refresh_from_db()
+        type_def = export_cot(cot)
+        next_id = self._next_field_id(cot)
+        type_def["fields"].append({"id": next_id, "name": "added", "type": "text"})
+        schema_doc = {"schema_version": "1", "types": [type_def]}
+        resp = self.client.post(self.apply_url, data=self._apply_body(schema_doc), format="json")
+        self.assertEqual(resp.status_code, status.HTTP_200_OK)
+        self.assertTrue(cot.fields.filter(name="added").exists())
+
+    def test_apply_without_allow_destructive_returns_409(self):
+        cot = self.create_custom_object_type(name='applydest', slug='apply-dest')
+        field = self.create_custom_object_type_field(cot, name='bye', type='text')
+        sid = field.schema_id
+        type_def = export_cot(cot)
+        type_def["fields"] = []
+        type_def["removed_fields"] = [{"id": sid, "name": "bye", "type": "text"}]
+        schema_doc = {"schema_version": "1", "types": [type_def]}
+        resp = self.client.post(self.apply_url, data=self._apply_body(schema_doc), format="json")
+        self.assertEqual(resp.status_code, status.HTTP_409_CONFLICT)
+        self.assertEqual(resp.data["error"], "destructive_changes")
+
+    def test_apply_409_includes_destructive_slugs(self):
+        cot = self.create_custom_object_type(name='applyslug', slug='apply-slug')
+        field = self.create_custom_object_type_field(cot, name='gone', type='text')
+        sid = field.schema_id
+        type_def = export_cot(cot)
+        type_def["fields"] = []
+        type_def["removed_fields"] = [{"id": sid, "name": "gone", "type": "text"}]
+        schema_doc = {"schema_version": "1", "types": [type_def]}
+        resp = self.client.post(self.apply_url, data=self._apply_body(schema_doc), format="json")
+        self.assertIn("apply-slug", resp.data["destructive_slugs"])
+
+    def test_apply_409_does_not_remove_field(self):
+        cot = self.create_custom_object_type(name='applyguard', slug='apply-guard')
+        field = self.create_custom_object_type_field(cot, name='keep', type='text')
+        sid = field.schema_id
+        type_def = export_cot(cot)
+        type_def["fields"] = []
+        type_def["removed_fields"] = [{"id": sid, "name": "keep", "type": "text"}]
+        schema_doc = {"schema_version": "1", "types": [type_def]}
+        self.client.post(self.apply_url, data=self._apply_body(schema_doc), format="json")
+        self.assertTrue(cot.fields.filter(schema_id=sid).exists())
+
+    def test_apply_with_allow_destructive_removes_field(self):
+        cot = self.create_custom_object_type(name='applyrm', slug='apply-rm')
+        field = self.create_custom_object_type_field(cot, name='victim', type='text')
+        sid = field.schema_id
+        type_def = export_cot(cot)
+        type_def["fields"] = []
+        type_def["removed_fields"] = [{"id": sid, "name": "victim", "type": "text"}]
+        schema_doc = {"schema_version": "1", "types": [type_def]}
+        resp = self.client.post(
+            self.apply_url,
+            data=self._apply_body(schema_doc, allow_destructive=True),
+            format="json",
+        )
+        self.assertEqual(resp.status_code, status.HTTP_200_OK)
+        self.assertFalse(cot.fields.filter(schema_id=sid).exists())
+
+    def test_apply_missing_schema_key_returns_400(self):
+        resp = self.client.post(
+            self.apply_url,
+            data={"allow_destructive": False},
+            format="json",
+        )
+        self.assertEqual(resp.status_code, status.HTTP_400_BAD_REQUEST)
+
+    def test_apply_schema_not_a_dict_returns_400(self):
+        resp = self.client.post(
+            self.apply_url,
+            data={"schema": "not a dict", "allow_destructive": False},
+            format="json",
+        )
+        self.assertEqual(resp.status_code, status.HTTP_400_BAD_REQUEST)
+
+    def test_apply_unauthenticated_returns_403_or_401(self):
+        anon = APIClient()
+        resp = anon.post(
+            self.apply_url,
+            data=self._apply_body({"schema_version": "1", "types": []}),
+            format="json",
+        )
+        self.assertIn(resp.status_code, (status.HTTP_401_UNAUTHORIZED, status.HTTP_403_FORBIDDEN))
+
+    def test_apply_unresolvable_object_type_returns_400(self):
+        cot = self.create_custom_object_type(name='applyrotfail', slug='apply-rot-fail')
+        self.create_custom_object_type_field(cot, name='ok', type='text')
+        cot.refresh_from_db()
+        type_def = export_cot(cot)
+        next_id = self._next_field_id(cot)
+        type_def["fields"].append({
+            "id": next_id, "name": "bad_obj", "type": "object",
+            "related_object_type": "does/notexist",
+        })
+        schema_doc = {"schema_version": "1", "types": [type_def]}
+        resp = self.client.post(self.apply_url, data=self._apply_body(schema_doc), format="json")
+        self.assertEqual(resp.status_code, status.HTTP_400_BAD_REQUEST)
+        self.assertEqual(resp.data["error"], "unresolvable_reference")
+
+    def test_apply_unresolvable_choice_set_returns_400(self):
+        cot = self.create_custom_object_type(name='applycsfail', slug='apply-cs-fail')
+        self.create_custom_object_type_field(cot, name='ok', type='text')
+        cot.refresh_from_db()
+        type_def = export_cot(cot)
+        next_id = self._next_field_id(cot)
+        type_def["fields"].append({
+            "id": next_id, "name": "bad_sel", "type": "select",
+            "choice_set": "NoSuchSet",
+        })
+        schema_doc = {"schema_version": "1", "types": [type_def]}
+        resp = self.client.post(self.apply_url, data=self._apply_body(schema_doc), format="json")
+        self.assertEqual(resp.status_code, status.HTTP_400_BAD_REQUEST)
+        self.assertEqual(resp.data["error"], "unresolvable_reference")
+
+    def test_apply_schema_document_persisted_after_apply(self):
+        cot = self.create_custom_object_type(name='applydoccheck', slug='apply-doc-check')
+        type_def = export_cot(cot)
+        schema_doc = {"schema_version": "1", "types": [type_def]}
+        self.client.post(self.apply_url, data=self._apply_body(schema_doc), format="json")
+        cot.refresh_from_db()
+        self.assertIsNotNone(cot.schema_document)
+
+    def test_apply_noop_returns_200(self):
+        cot = self.create_custom_object_type(name='applynoop', slug='apply-noop')
+        self.create_custom_object_type_field(cot, name='stable', type='text')
+        type_def = export_cot(cot)
+        schema_doc = {"schema_version": "1", "types": [type_def]}
+        # First apply initialises schema_document; second apply is a true no-op.
+        self.client.post(self.apply_url, data=self._apply_body(schema_doc), format="json")
+        resp = self.client.post(self.apply_url, data=self._apply_body(schema_doc), format="json")
+        self.assertEqual(resp.status_code, status.HTTP_200_OK)
+        self.assertTrue(resp.data["applied"])
+        self.assertFalse(resp.data["diffs"][0]["has_changes"])
+
+    def test_apply_allow_destructive_string_returns_400(self):
+        schema_doc = {"schema_version": "1", "types": []}
+        resp = self.client.post(
+            self.apply_url,
+            data={"schema": schema_doc, "allow_destructive": "true"},
+            format="json",
+        )
+        self.assertEqual(resp.status_code, status.HTTP_400_BAD_REQUEST)
+        self.assertIn("allow_destructive", resp.data)
diff --git a/netbox_custom_objects/tests/test_api.py b/netbox_custom_objects/tests/test_api.py
index bfd824c1..90a5138e 100644
--- a/netbox_custom_objects/tests/test_api.py
+++ b/netbox_custom_objects/tests/test_api.py
@@ -6,6 +6,7 @@
 
 from utilities.testing import APIViewTestCases, create_test_user
 from rest_framework import status
+from rest_framework.test import APIClient
 
 from netbox_custom_objects.models import CustomObjectType, CustomObjectTypeField
 from .base import CustomObjectsTestCase
@@ -791,7 +792,8 @@ class ContextFieldApiTestCase(CustomObjectsTestCase, TestCase):
     def setUp(self):
         self.user = create_test_user('ctxapiuser')
         token_key = create_token(self.user)
-        self.header = {'HTTP_AUTHORIZATION': f'Token {token_key}'}
+        self.client = APIClient()
+        self.client.credentials(HTTP_AUTHORIZATION=f'Token {token_key}')
 
         # --- COT A: primary field + context field ---
         self.cot_with_primary = CustomObjectsTestCase.create_custom_object_type(
@@ -889,7 +891,7 @@ def test_display_equals_primary_field_value(self):
         """display must be the primary field value, not the fallback."""
         instance = self.model_with_primary.objects.create(name='Route-A', owner='Alice')
         response = self.client.get(
-            self._detail_url(self.cot_with_primary, instance), **self.header
+            self._detail_url(self.cot_with_primary, instance)
         )
         self.assertEqual(response.status_code, status.HTTP_200_OK)
         self.assertEqual(response.data['display'], 'Route-A')
@@ -898,7 +900,7 @@ def test_context_display_value_with_primary_field(self):
         """_context.display must equal the context field value when primary is set."""
         instance = self.model_with_primary.objects.create(name='Route-A', owner='Alice')
         response = self.client.get(
-            self._detail_url(self.cot_with_primary, instance), **self.header
+            self._detail_url(self.cot_with_primary, instance)
         )
         self.assertEqual(response.status_code, status.HTTP_200_OK)
         self.assertIsNotNone(response.data['_context'])
@@ -908,7 +910,7 @@ def test_context_null_when_context_field_has_no_value(self):
         """_context must be null when the context field carries no value."""
         instance = self.model_with_primary.objects.create(name='Route-B')
         response = self.client.get(
-            self._detail_url(self.cot_with_primary, instance), **self.header
+            self._detail_url(self.cot_with_primary, instance)
         )
         self.assertEqual(response.status_code, status.HTTP_200_OK)
         self.assertIsNone(response.data['_context'])
@@ -920,7 +922,7 @@ def test_display_uses_fallback_when_no_primary_field(self):
         instance = self.model_no_primary.objects.create(owner='Bob')
         expected = f"{self.cot_no_primary.display_name} {instance.id}"
         response = self.client.get(
-            self._detail_url(self.cot_no_primary, instance), **self.header
+            self._detail_url(self.cot_no_primary, instance)
         )
         self.assertEqual(response.status_code, status.HTTP_200_OK)
         self.assertEqual(response.data['display'], expected)
@@ -929,7 +931,7 @@ def test_context_display_value_with_fallback_display(self):
         """_context.display must work correctly even when display uses the fallback name."""
         instance = self.model_no_primary.objects.create(owner='Bob')
         response = self.client.get(
-            self._detail_url(self.cot_no_primary, instance), **self.header
+            self._detail_url(self.cot_no_primary, instance)
         )
         self.assertEqual(response.status_code, status.HTTP_200_OK)
         self.assertIsNotNone(response.data['_context'])
@@ -943,7 +945,7 @@ def test_multiple_context_fields_joined_in_display(self):
             name='Route-C', owner='Carol', region='EU'
         )
         response = self.client.get(
-            self._detail_url(self.cot_multi_ctx, instance), **self.header
+            self._detail_url(self.cot_multi_ctx, instance)
         )
         self.assertEqual(response.status_code, status.HTTP_200_OK)
         self.assertIsNotNone(response.data['_context'])
@@ -954,13 +956,102 @@ def test_multiple_context_fields_omits_empty_values(self):
         instance = self.model_multi_ctx.objects.create(name='Route-D', owner='Dave')
         # region (second context field) is not set
         response = self.client.get(
-            self._detail_url(self.cot_multi_ctx, instance), **self.header
+            self._detail_url(self.cot_multi_ctx, instance)
         )
         self.assertEqual(response.status_code, status.HTTP_200_OK)
         self.assertIsNotNone(response.data['_context'])
         self.assertEqual(response.data['_context']['display'], 'Dave')
 
 
+# ---------------------------------------------------------------------------
+# PEP 440 version string validation — API layer (issue #392)
+# ---------------------------------------------------------------------------
+
+class Pep440APIValidationTestCase(CustomObjectsTestCase, TestCase):
+    """
+    Verify that ``validate_pep440`` surfaces as a 400 at the API layer for
+    ``CustomObjectType.version`` and ``CustomObjectTypeField.deprecated_since``
+    / ``scheduled_removal``.
+
+    DRF's ModelSerializer copies model-field validators into the serializer
+    field, so these should be enforced during deserialization without any
+    extra serializer code.
+    """
+
+    def setUp(self):
+        super().setUp()
+        token_key = create_token(self.user)
+        self.client = APIClient()
+        self.client.credentials(HTTP_AUTHORIZATION=f'Token {token_key}')
+
+        # Permission to create/change CustomObjectType records.
+        add_cot_perm = ObjectPermission(name='pep440_add_cot', actions=['add', 'change'])
+        add_cot_perm.save()
+        add_cot_perm.users.add(self.user)
+        add_cot_perm.object_types.add(ObjectType.objects.get_for_model(CustomObjectType))
+
+        # Permission to change CustomObjectTypeField records.
+        from netbox_custom_objects.models import CustomObjectTypeField  # noqa: PLC0415
+        change_field_perm = ObjectPermission(name='pep440_change_field', actions=['add', 'change'])
+        change_field_perm.save()
+        change_field_perm.users.add(self.user)
+        change_field_perm.object_types.add(ObjectType.objects.get_for_model(CustomObjectTypeField))
+
+    # ------------------------------------------------------------------
+    # CustomObjectType.version
+    # ------------------------------------------------------------------
+
+    def test_create_cot_invalid_version_returns_400(self):
+        url = reverse('plugins-api:netbox_custom_objects-api:customobjecttype-list')
+        data = {'name': 'vertest', 'slug': 'ver-test', 'version': 'not-a-version'}
+        response = self.client.post(url, data, format='json')
+        self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
+        self.assertIn('version', response.data)
+
+    def test_create_cot_valid_version_accepted(self):
+        url = reverse('plugins-api:netbox_custom_objects-api:customobjecttype-list')
+        data = {'name': 'vertest2', 'slug': 'ver-test-2', 'version': '1.2.3'}
+        response = self.client.post(url, data, format='json')
+        self.assertEqual(response.status_code, status.HTTP_201_CREATED)
+
+    # ------------------------------------------------------------------
+    # CustomObjectTypeField.deprecated_since / scheduled_removal
+    # ------------------------------------------------------------------
+
+    def test_patch_field_invalid_deprecated_since_returns_400(self):
+        cot = self.create_custom_object_type(name='pep440cot', slug='pep440-cot')
+        field = self.create_custom_object_type_field(cot, name='alpha', type='text')
+        url = reverse(
+            'plugins-api:netbox_custom_objects-api:customobjecttypefield-detail',
+            kwargs={'pk': field.pk},
+        )
+        response = self.client.patch(url, {'deprecated_since': 'latest'}, format='json')
+        self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
+        self.assertIn('deprecated_since', response.data)
+
+    def test_patch_field_invalid_scheduled_removal_returns_400(self):
+        cot = self.create_custom_object_type(name='pep440cot2', slug='pep440-cot-2')
+        field = self.create_custom_object_type_field(cot, name='beta', type='text')
+        url = reverse(
+            'plugins-api:netbox_custom_objects-api:customobjecttypefield-detail',
+            kwargs={'pk': field.pk},
+        )
+        response = self.client.patch(url, {'scheduled_removal': '1.x.0'}, format='json')
+        self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
+        self.assertIn('scheduled_removal', response.data)
+
+    def test_patch_cot_valid_version_accepted(self):
+        # PATCH CustomObjectType.version (no DDL on COT update) verifies the
+        # validator doesn't reject a valid PEP 440 string.
+        cot = self.create_custom_object_type(name='pep440cot3', slug='pep440-cot-3')
+        url = reverse(
+            'plugins-api:netbox_custom_objects-api:customobjecttype-detail',
+            kwargs={'pk': cot.pk},
+        )
+        response = self.client.patch(url, {'version': '2.0.0'}, format='json')
+        self.assertEqual(response.status_code, status.HTTP_200_OK)
+
+
 class SchemaIdReadOnlyTest(CustomObjectsTestCase, TestCase):
     """
     schema_id on CustomObjectTypeField is read-only via the API.
diff --git a/netbox_custom_objects/tests/test_comparator.py b/netbox_custom_objects/tests/test_comparator.py
index fcb3de05..1d50bc26 100644
--- a/netbox_custom_objects/tests/test_comparator.py
+++ b/netbox_custom_objects/tests/test_comparator.py
@@ -26,7 +26,7 @@
     diff_cot,
     diff_document,
 )
-from netbox_custom_objects.exporter import export_cot, export_cots
+from netbox_custom_objects.schema.exporter import export_cot, export_cots
 from netbox_custom_objects.models import CustomObjectTypeField
 
 from .base import CustomObjectsTestCase
diff --git a/netbox_custom_objects/tests/test_field_types.py b/netbox_custom_objects/tests/test_field_types.py
index 4dff11da..6f4ddb52 100644
--- a/netbox_custom_objects/tests/test_field_types.py
+++ b/netbox_custom_objects/tests/test_field_types.py
@@ -612,6 +612,23 @@ def test_object_field_creation(self):
         self.assertEqual(field.type, "object")
         self.assertEqual(field.related_object_type, self.device_object_type)
 
+    def test_get_model_field_raises_not_implemented_error_for_null_related_object_type(self):
+        """get_model_field() must raise NotImplementedError (not ContentType.DoesNotExist)
+        when related_object_type_id is NULL.  All callers catch NotImplementedError to
+        skip broken fields; an unexpected ContentType.DoesNotExist would propagate up
+        and crash model generation or the serializer."""
+        field = self.create_custom_object_type_field(
+            self.custom_object_type,
+            name="broken_obj",
+            label="Broken",
+            type="object",
+            related_object_type=self.device_object_type,
+        )
+        CustomObjectTypeField.objects.filter(pk=field.pk).update(related_object_type=None)
+        field.refresh_from_db()
+        with self.assertRaises(NotImplementedError):
+            ObjectFieldType().get_model_field(field)
+
     def test_object_field_model_generation(self):
         """Test object field model generation."""
         self.create_custom_object_type_field(
@@ -666,6 +683,21 @@ def test_multiobject_field_creation(self):
         self.assertEqual(field.type, "multiobject")
         self.assertEqual(field.related_object_type, self.device_object_type)
 
+    def test_get_model_field_raises_not_implemented_error_for_null_related_object_type(self):
+        """get_model_field() must raise NotImplementedError (not ContentType.DoesNotExist)
+        when related_object_type_id is NULL so callers handle it consistently."""
+        field = self.create_custom_object_type_field(
+            self.custom_object_type,
+            name="broken_multi",
+            label="Broken Multi",
+            type="multiobject",
+            related_object_type=self.device_object_type,
+        )
+        CustomObjectTypeField.objects.filter(pk=field.pk).update(related_object_type=None)
+        field.refresh_from_db()
+        with self.assertRaises(NotImplementedError):
+            MultiObjectFieldType().get_model_field(field)
+
     def test_multiobject_field_model_generation(self):
         """Test multiobject field model generation."""
         self.create_custom_object_type_field(
diff --git a/netbox_custom_objects/tests/test_models.py b/netbox_custom_objects/tests/test_models.py
index 6e55d5bc..cc7831a8 100644
--- a/netbox_custom_objects/tests/test_models.py
+++ b/netbox_custom_objects/tests/test_models.py
@@ -203,6 +203,36 @@ def test_custom_object_type_save_creates_table(self):
             expected_table = f"custom_objects_{custom_object_type.id}"
             self.assertIn(expected_table, tables)
 
+    def test_register_search_index_skips_object_field_absent_from_stub_model(self):
+        """register_custom_object_search_index() must use local_fields/local_many_to_many
+        rather than _meta.get_field() to check field presence.  _meta.get_field() for a
+        name not in _forward_fields_map triggers Django's lazy _relation_tree computation,
+        which calls apps.get_models() → our override → get_model() for every COT →
+        infinite recursion when called during model registration.
+
+        Regression for PR #474: the stub model generated with skip_object_fields=True
+        does not have the OBJECT field, but self.fields.filter(search_weight__gt=0)
+        still returns it from the database.
+        """
+        cot = self.create_custom_object_type(name="StubSearchTest", slug="stub-search-test")
+        self.create_custom_object_type_field(
+            cot, name="name", label="Name", type="text", primary=True, search_weight=1000,
+        )
+        self.create_custom_object_type_field(
+            cot, name="ref_site", label="Site", type="object",
+            related_object_type=self.get_site_object_type(),
+            search_weight=500,
+        )
+        stub_model = cot.get_model(skip_object_fields=True)
+        model_field_names = (
+            {f.name for f in stub_model._meta.local_fields}
+            | {f.name for f in stub_model._meta.local_many_to_many}
+        )
+        self.assertNotIn("ref_site", model_field_names,
+                         "OBJECT field must be absent from stub model")
+        # Must not raise FieldDoesNotExist, RecursionError, or any other exception.
+        cot.register_custom_object_search_index(stub_model)
+
     @skip("Fails in suite but not individually")
     def test_custom_object_type_delete_removes_table(self):
         """Test that deleting a custom object type removes the database table."""
@@ -1490,3 +1520,142 @@ def test_stub_model_search_index_excludes_absent_fields(self):
             indexed_field_names,
             "Object-type fields absent from the stub must be excluded from the search index",
         )
+
+
+# ---------------------------------------------------------------------------
+# Semver / version string validation (issue #392)
+# ---------------------------------------------------------------------------
+
+class SemverValidationTestCase(CustomObjectsTestCase, TestCase):
+    """Validate that version-string fields reject non-PEP-440 values."""
+
+    # ------------------------------------------------------------------
+    # CustomObjectType.version
+    # ------------------------------------------------------------------
+
+    def test_cot_version_blank_is_valid(self):
+        cot = self.create_custom_object_type(name='semver_cot', slug='semver-cot')
+        cot.version = ''
+        cot.full_clean()  # must not raise
+
+    def test_cot_version_valid_semver(self):
+        cot = self.create_custom_object_type(name='semver_cot2', slug='semver-cot-2')
+        for v in ('1.0.0', '2.3.4', '0.0.1', '1.0.0.post1', '1.0.0a1'):
+            cot.version = v
+            cot.full_clean()  # must not raise
+
+    def test_cot_version_invalid_raises_validation_error(self):
+        cot = self.create_custom_object_type(name='semver_cot3', slug='semver-cot-3')
+        for bad in ('not-a-version', '1.x.0', 'latest', '!!invalid!!'):
+            cot.version = bad
+            with self.assertRaises(ValidationError, msg=f"Expected ValidationError for version={bad!r}"):
+                cot.full_clean()
+
+    # ------------------------------------------------------------------
+    # CustomObjectTypeField.deprecated_since
+    # ------------------------------------------------------------------
+
+    def test_field_deprecated_since_blank_is_valid(self):
+        cot = self.create_custom_object_type(name='semver_f1', slug='semver-f1')
+        field = self.create_custom_object_type_field(cot, name='alpha', type='text')
+        field.deprecated_since = ''
+        field.full_clean()
+
+    def test_field_deprecated_since_valid_semver(self):
+        cot = self.create_custom_object_type(name='semver_f2', slug='semver-f2')
+        field = self.create_custom_object_type_field(cot, name='beta', type='text')
+        field.deprecated_since = '2.0.0'
+        field.full_clean()
+
+    def test_field_deprecated_since_invalid_raises(self):
+        cot = self.create_custom_object_type(name='semver_f3', slug='semver-f3')
+        field = self.create_custom_object_type_field(cot, name='gamma', type='text')
+        for bad in ('not-a-version', '1.x.0', 'latest', '!!invalid!!'):
+            field.deprecated_since = bad
+            with self.assertRaises(ValidationError, msg=f"Expected ValidationError for deprecated_since={bad!r}"):
+                field.full_clean()
+
+    # ------------------------------------------------------------------
+    # CustomObjectTypeField.scheduled_removal
+    # ------------------------------------------------------------------
+
+    def test_field_scheduled_removal_blank_is_valid(self):
+        cot = self.create_custom_object_type(name='semver_f4', slug='semver-f4')
+        field = self.create_custom_object_type_field(cot, name='delta', type='text')
+        field.scheduled_removal = ''
+        field.full_clean()
+
+    def test_field_scheduled_removal_valid_semver(self):
+        cot = self.create_custom_object_type(name='semver_f5', slug='semver-f5')
+        field = self.create_custom_object_type_field(cot, name='epsilon', type='text')
+        field.scheduled_removal = '3.0.0'
+        field.full_clean()
+
+    def test_field_scheduled_removal_invalid_raises(self):
+        cot = self.create_custom_object_type(name='semver_f6', slug='semver-f6')
+        field = self.create_custom_object_type_field(cot, name='zeta', type='text')
+        for bad in ('v-bad', '1.x.0', 'latest', '!!invalid!!'):
+            field.scheduled_removal = bad
+            with self.assertRaises(ValidationError, msg=f"Expected ValidationError for scheduled_removal={bad!r}"):
+                field.full_clean()
+
+
+class NullRelatedObjectTypeTestCase(CustomObjectsTestCase, TestCase):
+    """Regression tests for graceful handling of OBJECT/MULTIOBJECT fields whose
+    related_object_type_id is NULL or points to a deleted ContentType.
+
+    A NULL FK can occur when a COT field is created via direct DB manipulation or
+    when the referenced ContentType is deleted.  All code paths that build the
+    dynamic model or serializer must skip such fields rather than crashing.
+
+    Covers the fixes in _fetch_and_generate_field_attrs (ContentType.DoesNotExist →
+    NotImplementedError) and get_serializer_class (Meta.fields/attrs mismatch guard).
+    """
+
+    def _make_cot_with_null_object_field(self, name, slug, field_name="broken_ref"):
+        from netbox_custom_objects.models import CustomObjectType
+        cot = self.create_custom_object_type(name=name, slug=slug)
+        self.create_custom_object_type_field(
+            cot, name="title", label="Title", type="text", primary=True, required=True,
+        )
+        field = self.create_custom_object_type_field(
+            cot, name=field_name, label="Broken Ref", type="object",
+            related_object_type=self.get_site_object_type(),
+        )
+        # Force the FK to NULL to simulate stale/corrupt data (e.g. ContentType deleted)
+        CustomObjectTypeField.objects.filter(pk=field.pk).update(related_object_type=None)
+        CustomObjectType.clear_model_cache()
+        return cot
+
+    def test_get_model_skips_object_field_with_null_related_object_type(self):
+        """get_model() must succeed and silently skip an OBJECT field whose
+        related_object_type_id is NULL rather than raising ContentType.DoesNotExist."""
+        cot = self._make_cot_with_null_object_field("NullRelObj", "null-rel-obj")
+        model = cot.get_model()
+        self.assertIsNotNone(model)
+        model_field_names = (
+            {f.name for f in model._meta.local_fields}
+            | {f.name for f in model._meta.local_many_to_many}
+        )
+        self.assertNotIn("broken_ref", model_field_names,
+                         "Field with null FK must be absent from the generated model")
+        self.assertIn("title", model_field_names,
+                      "Normal fields must still be present")
+
+    def test_get_serializer_class_handles_null_related_object_type(self):
+        """get_serializer_class() must not raise AttributeError when an OBJECT field
+        was skipped during model generation due to a NULL related_object_type_id.
+        Regression for the Meta.fields/attrs mismatch that caused DRF to raise a
+        validation error at serializer initialization time."""
+        from netbox_custom_objects.api.serializers import get_serializer_class
+
+        cot = self._make_cot_with_null_object_field(
+            "NullRelSerializer", "null-rel-serializer"
+        )
+        model = cot.get_model()
+        serializer_cls = get_serializer_class(model)
+        self.assertIsNotNone(serializer_cls)
+        self.assertNotIn("broken_ref", serializer_cls.Meta.fields,
+                         "Null-FK field must not appear in serializer Meta.fields")
+        self.assertIn("title", serializer_cls.Meta.fields,
+                      "Normal fields must still be present in serializer")
diff --git a/netbox_custom_objects/views.py b/netbox_custom_objects/views.py
index 352238cd..34d764ac 100644
--- a/netbox_custom_objects/views.py
+++ b/netbox_custom_objects/views.py
@@ -527,7 +527,6 @@ def custom_init(self, *args, **kwargs):
                             content_type = field_obj.related_object_type
                             if content_type.app_label == APP_LABEL:
                                 # Custom object type
-                                from netbox_custom_objects.models import CustomObjectType
                                 custom_object_type_id = extract_cot_id_from_model_name(content_type.model)
                                 if custom_object_type_id is None:
                                     raise ValueError(
diff --git a/pyproject.toml b/pyproject.toml
index 8355ceb8..958fc248 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -25,6 +25,7 @@ classifiers = [
 
 dependencies = [
     "Django",
+    "packaging",
 ]
 
 [project.optional-dependencies]