diff --git a/.github/workflows/python-publish.yml b/.github/workflows/python-publish.yml
index bdaab28..af4b79d 100644
--- a/.github/workflows/python-publish.yml
+++ b/.github/workflows/python-publish.yml
@@ -1,11 +1,4 @@
-# This workflow will upload a Python Package using Twine when a release is created
-# For more information see: https://docs.github.com/en/actions/automating-builds-and-tests/building-and-testing-python#publishing-to-package-registries
-
-# This workflow uses actions that are not certified by GitHub.
-# They are provided by a third-party and are governed by
-# separate terms of service, privacy policy, and support
-# documentation.
-
+# Build and upload to PyPI when a GitHub release is published.
 name: Upload Python Package
 
 on:
@@ -17,23 +10,19 @@ permissions:
 
 jobs:
   deploy:
-
     runs-on: ubuntu-latest
-
     steps:
-    - uses: actions/checkout@v3
-    - name: Set up Python
-      uses: actions/setup-python@v3
-      with:
-        python-version: '3.x'
-    - name: Install dependencies
-      run: |
-        python -m pip install --upgrade pip
-        pip install build
-    - name: Build package
-      run: python -m build
-    - name: Publish package
-      uses: pypa/gh-action-pypi-publish@27b31702a0e7fc50959f5ad993c78deac1bdfc29
-      with:
-        user: __token__
-        password: ${{ secrets.PYPI_API_TOKEN }}
+      - uses: actions/checkout@v4
+      - uses: actions/setup-python@v5
+        with:
+          python-version: "3.12"
+      - name: Build package
+        run: |
+          python -m pip install --upgrade pip build twine
+          python -m build
+          twine check dist/*
+      - name: Publish package
+        uses: pypa/gh-action-pypi-publish@release/v1
+        with:
+          user: __token__
+          password: ${{ secrets.PYPI_API_TOKEN }}
diff --git a/CHANGELOG.md b/CHANGELOG.md
index 3f9bfb5..d462770 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -32,5 +32,10 @@ Aggify is now a standalone, async-first MongoDB ODM with a Django-style API. **M
 - `$subtract` / `$divide` with 3 operands, mutation of shared `F` / `Q` objects, `$unwind` using the wrong path,
   silently dropped `$match` conditions, magic `let` substitution, `[n]` meaning `limit n`.
 
+### Fixed before release
+- Pipeline/queryset aggregate options (`default_max_time_ms`, `options(max_time_ms=..., allow_disk_use=...)`) were sent with the wrong
+  option names and rejected by the server; they are now translated to `maxTimeMS` / `allowDiskUse` / `batchSize`.
+- A queryset annotation missing from a result row is `None` instead of an unset attribute; dangling references keep their id on `fetch_related`.
+
 ### Breaking
 See the [upgrade guide](docs/migration/aggify-0.md).
diff --git a/aggify/contrib/django/forms.py b/aggify/contrib/django/forms.py
index 3e4bb0b..630acab 100644
--- a/aggify/contrib/django/forms.py
+++ b/aggify/contrib/django/forms.py
@@ -2,6 +2,7 @@
 
 from __future__ import annotations
 
+import inspect
 from typing import Any, Dict, Iterable, Optional, Type
 
 from django import forms
@@ -9,6 +10,8 @@
 from aggify import fields as f
 from aggify.exceptions import ValidationError as AggifyValidationError
 
+_URL_HAS_ASSUME_SCHEME = "assume_scheme" in inspect.signature(forms.URLField.__init__).parameters
+
 
 def _form_field(field: f.Field) -> Optional[forms.Field]:
     kw: Dict[str, Any] = {"required": field.required, "help_text": field.help_text}
@@ -20,6 +23,8 @@ def _form_field(field: f.Field) -> Optional[forms.Field]:
     if isinstance(field, f.EmailField):
         return forms.EmailField(max_length=field.max_length, **kw)
     if isinstance(field, f.URLField):
+        if _URL_HAS_ASSUME_SCHEME:  # pragma: no branch -- depends on the installed Django; Django >= 5.0: silence the "default scheme will change" deprecation
+            kw["assume_scheme"] = "https"
         return forms.URLField(max_length=field.max_length, **kw)
     if isinstance(field, f.StringField):
         return forms.CharField(min_length=field.min_length, max_length=field.max_length, strip=field.strip, **kw)
diff --git a/aggify/documents.py b/aggify/documents.py
index 7da030e..e23184e 100644
--- a/aggify/documents.py
+++ b/aggify/documents.py
@@ -158,8 +158,6 @@ def __new__(mcs, name: str, bases: Tuple[type, ...], attrs: Dict[str, Any], **kw
                     inherited.append((fname, fld.clone()))
         own = sorted(declared, key=lambda kv: kv[1]._creation_order)
         for fname, fld in [*inherited, *own]:
-            if fname in opts.fields and fname not in dict(own):
-                continue
             fld.contribute_to_class(cls, fname)
             opts.fields[fname] = fld
 
@@ -692,7 +690,9 @@ def fetch_related(self, *names: str) -> Document:
             field = self._meta.get_field(n)
             current = self._data.get(n)
             if current is not None and not hasattr(current, "_meta"):
-                self._data[n] = field.document_type.objects.filter(pk=current).first()
+                target = field.document_type.objects.filter(pk=current).first()
+                if target is not None:  # a dangling reference keeps its id
+                    self._data[n] = target
         return self
 
     async def afetch_related(self, *names: str) -> Document:
@@ -700,7 +700,9 @@ async def afetch_related(self, *names: str) -> Document:
             field = self._meta.get_field(n)
             current = self._data.get(n)
             if current is not None and not hasattr(current, "_meta"):
-                self._data[n] = await field.document_type.objects.filter(pk=current).afirst()
+                target = await field.document_type.objects.filter(pk=current).afirst()
+                if target is not None:
+                    self._data[n] = target
         return self
 
     # ------------------------------------------------------------------ #
diff --git a/aggify/fields.py b/aggify/fields.py
index 60ea803..5e5c513 100644
--- a/aggify/fields.py
+++ b/aggify/fields.py
@@ -43,7 +43,7 @@
 from aggify import registry
 from aggify.conf import settings
 from aggify.exceptions import ValidationError
-from aggify.lookups import LOOKUPS, get_lookup, prepare_kind
+from aggify.lookups import get_lookup, prepare_kind
 from aggify.security import Raw, assert_safe_value, validate_dict_key
 
 T = TypeVar("T")
@@ -119,9 +119,10 @@ def contribute_to_class(self, model: Any, name: str) -> None:
         setattr(model, name, self)
 
     @overload
-    def __get__(self, instance: None, owner: Any) -> Field[T]: ...
+    def __get__(self, instance: None, owner: Any) -> Field[T]: ...  # pragma: no cover
+
     @overload
-    def __get__(self, instance: Any, owner: Any) -> T: ...
+    def __get__(self, instance: Any, owner: Any) -> T: ...  # pragma: no cover
 
     def __get__(self, instance: Any, owner: Any = None) -> Any:
         if instance is None:
@@ -410,8 +411,6 @@ def to_python(self, value: Any) -> Any:
                 return int(value.strip())
             except ValueError:
                 return value
-        if isinstance(value, bson.Int64):
-            return int(value)
         return value
 
     def to_mongo(self, value: Any) -> Any:
@@ -788,9 +787,7 @@ def validate(self, value: Any) -> None:
     def lookup_field(self, lookup: str, value: Any) -> Field:
         if lookup in ("exact", "ne") and isinstance(value, (list, tuple)):
             return self  # whole array comparison
-        if lookup in ("in", "nin", "all") or lookup in LOOKUPS:
-            return self.field
-        return self
+        return self.field
 
     def _prepare_exact(self, value: Any) -> Any:
         if isinstance(value, (list, tuple)):
diff --git a/aggify/pipeline.py b/aggify/pipeline.py
index 501ec05..512995e 100644
--- a/aggify/pipeline.py
+++ b/aggify/pipeline.py
@@ -645,8 +645,7 @@ def _root_scope(self, expr: Any) -> Tuple[Any, Scope, Context]:
             f = r.field
             if isinstance(f, ListField):
                 f = f.field
-            if isinstance(f, EmbeddedField):
-                scope = Scope.for_model(f.document_type)
+            scope = Scope.for_model(f.document_type)  # _replace() has verified it is embedded
             return "$" + r.db_path, scope, ctx
         return wrap(expr).resolve(ctx), scope, ctx
 
@@ -809,11 +808,15 @@ def options(self, **opts: Any) -> Pipeline:
         new._options.update(opts)
         return new
 
-    def _aggregate_kwargs(self) -> Dict[str, Any]:
+    #: option names -> the spelling of the ``aggregate`` command (pymongo forwards unknown kwargs verbatim)
+    _COMMAND_OPTIONS = {"allow_disk_use": "allowDiskUse", "max_time_ms": "maxTimeMS", "batch_size": "batchSize"}
+
+    def _aggregate_kwargs(self, **extra: Any) -> Dict[str, Any]:
         kw = dict(self._options)
         if "max_time_ms" not in kw and settings.default_max_time_ms:
             kw["max_time_ms"] = settings.default_max_time_ms
-        return kw
+        kw.update(extra)
+        return {self._COMMAND_OPTIONS.get(k, k): v for k, v in kw.items()}
 
     def _alias(self) -> str:
         return self.model._meta.db_alias if self.model is not None else connection.DEFAULT_ALIAS
diff --git a/aggify/queryset.py b/aggify/queryset.py
index c2bb706..9009794 100644
--- a/aggify/queryset.py
+++ b/aggify/queryset.py
@@ -492,15 +492,14 @@ def _hydrate_row(self, row: Dict[str, Any], scope: Optional[Scope], loaded: Opti
         obj = self.model.from_mongo(row, loaded=loaded)
         scope_ = scope
         for name in self._annotations:
-            if name in row:
-                f = scope_.fields.get(name) if scope_ is not None else None
-                value = row[name]
-                if f is not None and value is not None:
-                    try:
-                        value = f.from_mongo(value)
-                    except Exception:  # noqa: S110 - keep the raw value if it cannot be converted
-                        pass
-                setattr(obj, name, value)
+            f = scope_.fields.get(name) if scope_ is not None else None
+            value = row.get(name)  # a missing key means "no value": None, like Django
+            if f is not None and value is not None:
+                try:
+                    value = f.from_mongo(value)
+                except Exception:  # noqa: S110 - keep the raw value if it cannot be converted
+                    pass
+            setattr(obj, name, value)
         for name in self._select_related:
             _attach_related(obj, self.model, name, row)
         return obj
@@ -593,9 +592,8 @@ def iterator(self, chunk_size: int = 1000) -> Iterator[M]:
         else:
             cursor = coll.aggregate(
                 plan._final_stages(),
-                batchSize=chunk_size,
                 **connection.session_kwargs(self.model._meta.db_alias),
-                **plan._aggregate_kwargs(),
+                **plan._aggregate_kwargs(batch_size=chunk_size),
             )
         try:
             for r in cursor:
@@ -764,8 +762,7 @@ def bulk_create(
         kw = connection.session_kwargs(self.model._meta.db_alias)
         try:
             for docs in batches:
-                if docs:
-                    coll.insert_many(docs, ordered=ordered, **kw)
+                coll.insert_many(docs, ordered=ordered, **kw)
         except BulkWriteError as exc:
             raise NotUniqueError(str(exc.details.get("writeErrors", [{}])[0].get("errmsg", exc))) from exc
         self._finish_bulk(objs)
@@ -867,9 +864,8 @@ async def aiterator(self, chunk_size: int = 1000) -> AsyncIterator[M]:
         else:
             cursor = await coll.aggregate(
                 plan._final_stages(),
-                batchSize=chunk_size,
                 **connection.session_kwargs(self.model._meta.db_alias),
-                **plan._aggregate_kwargs(),
+                **plan._aggregate_kwargs(batch_size=chunk_size),
             )
         try:
             async for r in cursor:
@@ -983,8 +979,7 @@ async def abulk_create(
         kw = connection.session_kwargs(self.model._meta.db_alias)
         try:
             for docs in batches:
-                if docs:
-                    await coll.insert_many(docs, ordered=ordered, **kw)
+                await coll.insert_many(docs, ordered=ordered, **kw)
         except BulkWriteError as exc:
             raise NotUniqueError(str(exc.details.get("writeErrors", [{}])[0].get("errmsg", exc))) from exc
         self._finish_bulk(objs)
@@ -1040,9 +1035,7 @@ def _invert(o: Any) -> Any:
         return o[1:] if o.startswith("-") else "-" + o.lstrip("+")
     if isinstance(o, OrderBy):
         return OrderBy(o.expression, not o.descending)
-    if isinstance(o, F):
-        return OrderBy(o, True)
-    return o
+    return OrderBy(o, True)
 
 
 def _getattr_field(obj: Any, name: str) -> Any:
diff --git a/aggify/registry.py b/aggify/registry.py
index 7c4f25f..1be2767 100644
--- a/aggify/registry.py
+++ b/aggify/registry.py
@@ -9,12 +9,8 @@
 _models: Dict[str, List[type]] = {}
 
 
-def register(model: type) -> None:
-    keys = {model.__name__}
-    label = getattr(getattr(model, "_meta", None), "label", None)
-    if label:
-        keys.add(label)
-    for key in keys:
+def register(model: Any) -> None:
+    for key in {model.__name__, model._meta.label}:
         bucket = _models.setdefault(key, [])
         # re-definition of the same class name (autoreload, tests) replaces the old entry
         bucket[:] = [
@@ -47,8 +43,4 @@ def resolve(ref: Union[str, type], *, relative_to: Any = None) -> type:
     return candidates[0]
 
 
-def clear() -> None:  # used by tests
-    _models.clear()
-
-
 Model = Type[Any]
diff --git a/aggify/schema.py b/aggify/schema.py
index 3e41acd..e09bc74 100644
--- a/aggify/schema.py
+++ b/aggify/schema.py
@@ -17,7 +17,7 @@
 from typing import Any, Dict, Iterable, List, Mapping, Optional, Tuple
 
 from aggify.conf import settings
-from aggify.exceptions import InvalidField, InvalidLookup, UnsafeQueryError
+from aggify.exceptions import InvalidField, UnsafeQueryError
 from aggify.fields import AnyField, DictField, EmbeddedField, Field, ListField, ReferenceField
 from aggify.lookups import LOOKUPS
 
@@ -235,9 +235,6 @@ def resolve_path(
             cur = None
         i += 1
 
-    if lookup is not None and lookup not in LOOKUPS:
-        raise InvalidLookup(lookup, path)
-
     return Resolved(
         db_path=".".join(db_parts),
         field=field,
diff --git a/aggify/updates.py b/aggify/updates.py
index 52b7c7d..9568876 100644
--- a/aggify/updates.py
+++ b/aggify/updates.py
@@ -108,7 +108,7 @@ def build_update(model: Any, changes: Dict[str, Any]) -> Union[Dict[str, Any], L
             bucket[db_path] = value
         elif op == "rename":
             bucket[db_path] = validate_field_name(str(value), allow_dots=True)
-        elif op == "current_date":
+        else:  # current_date
             bucket[db_path] = True
     return ops
 
@@ -148,9 +148,7 @@ def _build_pipeline_update(
             set_[db_path] = {fn: [arr, [{"$literal": item}]]}
         else:
             raise InvalidArgument(f"{op}__ cannot be combined with expression assignments.")
-    stages: List[Dict[str, Any]] = []
-    if set_:
-        stages.append({"$set": set_})
+    stages: List[Dict[str, Any]] = [{"$set": set_}]  # an expression assignment is always present here
     if unset:
         stages.append({"$unset": unset})
     return stages
diff --git a/tests/contrib/conftest.py b/tests/contrib/conftest.py
new file mode 100644
index 0000000..d08654a
--- /dev/null
+++ b/tests/contrib/conftest.py
@@ -0,0 +1,19 @@
+"""One Django configuration for every test module in this package (settings are process-global)."""
+
+import django
+from django.conf import settings
+
+from tests.conftest import MONGO_URI
+
+if not settings.configured:
+    settings.configure(
+        SECRET_KEY="test",
+        USE_TZ=True,
+        DATABASES={"default": {"ENGINE": "django.db.backends.sqlite3", "NAME": ":memory:"}},
+        INSTALLED_APPS=["django.contrib.contenttypes", "aggify.contrib.django"],
+        ROOT_URLCONF="tests.contrib.django_urls",
+        ALLOWED_HOSTS=["*"],
+        AGGIFY={"django-test": {"URI": MONGO_URI, "NAME": "aggify_django_tests", "OPTIONS": {"maxPoolSize": 7}}},
+        DEFAULT_AUTO_FIELD="django.db.models.AutoField",
+    )
+    django.setup()
diff --git a/tests/contrib/test_cov_contrib.py b/tests/contrib/test_cov_contrib.py
new file mode 100644
index 0000000..6a87e45
--- /dev/null
+++ b/tests/contrib/test_cov_contrib.py
@@ -0,0 +1,254 @@
+"""Remaining branches of the framework integrations."""
+
+import asyncio
+import datetime as dt
+import uuid
+
+import django
+import grpc
+import pytest
+from django.conf import settings
+from django.test import override_settings
+
+from tests.conftest import MONGO_URI
+
+import aggify  # noqa: E402
+from aggify import (  # noqa: E402
+    BooleanField,
+    DateField,
+    DateTimeField,
+    DecimalField,
+    DictField,
+    Document,
+    EmbeddedDocument,
+    EmbeddedField,
+    EmailField,
+    FloatField,
+    IntField,
+    ListField,
+    ObjectIdField,
+    ReferenceField,
+    StringField,
+    URLField,
+    UUIDField,
+)
+from aggify.contrib.django import conf as dconf  # noqa: E402
+from aggify.contrib.django.fields import ObjectIdField as DjangoObjectIdField  # noqa: E402
+from aggify.contrib.django.forms import document_form  # noqa: E402
+from aggify.contrib.pydantic import pydantic_model  # noqa: E402
+from tests.models import Account  # noqa: E402
+
+
+class TestDjangoConf:
+    def test_mongodb_uri_shortcut(self):
+        with override_settings(AGGIFY=None, MONGODB_URI=MONGO_URI, MONGODB_NAME="aggify_conf_shortcut"):
+            out = dconf.setup_from_settings()
+        assert out["default"].db == "aggify_conf_shortcut"
+
+    def test_no_settings_is_a_noop(self):
+        with override_settings(AGGIFY=None, MONGODB_URI=None):
+            assert dconf.setup_from_settings() == {}
+
+    def test_invalid_settings(self):
+        from django.core.exceptions import ImproperlyConfigured
+
+        with override_settings(AGGIFY=["nope"]):
+            with pytest.raises(ImproperlyConfigured):
+                dconf.setup_from_settings()
+        with override_settings(AGGIFY={"x": {"NAME": "db"}}):
+            with pytest.raises(ImproperlyConfigured, match="URI"):
+                dconf.setup_from_settings()
+
+    def test_extra_aggify_settings(self):
+        with override_settings(
+            AGGIFY={"conf-extra": {"URI": MONGO_URI, "NAME": "aggify_conf_extra"}}, AGGIFY_SETTINGS={"max_limit": 7}
+        ):
+            try:
+                dconf.setup_from_settings()
+                assert aggify.settings.max_limit == 7
+            finally:
+                aggify.disconnect("conf-extra")
+
+
+@pytest.fixture(autouse=True)
+def _restore_default_connection(db_name):
+    yield
+    aggify.connect(MONGO_URI, db_name)
+
+
+def test_django_objectid_field_deconstruct():
+    name, path, args, kwargs = DjangoObjectIdField(db_index=True).deconstruct()
+    assert "max_length" not in kwargs and kwargs["db_index"] is True
+
+
+class Sub(EmbeddedDocument):
+    x = IntField()
+
+
+class Everything(Document):
+    text = StringField(verbose_name="Free text", choices=[("a", "Alpha"), "b"])
+    url = URLField()
+    ratio = FloatField(min_value=0, max_value=1)
+    money = DecimalField(max_digits=6, decimal_places=2)
+    when = DateTimeField()
+    day = DateField()
+    key = UUIDField()
+    oid = ObjectIdField()
+    ref = ReferenceField(Account)
+    blob = DictField()
+    items = ListField(StringField())
+    sub = EmbeddedField(Sub)
+    flag = BooleanField()
+    mail = EmailField()
+    anything = aggify.AnyField()
+    created = DateTimeField(auto_now_add=True)
+
+    class Meta:
+        collection = "everything_form"
+
+
+def test_form_field_mapping_covers_every_field_type():
+    from django import forms as dj
+
+    Form = document_form(Everything)
+    f = Form().fields
+    assert isinstance(f["text"], dj.ChoiceField) and f["text"].label == "Free text"
+    assert dict(f["text"].choices) == {"a": "Alpha", "b": "b"}
+    assert isinstance(f["url"], dj.URLField) and isinstance(f["ratio"], dj.FloatField)
+    assert isinstance(f["money"], dj.DecimalField) and isinstance(f["when"], dj.DateTimeField)
+    assert isinstance(f["day"], dj.DateField) and isinstance(f["key"], dj.UUIDField)
+    assert isinstance(f["oid"], dj.CharField) and isinstance(f["ref"], dj.CharField)
+    for name in ("blob", "items", "sub"):
+        assert isinstance(f[name], dj.JSONField)
+    assert isinstance(f["flag"], dj.BooleanField) and isinstance(f["mail"], dj.EmailField)
+    assert "anything" not in f and "created" not in f and "id" not in f
+
+
+def test_form_excludes_and_validates_through_the_document():
+    Form = document_form(Everything, exclude=["blob", "items", "sub", "oid", "ref", "key"], name="EForm")
+    assert Form.__name__ == "EForm"
+    bad = Form({"ratio": "5", "money": "1234567.1", "url": "ftp://x"})
+    assert not bad.is_valid() and {"ratio", "url"} <= set(bad.errors)
+    good = Form({"ratio": "0.5", "money": "1.25", "text": "a", "when": "2024-01-02 03:04", "day": "2024-01-02"})
+    assert good.is_valid(), good.errors
+    obj = good.save(commit=False)
+    assert obj.ratio == 0.5 and obj._state.adding
+
+
+async def test_async_middleware_and_markcoroutine():
+    from asgiref.sync import iscoroutinefunction
+    from django.http import HttpResponse
+
+    from aggify.contrib.django.middleware import AggifyAtomicMiddleware
+
+    async def view(request):
+        return HttpResponse("ok")
+
+    mw = AggifyAtomicMiddleware(view)
+    assert iscoroutinefunction(mw)
+    # no replica set needed to exercise the wiring: stub the transaction context
+    import contextlib
+    from unittest import mock
+
+    @contextlib.asynccontextmanager
+    async def fake(*a, **k):
+        yield None
+
+    with mock.patch("aggify.atransaction", fake):
+        response = await mw(object())
+    assert response.content == b"ok"
+
+
+class TestGrpcBranches:
+    async def test_stream_wrapper_maps_and_reraises(self):
+        from aggify.contrib.grpc import grpc_errors
+
+        aborted = []
+
+        class Ctx:
+            async def abort(self, code, message):
+                aborted.append(code)
+                raise RuntimeError("aborted")
+
+        class Svc:
+            @grpc_errors
+            async def Stream(self, request, context):
+                yield 1
+                raise aggify.ValidationError("bad")
+
+            @grpc_errors
+            async def Other(self, request, context):
+                yield 1
+                raise KeyError("not aggify")
+
+            @grpc_errors
+            async def Odd(self, request, context):
+                raise aggify.AggifyError("unmapped")
+                yield 0
+
+        got = []
+        with pytest.raises(RuntimeError, match="aborted"):
+            async for item in Svc().Stream(None, Ctx()):
+                got.append(item)
+        assert got == [1] and aborted == [grpc.StatusCode.INVALID_ARGUMENT]
+        with pytest.raises(KeyError):
+            async for _ in Svc().Other(None, Ctx()):
+                pass
+        with pytest.raises(aggify.AggifyError):
+            async for _ in Svc().Odd(None, Ctx()):
+                pass
+
+        class Plain:
+            @grpc_errors
+            async def Unary(self, request, context):
+                raise aggify.AggifyError("unmapped")
+
+        with pytest.raises(aggify.AggifyError):
+            await Plain().Unary(None, Ctx())
+
+
+class TestPydanticBranches:
+    def test_constraints_are_mapped(self):
+        class Strict(Document):
+            code = StringField(min_length=2, max_length=5, regex=r"^[a-z]+$")
+            score = FloatField(min_value=0, max_value=10)
+            n = IntField(min_value=1, max_value=3)
+            tags = ListField(StringField(), min_length=1, max_length=2)
+
+            class Meta:
+                collection = "strict_schema"
+
+        Schema = pydantic_model(Strict, name="StrictIn", for_input=True)
+        props = Schema.model_json_schema()["properties"]
+        assert props["code"]["anyOf"][0]["pattern"] == "^[a-z]+$"
+        assert props["score"]["anyOf"][0]["minimum"] == 0 and props["score"]["anyOf"][0]["maximum"] == 10
+        assert props["n"]["anyOf"][0]["maximum"] == 3
+        assert props["tags"]["anyOf"][0]["maxItems"] == 2 and props["tags"]["anyOf"][0]["minItems"] == 1
+
+    def test_include_required_defaults_and_recursion_guard(self):
+        class Defaulted(Document):
+            a = StringField(required=True, default="x")
+            b = IntField(default=3)
+
+            class Meta:
+                collection = "defaulted_schema"
+
+        Schema = pydantic_model(Defaulted, name="DefIn", for_input=True, include=["a", "b"])
+        assert Schema.model_validate({}).a == "x" or Schema.model_validate({}).a is None
+        assert Schema.model_validate({}).b == 3
+
+        class Loop(EmbeddedDocument):
+            pass
+
+        Loop._meta.fields["me"] = EmbeddedField(Loop)
+        Loop._meta.fields["me"].name = "me"
+        with pytest.raises(ValueError, match="Recursive"):
+            pydantic_model(Loop, name="LoopOut")
+
+
+async def test_form_asave_without_commit():
+    Form = document_form(Account, fields=["username"])
+    form = Form({"username": "nocommit"})
+    assert form.is_valid()
+    obj = await form.asave(commit=False)
+    assert obj._state.adding
diff --git a/tests/contrib/test_django.py b/tests/contrib/test_django.py
index 3bcebb4..efe6bb7 100644
--- a/tests/contrib/test_django.py
+++ b/tests/contrib/test_django.py
@@ -2,32 +2,14 @@
 
 import io
 
-import django
 import pytest
-from django.conf import settings
-
-from tests.conftest import MONGO_URI
-
-if not settings.configured:
-    settings.configure(
-        SECRET_KEY="test",
-        USE_TZ=True,
-        DATABASES={"default": {"ENGINE": "django.db.backends.sqlite3", "NAME": ":memory:"}},
-        INSTALLED_APPS=["django.contrib.contenttypes", "aggify.contrib.django"],
-        ROOT_URLCONF="tests.contrib.django_urls",
-        ALLOWED_HOSTS=["*"],
-        AGGIFY={"django-test": {"URI": MONGO_URI, "NAME": "aggify_django_tests", "OPTIONS": {"maxPoolSize": 7}}},
-        DEFAULT_AUTO_FIELD="django.db.models.AutoField",
-    )
-    django.setup()
-
-from django.core.management import call_command  # noqa: E402
-from django.core.paginator import Paginator  # noqa: E402
-from django.db import connection, models  # noqa: E402
-from django.http import Http404  # noqa: E402
-from django.test import AsyncClient, Client, RequestFactory  # noqa: E402
-
-import aggify  # noqa: E402
+from django.core.management import call_command
+from django.core.paginator import Paginator
+from django.db import connection, models
+from django.http import Http404
+from django.test import AsyncClient, Client, RequestFactory
+
+import aggify
 from aggify.contrib.django.fields import ObjectIdField  # noqa: E402
 from aggify.contrib.django.forms import document_form  # noqa: E402
 from aggify.contrib.django.shortcuts import aget_object_or_404, get_list_or_404, get_object_or_404  # noqa: E402
@@ -118,7 +100,7 @@ def test_management_command(clean_db):
     assert {"username", "age"} <= names
 
 
-@pytest.mark.django_db if False else pytest.mark.mongo
+@pytest.mark.mongo
 def test_objectid_field_in_django_orm_model(clean_db):
     class Invoice(models.Model):
         customer = ObjectIdField(db_index=True)
diff --git a/tests/integration/test_cov_db.py b/tests/integration/test_cov_db.py
new file mode 100644
index 0000000..952f4f7
--- /dev/null
+++ b/tests/integration/test_cov_db.py
@@ -0,0 +1,329 @@
+"""Database-backed tests for the remaining queryset / document / connection branches."""
+
+import asyncio
+import os
+
+import pytest
+
+import aggify
+from aggify import (
+    Aggify,
+    Count,
+    F,
+    InvalidArgument,
+    InvalidField,
+    NotUniqueError,
+    Q,
+    QuerySet,
+    Sum,
+    SynchronousOnlyOperation,
+)
+from aggify.exceptions import ConfigurationError
+from aggify.monitoring import capture_commands
+from tests.conftest import MONGO_URI
+from tests.models import Account, Comment, Post
+
+
+@pytest.fixture
+def seeded(clean_db):
+    accs = [Account.objects.create(username=f"user{i}", age=i, tags=["t"]) for i in range(4)]
+    for a in accs:
+        Post.objects.create(owner=a, caption=f"c{a.age}", likes=a.age)
+    return accs
+
+
+class TestConnectionBranches:
+    def test_bad_uri_and_missing_database(self):
+        with pytest.raises(ConfigurationError, match="Invalid MongoDB URI"):
+            aggify.connect("definitely not a uri", alias="bad")
+        with pytest.raises(ConfigurationError, match="No database name"):
+            aggify.connect("mongodb://localhost:27017", alias="nodb")
+
+    def test_reconnecting_with_other_options_replaces_client(self):
+        aggify.connect(MONGO_URI, "aggify_reconnect", alias="rc")
+        try:
+            first = aggify.get_client("rc")
+            aggify.connect(MONGO_URI, "aggify_reconnect", alias="rc", maxPoolSize=3)
+            assert aggify.get_client("rc") is not first and aggify.is_connected("rc")
+            assert not aggify.is_connected("never")
+        finally:
+            aggify.disconnect("rc")
+
+    async def test_unsafe_escape_hatch_env(self, monkeypatch):
+        monkeypatch.setenv("AGGIFY_ALLOW_ASYNC_UNSAFE", "1")
+        assert await asyncio.to_thread(lambda: True)
+        aggify.connection.check_sync_allowed()  # returns instead of raising inside the loop
+        monkeypatch.delenv("AGGIFY_ALLOW_ASYNC_UNSAFE")
+        with pytest.raises(SynchronousOnlyOperation):
+            aggify.connection.check_sync_allowed()
+
+    def test_async_client_needs_a_loop(self):
+        with pytest.raises(ConfigurationError, match="running event loop"):
+            aggify.get_async_client()
+
+    async def test_aclose_clients_and_nested_atransaction(self, clean_db):
+        await Account.objects.acount()
+        await aggify.aclose_clients()
+        await Account.objects.acount()  # lazily recreated
+        await aggify.aclose_clients("default")
+
+    @pytest.mark.replica
+    async def test_nested_atransaction_joins_outer(self, clean_db):
+        await Account.aensure_indexes()
+        async with aggify.atransaction() as outer:
+            async with aggify.atransaction() as inner:
+                assert inner is outer
+
+    async def test_adisconnect_alias_roundtrip(self):
+        aggify.connect(MONGO_URI, "aggify_tmp_disc", alias="tmpd")
+        client = aggify.get_async_client("tmpd")
+        assert client is not None
+        await aggify.adisconnect("tmpd")
+        assert not aggify.is_connected("tmpd")
+
+
+class TestDocumentBranches:
+    def test_fetch_related_sync(self, seeded):
+        post = Post.objects.get(caption="c1")
+        assert post.fetch_related("owner").owner.username == "user1"
+        post.fetch_related("owner")  # already loaded: nothing to do
+
+    async def test_async_save_of_deleted_document_reinserts_and_refresh_missing(self, seeded):
+        acc = seeded[0]
+        await Account.objects.filter(pk=acc.pk).adelete()
+        acc.age = 99
+        await acc.asave()
+        assert (await Account.objects.aget(pk=acc.pk)).age == 99
+        await acc.adelete()
+        with pytest.raises(Account.DoesNotExist):
+            await acc.arefresh_from_db()
+        await Account.adrop_collection()
+        assert await Account.objects.acount() == 0
+
+    async def test_async_save_without_changes_and_update_fields(self, seeded):
+        acc = await Account.objects.aget(username="user1")
+        acc.age, acc.display_name = 50, "x"
+        await acc.asave(update_fields=["age"])
+        again = await Account.objects.aget(username="user1")
+        assert (again.age, again.display_name) == (50, None)
+
+    def test_delete_pk_required(self):
+        with pytest.raises(ValueError):
+            Account(id=None, username="ab").delete()
+
+
+class TestQuerySetBranches:
+    def test_argument_validation(self, seeded):
+        with pytest.raises(InvalidArgument):
+            Account.objects.values_list("a", "b", flat=True)
+        with pytest.raises(InvalidArgument):
+            Account.objects.annotate(F("age"))
+        with pytest.raises(TypeError):
+            Account.objects.all()["x"]
+        with pytest.raises(InvalidField):
+            list(Account.objects.only("nope"))
+        with pytest.raises(InvalidField):
+            list(Account.objects.defer("nope"))
+        with pytest.raises(InvalidField):
+            list(Post.objects.select_related("caption"))
+        with pytest.raises(InvalidField):
+            list(Account.objects.prefetch_related("age"))
+        with pytest.raises(InvalidArgument):
+            list(Post.objects.values("caption").prefetch_related("owner"))
+        with pytest.raises(InvalidArgument):
+            Account.objects.latest()
+        with pytest.raises(InvalidArgument):
+            Account.objects.aggregate()
+        with pytest.raises(InvalidArgument):
+            Account.objects.order_by(5).query
+        with pytest.raises(InvalidArgument):
+            Account.objects.order_by(F("age") + 1).query
+
+    def test_options_hint_and_text_language(self, seeded):
+        Account.ensure_indexes()
+        qs = Account.objects.hint("username_1").options(max_time_ms=5000, batch_size=2, comment="c")
+        assert len(list(qs)) == 4
+        assert len(list(Account.objects.options(max_time_ms=5000).filter(F("age") > 1).annotate(x=F("age")))) == 2
+        assert (
+            Account.objects.search("x", language="english").query["find"]["filter"]["$text"]["$language"] == "english"
+        )
+
+    def test_default_max_time_is_applied(self, seeded):
+        aggify.configure(default_max_time_ms=9999)
+        with capture_commands() as log:
+            list(Account.objects.all())
+            Account.objects.count()
+            Account.objects.annotate(x=F("age")).count()
+            Account.objects.distinct("age")
+        assert all(r.command.get("maxTimeMS") == 9999 for r in log.records)
+
+    def test_cached_indexing_and_len_count(self, seeded):
+        qs = Account.objects.order_by("age")
+        list(qs)
+        assert qs[1].age == 1 and qs.count() == 4 and qs.list()[0].age == 0
+
+    def test_ordering_with_expressions(self, seeded):
+        assert [a.age for a in Account.objects.order_by(F("age").desc())] == [3, 2, 1, 0]
+        assert [a.age for a in Account.objects.order_by(F("age"))] == [0, 1, 2, 3]
+        assert [a.age for a in Account.objects.order_by(F("age")).reverse()] == [3, 2, 1, 0]
+        assert [a.age for a in Account.objects.order_by(F("age").desc()).reverse()] == [0, 1, 2, 3]
+
+    def test_pipeline_skip_with_annotation_and_select_related_with_only(self, seeded):
+        assert [a.age for a in Account.objects.annotate(x=F("age")).order_by("age")[2:]] == [2, 3]
+        post = Post.objects.select_related("owner").only("caption").get(caption="c2")
+        assert post.caption == "c2" and post.owner.username == "user2"
+
+    def test_nested_select_related(self, seeded):
+        class ChainNode(aggify.Document):
+            parent = aggify.ReferenceField("ChainNode")
+            name = aggify.StringField()
+
+            class Meta:
+                collection = "node_chain"
+
+        try:
+            root = ChainNode.objects.create(name="root")
+            mid = ChainNode.objects.create(name="mid", parent=root)
+            leaf = ChainNode.objects.create(name="leaf", parent=mid)
+            got = ChainNode.objects.select_related("parent__parent").get(pk=leaf.pk)
+            assert got.parent.parent.name == "root"
+            orphan = ChainNode.objects.select_related("parent__parent").get(pk=root.pk)
+            assert orphan.parent_id is None
+        finally:
+            ChainNode.drop_collection()
+
+    def test_explain(self, seeded):
+        assert "queryPlanner" in str(Account.objects.filter(age=1).order_by("age").explain())
+        assert Account.objects.annotate(x=F("age")).explain()
+
+    def test_none_iterators_and_streaming_aggregates(self, seeded):
+        assert list(Account.objects.none().iterator()) == []
+        assert [a.age for a in Account.objects.annotate(x=F("age")).order_by("age").iterator(chunk_size=2)] == [
+            0,
+            1,
+            2,
+            3,
+        ]
+
+    def test_count_variants(self, seeded):
+        aggify.configure(default_max_time_ms=1000)
+        assert Account.objects.order_by("age")[1:3].count() == 2
+        assert Account.objects.values("age").annotate(n=Count()).count() == 4
+
+    def test_get_or_create_race_is_resolved(self, seeded, monkeypatch):
+        Account.ensure_indexes()
+        real, calls = QuerySet.get, {"n": 0}
+
+        def fake(self, *a, **k):
+            calls["n"] += 1
+            if calls["n"] == 1:
+                raise Account.DoesNotExist()
+            return real(self, *a, **k)
+
+        monkeypatch.setattr(QuerySet, "get", fake)
+        obj, created = Account.objects.get_or_create(username="user1")
+        assert not created and obj.age == 1
+
+    def test_bulk_create_duplicate_error(self, seeded):
+        Account.ensure_indexes()
+        with pytest.raises(NotUniqueError):
+            Account.objects.bulk_create([Account(username="user1")])
+
+    def test_update_or_create_without_defaults(self, seeded):
+        obj, created = Account.objects.update_or_create(username="user1")
+        assert not created and obj.age == 1
+
+    def test_prefetch_with_nothing_to_fetch(self, clean_db):
+        Post.objects.create(caption="orphan")
+        assert [p.owner_id for p in Post.objects.prefetch_related("owner")] == [None]
+
+    def test_annotation_values_with_unconvertible_data(self, seeded):
+        class Bad(aggify.Document):
+            n = aggify.IntField()
+
+            class Meta:
+                collection = "bad_rows"
+
+        try:
+            Bad.objects.bulk_create([Bad(n=1)])
+            # write a value of the wrong type behind the model's back: loading must not crash
+            aggify.get_collection(Bad).update_many({}, {"$set": {"n": "oops"}})
+            assert list(Bad.objects.annotate(x=F("n")))[0].x == "oops"
+        finally:
+            Bad.drop_collection()
+
+
+class TestAsyncQuerySetBranches:
+    async def test_none_and_empty_paths(self, seeded):
+        assert await Account.objects.none().alist() == []
+        assert await Account.objects.none().acount() == 0 and not await Account.objects.none().aexists()
+        assert [x async for x in Account.objects.none().aiterator()] == []
+
+    async def test_count_with_pipeline_slice_and_limits(self, seeded):
+        aggify.configure(default_max_time_ms=1000)
+        assert await Account.objects.values("age").annotate(n=Count()).acount() == 4
+        assert await Account.objects.order_by("age")[1:3].acount() == 2
+        assert await Account.objects.annotate(x=F("age")).acount() == 4
+
+    async def test_get_variants_and_edges(self, seeded):
+        with pytest.raises(Account.MultipleObjectsReturned):
+            await Account.objects.aget(tags="t")
+        with pytest.raises(Account.DoesNotExist):
+            await Account.objects.aget(username="none")
+        assert (await Account.objects.alast()).username == "user3"
+        assert (await Account.objects.aearliest("age")).age == 0
+        assert (await Account.objects.alatest("age")).age == 3
+        with pytest.raises(InvalidArgument):
+            await Account.objects.alatest()
+        with pytest.raises(Account.DoesNotExist):
+            await Account.objects.filter(age=99).alatest("age")
+
+    async def test_aiterator_over_a_pipeline(self, seeded):
+        rows = [a.age async for a in Account.objects.annotate(x=F("age")).order_by("age").aiterator(chunk_size=2)]
+        assert rows == [0, 1, 2, 3]
+
+    async def test_explain_and_update_one(self, seeded):
+        assert "queryPlanner" in str(await Account.objects.filter(age=1).aexplain())
+        assert await Account.objects.annotate(x=F("age")).aexplain()
+        assert await Account.objects.filter(age=1).aupdate_one(age=10) == 1
+        assert await Account.objects.filter(username="fresh").aupdate_one(upsert=True, age=5) == 1
+
+    async def test_async_races_and_duplicates(self, seeded, monkeypatch):
+        await Account.aensure_indexes()
+        real, calls = QuerySet.aget, {"n": 0}
+
+        async def fake(self, *a, **k):
+            calls["n"] += 1
+            if calls["n"] == 1:
+                raise Account.DoesNotExist()
+            return await real(self, *a, **k)
+
+        monkeypatch.setattr(QuerySet, "aget", fake)
+        obj, created = await Account.objects.aget_or_create(username="user1")
+        assert not created
+        monkeypatch.undo()
+        with pytest.raises(NotUniqueError):
+            await Account.objects.abulk_create([Account(username="user1")])
+
+    async def test_async_prefetch_missing_and_distinct_values(self, seeded):
+        await Post.objects.acreate(caption="orphan")
+        posts = await Post.objects.prefetch_related("owner").order_by("caption").alist()
+        assert any(p.owner_id is None for p in posts)
+
+
+class TestPipelineExecution:
+    def test_iterate_and_options(self, seeded):
+        p = Aggify(Account).order_by("age").project(age=1, pk=0)
+        assert [r["age"] for r in p.iterate()] == [0, 1, 2, 3]
+        aggify.configure(default_max_time_ms=1000)
+        assert len(p.execute()) == 4
+        assert p.explain()
+
+    async def test_async_explain(self, seeded):
+        assert await Aggify(Account).filter(age=1).aexplain()
+        assert [r["age"] async for r in Aggify(Account).order_by("age").project(age=1, pk=0).aiterate(raw=True)] == [
+            0,
+            1,
+            2,
+            3,
+        ]
diff --git a/tests/integration/test_cov_db2.py b/tests/integration/test_cov_db2.py
new file mode 100644
index 0000000..32c3cac
--- /dev/null
+++ b/tests/integration/test_cov_db2.py
@@ -0,0 +1,131 @@
+"""Last database-backed branches."""
+
+import pytest
+
+import aggify
+from aggify import Aggify, Count, DictField, Document, F, IntField, Q, ReferenceField, StringField
+from tests.models import Account, Post
+
+
+@pytest.fixture
+def data(clean_db):
+    accs = [Account.objects.create(username=f"user{i}", age=i) for i in range(3)]
+    for a in accs:
+        Post.objects.create(owner=a, caption=f"c{a.age}")
+    return accs
+
+
+def test_empty_bulk_create_and_asave_noop(data):
+    assert Account.objects.bulk_create([]) == []
+
+
+async def test_async_branches(data):
+    assert await Account.objects.abulk_create([]) == []
+    acc = await Account.objects.aget(username="user1")
+    await acc.asave()  # nothing changed: no command
+    acc.age = None
+    await acc.asave()  # only $unset
+    assert (await Account.objects.aget(username="user1")).age is None
+    qs = Account.objects.order_by("username")
+    assert len(await qs.alist()) == 3 and len(await qs.alist()) == 3
+    obj, created = await Account.objects.aupdate_or_create(username="user2")
+    assert not created
+    # prefetch with nothing to fetch, and with an already-loaded relation
+    await Post.objects.filter(caption="c0").aupdate(unset__owner=True)
+    posts = await Post.objects.prefetch_related("owner").alist()
+    assert sum(p.owner_id is None for p in posts) == 1
+    p = await Post.objects.afirst()
+    await p.afetch_related("owner")
+    await p.afetch_related("owner")  # second call has nothing left to do
+    await Post.objects.filter(caption="c1").aupdate(unset__owner=True)
+    orphan = await Post.objects.aget(caption="c1")
+    await orphan.afetch_related("owner")
+
+
+def test_sync_fetch_related_with_missing_target(data):
+    post = Post.objects.get(caption="c2")
+    Account.objects.filter(pk=post.owner_id).delete()
+    owner_id = post.owner_id
+    assert post.fetch_related("owner").owner_id == owner_id  # dangling reference keeps its id
+
+
+def test_explain_without_sort_and_with_projection(data):
+    assert Post.objects.filter(caption="c1").explain()
+    assert Account.objects.filter(age=1).only("username").explain()
+
+
+async def test_aexplain_without_sort(data):
+    assert await Post.objects.filter(caption="c1").aexplain()
+
+
+def test_values_with_pk_and_defer_with_select_related(data):
+    rows = list(Account.objects.annotate(x=F("age")).values("pk", "x").order_by("x"))
+    assert rows[0]["x"] == 0 and "id" in rows[0]
+    post = Post.objects.select_related("owner").defer("location").get(caption="c1")
+    assert post.owner.username == "user1"
+
+
+def test_annotation_missing_from_some_rows(data):
+    Account.objects.filter(username="user0").update(unset__age=True)
+    got = {a.username: getattr(a, "x", "absent") for a in Account.objects.annotate(x=F("age"))}
+    assert got["user1"] == 1
+    assert got["user0"] is None
+
+
+def test_desc_expression_ordering_errors_and_reverse(data):
+    with pytest.raises(aggify.InvalidArgument):
+        Account.objects.order_by((F("age") + 1).desc()).query
+    assert [a.age for a in Account.objects.order_by(F("age").desc()).reverse()] == [0, 1, 2]
+
+
+def test_pipeline_with_explicit_scope_and_join_in_unwind(data):
+    p = Aggify(Post).unwind("owner__username")
+    assert p.to_list()[0]["$lookup"]
+    p2 = Aggify(Post).order_by("owner__age").limit(1)
+    assert len(p2.execute()) == 1
+
+
+def test_connection_helpers_for_unknown_aliases():
+    import asyncio
+
+    async def main():
+        await aggify.aclose_clients("nope")
+        await aggify.adisconnect("nope")
+
+    asyncio.run(main())
+
+
+async def test_async_prefetch_when_no_row_has_a_reference(data):
+    await Post.objects.aupdate(unset__owner=True)
+    assert all(p.owner_id is None for p in await Post.objects.prefetch_related("owner").alist())
+
+
+def test_child_overrides_inherited_manager(data):
+    class Parent(Document):
+        objects = aggify.Manager()
+
+    class Child(Parent):
+        objects = aggify.Manager()
+
+    assert Child.objects.model is Child
+
+
+def test_lazy_access_to_a_dangling_reference_returns_none_and_keeps_the_id(data):
+    post = Post.objects.get(caption="c1")
+    owner_id = post.owner_id
+    Account.objects.filter(pk=owner_id).delete()
+    assert post.owner is None and post.owner_id == owner_id
+
+
+async def test_async_fetch_of_a_dangling_reference(data):
+    post = await Post.objects.aget(caption="c1")
+    owner_id = post.owner_id
+    await Account.objects.filter(pk=owner_id).adelete()
+    await post.afetch_related("owner")
+    assert post.owner_id == owner_id
+
+
+def test_annotation_with_explicit_null_value(data):
+    aggify.get_collection(Account).update_many({"username": "user0"}, {"$set": {"deleted_at": None}})
+    got = {a.username: a.d for a in Account.objects.annotate(d=F("deleted_at"))}
+    assert got["user0"] is None
diff --git a/tests/integration/test_queryset.py b/tests/integration/test_queryset.py
index 26ce642..3519029 100644
--- a/tests/integration/test_queryset.py
+++ b/tests/integration/test_queryset.py
@@ -251,7 +251,7 @@ def test_values_and_values_list(self, data):
         assert list(Account.objects.order_by("username").values_list("username", "age"))[1] == ("bob", 25)
         row = list(Account.objects.order_by("username").values_list("username", "age", named=True))[0]
         assert (row.username, row.age) == ("alice", 30)
-        assert "id" in Account.objects.values("id", "username").first() if False else True
+        assert "id" in Account.objects.values("id", "username").first()
 
     def test_values_over_joins_and_embedded(self, data):
         rows = list(Post.objects.order_by("caption").values("caption", "owner__username"))
diff --git a/tests/models.py b/tests/models.py
index f430a0d..b4b6b65 100644
--- a/tests/models.py
+++ b/tests/models.py
@@ -5,8 +5,22 @@
 import enum
 
 from aggify import (
-    BooleanField, DateTimeField, DecimalField, DictField, Document, EmailField, EmbeddedDocument, EmbeddedField,
-    EnumField, FloatField, GeoPointField, Index, IntField, ListField, ReferenceField, StringField,
+    BooleanField,
+    DateTimeField,
+    DecimalField,
+    DictField,
+    Document,
+    EmailField,
+    EmbeddedDocument,
+    EmbeddedField,
+    EnumField,
+    FloatField,
+    GeoPointField,
+    Index,
+    IntField,
+    ListField,
+    ReferenceField,
+    StringField,
 )
 
 
diff --git a/tests/test_cov_final.py b/tests/test_cov_final.py
new file mode 100644
index 0000000..53952f6
--- /dev/null
+++ b/tests/test_cov_final.py
@@ -0,0 +1,247 @@
+"""The last uncovered branches (unit level)."""
+
+import asyncio
+
+import pytest
+from bson import Binary
+
+import aggify
+from aggify import (
+    Aggify,
+    DateToString,
+    DecimalField,
+    DictField,
+    Document,
+    EmbeddedDocument,
+    EmbeddedField,
+    F,
+    FloatField,
+    GeoPointField,
+    IntField,
+    ListField,
+    Manager,
+    Q,
+    ReferenceField,
+    StringField,
+    UUIDField,
+    ValidationError,
+)
+from aggify.expressions import DateAdd, DateDiff, DateTrunc, RegexMatch
+from aggify.fields import Field
+from aggify.schema import Context, Scope
+from aggify.updates import _element_field, _prepare, build_update
+from tests.models import Account, Post
+
+
+def res(expr, model=Account):
+    return expr.resolve(Context(Scope.for_model(model)))
+
+
+def test_field_base_and_float_edges():
+    assert Field().to_mongo(5) == 5
+    assert FloatField().to_python(True) is True
+    with pytest.raises(ValidationError):
+        FloatField().to_mongo(True)
+    with pytest.raises(ValidationError):
+        GeoPointField().to_mongo([1.0, 2.0])
+
+
+def test_reference_id_accessor_setter():
+    post = Post()
+    post.owner_id = Account(username="ab").pk
+    assert post.owner_id is not None and post.to_mongo()["owner_id"] == post.owner_id
+
+
+def test_reference_does_not_clobber_existing_id_attribute():
+    class Odd(Document):
+        thing = ReferenceField(Account)
+
+        @property
+        def thing_id(self):
+            return "mine"
+
+    # the class defined thing_id itself *after* the field was processed: ensure construction still works
+    assert Odd().thing is None
+
+    class Pre(Document):
+        thing_id = "custom"
+        thing = ReferenceField(Account)
+
+    assert Pre.thing_id == "custom"
+
+
+def test_list_of_fields_with_explicit_db_field_and_dict_none_values():
+    class L(Document):
+        xs = ListField(StringField(db_field="custom"))
+        d = DictField(IntField())
+
+    assert L._meta.fields["xs"].field.db_field == "custom"
+    L(d={"a": None, "b": 2}).full_clean()
+
+
+def test_diamond_inheritance_and_manager_inheritance():
+    class A(Document):
+        shared = StringField()
+
+        class Meta:
+            abstract = True
+
+    class B(Document):
+        shared = StringField()
+        other = IntField()
+
+        class Meta:
+            abstract = True
+
+    class C(A, B):
+        pass
+
+    assert list(C._meta.fields)[1:] == ["shared", "other"]
+
+    class Parent(Document):
+        things = Manager()
+
+    class Child(Parent):
+        pass
+
+    assert Child.things.model is Child and Parent.things.model is Parent
+
+
+def test_documents_unset_only_update_plan():
+    a = Account.from_mongo(Account(username="ab", age=3).to_mongo())
+    a.age = None
+    kind, flt, update = a._plan_save(False, None, False)
+    assert kind == "update" and update == {"$unset": {"age": ""}}
+
+
+def test_date_functions_without_optionals():
+    assert res(DateToString("created_at", "%Y", timezone="UTC"))["$dateToString"]["timezone"] == "UTC"
+    assert "timezone" not in res(DateAdd("created_at", "day", 1))["$dateAdd"]
+    assert "timezone" not in res(DateDiff("created_at", "created_at"))["$dateDiff"]
+    assert res(DateTrunc("created_at")) == {"$dateTrunc": {"date": "$created_at", "unit": "day"}}
+    assert res(DateTrunc("created_at", timezone="UTC"))["$dateTrunc"]["timezone"] == "UTC"
+    assert res(DateTrunc("created_at", start_of_week="mon"))["$dateTrunc"]["startOfWeek"] == "mon"
+    assert "options" not in res(RegexMatch("username", "^a"))["$regexMatch"]
+
+
+def test_near_without_distance():
+    assert Account.objects.filter(address__geo__near=(1, 2)).query["find"]["filter"] == {
+        "address.geo": {"$near": {"$geometry": {"type": "Point", "coordinates": [1, 2]}}}
+    }
+
+
+def test_pipeline_branches():
+    custom = Scope.for_model(Account).evolve(only=["age"])
+    assert set(Aggify(Account, scope=custom).scope.fields) == {"age"}
+    assert Aggify(Account).add_fields().to_list() == []
+    assert Aggify(Account).unwind("tags", include_array_index="i").to_list() == [
+        {"$unwind": {"path": "$tags", "includeArrayIndex": "i"}}
+    ]
+    assert Aggify(Account).unwind("address__city").to_list() == [{"$unwind": "$address.city"}]
+    p = Aggify(Account).lookup(Post, "empty", let={}, query=Q())
+    assert p.to_list()[0]["$lookup"]["pipeline"] == []
+    assert "sortBy" not in Aggify(Account).set_window_fields(rank=aggify.Rank()).to_list()[0]["$setWindowFields"]
+    assert Aggify(Account).replace_with("address").to_list() == [{"$replaceWith": "$address"}]
+
+
+def test_pipeline_convert_survives_bad_values():
+    from aggify.pipeline import convert_row
+
+    class Keyed(Document):
+        key = UUIDField()
+
+    bad = Binary(b"x" * 16, 3)  # legacy UUID subtype cannot be decoded with the standard representation
+    assert convert_row(Scope.for_model(Keyed), {"key": bad})["key"] == bad
+
+
+def test_queryset_hydration_survives_bad_annotation_values():
+    from aggify.queryset import QuerySet
+
+    class Keyed(Document):
+        key = UUIDField()
+
+        class Meta:
+            collection = "keyed_unit"
+
+    qs = Keyed.objects.annotate(k=F("key"))
+    scope = Scope.for_model(Keyed).evolve({"k": Keyed._meta.fields["key"]})
+    obj = QuerySet._hydrate_row(
+        qs,
+        {
+            "_id": aggify.ObjectId() if hasattr(aggify, "ObjectId") else __import__("bson").ObjectId(),
+            "k": Binary(b"x" * 16, 3),
+        },
+        scope,
+        None,
+    )
+    assert obj.k == Binary(b"x" * 16, 3)
+
+
+def test_update_builder_branches():
+    with pytest.raises(aggify.InvalidArgument, match="updated twice"):
+        build_update(Account, {"age": 1, "set__age": 2})
+    with pytest.raises(ValidationError):
+        build_update(Account, {"age": F("age") + 1, "inc__score": "x"})
+    assert build_update(Account, {"age": F("age") + 1, "min__score": 3.0}) == [
+        {"$set": {"age": {"$add": ["$age", 1]}, "score": {"$min": ["$score", 3.0]}}}
+    ]
+    assert _prepare(None, {"a": 1}) == {"a": 1} and _prepare(None, aggify.Raw({"$x": 1})) == {"$x": 1}
+    assert isinstance(_element_field(None, "p"), aggify.AnyField)
+    only_unset = build_update(Account, {"age": F("age") + 1, "unset__score": True})
+    assert only_unset[1] == {"$unset": ["score"]}
+    only_unset = build_update(Account, {"unset__score": True, "age": F("age") + 1})
+    assert only_unset[0]["$set"] and only_unset[1]
+    assert build_update(Account, {"current_date__created_at": True}) == {"$currentDate": {"created_at": True}}
+    mm = build_update(Account, {"age": F("age") + 1, "max__score": 3.0})
+    assert mm[0]["$set"]["score"] == {"$max": ["$score", 3.0]}
+
+
+def test_signal_asend_with_sync_receiver():
+    sig = aggify.Signal("x")
+    seen = []
+    sig.connect(lambda sender, **kw: seen.append(sender))
+    asyncio.run(sig.asend("s"))
+    assert seen == ["s"]
+
+
+def test_monitoring_matches_the_right_request():
+    from types import SimpleNamespace as NS
+
+    from aggify.monitoring import LISTENER, capture_commands
+
+    with capture_commands() as log:
+        for rid in (1, 2):
+            LISTENER.started(NS(command_name="find", database_name="d", command={"find": "c"}, request_id=rid))
+        LISTENER.succeeded(NS(request_id=1, duration_micros=9))
+        LISTENER.failed(NS(request_id=1))
+    assert log.records[0].duration_micros == 9 and log.records[0].failed and not log.records[1].failed
+    with capture_commands() as log2:
+        LISTENER.started(NS(command_name="find", database_name="d", command={"find": "c"}, request_id=5))
+        LISTENER.succeeded(NS(request_id=99, duration_micros=1))
+        LISTENER.failed(NS(request_id=99))
+    assert log2.records[0].duration_micros is None and not log2.records[0].failed
+
+
+def test_pydantic_decimal_without_digits():
+    from aggify.contrib.pydantic import pydantic_model
+
+    class Money(Document):
+        a = DecimalField()
+        b = DecimalField(max_digits=5)
+
+        class Meta:
+            collection = "money_schema"
+
+    assert set(pydantic_model(Money, name="MoneyOut").model_fields) == {"id", "a", "b"}
+
+
+def test_embedded_documents_inside_dict_field_resolve():
+    class Inner(EmbeddedDocument):
+        n = IntField()
+
+    class Holder2(Document):
+        by = DictField(EmbeddedField(Inner))
+
+    from aggify.schema import resolve_path
+
+    assert resolve_path(Scope.for_model(Holder2), "by__k__n").db_path == "by.k.n"
diff --git a/tests/test_cov_unit.py b/tests/test_cov_unit.py
new file mode 100644
index 0000000..0114e18
--- /dev/null
+++ b/tests/test_cov_unit.py
@@ -0,0 +1,744 @@
+"""Branch-level unit tests (no database): the less travelled paths of every module."""
+
+import datetime as dt
+import decimal
+import enum
+import uuid
+
+import pytest
+from bson import Binary, Decimal128, Int64, ObjectId
+
+import aggify
+from aggify import (
+    AddToSet,
+    AlreadyExistsField,
+    AnnotationError,
+    Avg,
+    BinaryField,
+    BooleanField,
+    Case,
+    Cond,
+    Count,
+    DateAdd,
+    DateDiff,
+    DatePart,
+    DateTrunc,
+    DecimalField,
+    DictField,
+    DocumentNumber,
+    Document,
+    EmbeddedField,
+    EnumField,
+    F,
+    FloatField,
+    Func,
+    GeoPointField,
+    IntField,
+    InvalidArgument,
+    InvalidField,
+    InvalidLookup,
+    InvalidProjection,
+    Last,
+    Let,
+    ListField,
+    Max,
+    MergeObjects,
+    Median,
+    Min,
+    MongoIndexError,
+    ObjectIdField,
+    Pipeline,
+    Push,
+    Q,
+    Raw,
+    ReferenceField,
+    Rank,
+    ReplaceAll,
+    Shift,
+    StdDevPop,
+    StdDevSamp,
+    StringField,
+    Sum,
+    URLField,
+    UUIDField,
+    ValidationError,
+    Value,
+    Var,
+    When,
+    Aggify,
+    RegexMatch,
+    Reduce,
+    Trim,
+    Rand,
+    Year,
+)
+from aggify import exceptions as exc
+from aggify import lookups, registry, security
+from aggify.fields import Field, NOT_PROVIDED
+from aggify.schema import Context, Scope, dynamic_field, resolve_path
+from tests.models import Account, Address, Post, Role
+
+
+def res(expr, model=Account):
+    return expr.resolve(Context(Scope.for_model(model)))
+
+
+class TestExceptions:
+    def test_validation_error_shapes(self):
+        e = exc.ValidationError({"a": "bad", "b": ["x", "y"]})
+        assert e.errors == {"a": ["bad"], "b": ["x", "y"]} and "a: bad" in str(e)
+        assert e.prefixed("p").errors == {"p.a": ["bad"], "p.b": ["x", "y"]}
+        assert exc.ValidationError("plain").prefixed("p").errors == {"p": ["plain"]}
+        assert exc.ValidationError("m", field="f").as_dict() == {"f": ["m"]}
+
+    def test_messages(self):
+        assert "one of" in str(exc.InvalidArgument(expected_list=["a"]))
+        assert "aggregation" in str(exc.InvalidAnnotateExpression())
+        assert "mix" in str(exc.InvalidProjection())
+        assert "Model" in str(exc.InvalidField("x", Account)) or "Account" in str(exc.InvalidField("x", Account))
+        assert str(exc.AlreadyExistsField("x")) and str(exc.InvalidEmbeddedField("x"))
+        assert str(exc.AggifyValueError([int], "s")) and str(exc.MongoIndexError())
+        assert str(exc.OutStageError("$x")) and str(exc.ConnectionNotConfigured("z"))
+        assert issubclass(exc.AggifyBaseException, Exception)
+
+
+class TestLookupsModule:
+    def test_registry_helpers(self):
+        assert lookups.is_lookup("gte") and not lookups.is_lookup("nope")
+        with pytest.raises(InvalidLookup):
+            lookups.get_lookup("nope", "f")
+
+    def test_expr_forms(self):
+        L = lookups.LOOKUPS
+        assert L["range"].to_expr("$a", [1, 2]) == {"$and": [{"$gte": ["$a", 1]}, {"$lte": ["$a", 2]}]}
+        assert L["nin"].to_expr("$a", [1]) == {"$not": [{"$in": ["$a", [1]]}]}
+        assert L["icontains"].to_expr("$a", "x") == {"$regexMatch": {"input": "$a", "regex": "x", "options": "i"}}
+        assert L["elem_match"].to_query("a", {"x": 1}) == {"a": {"$elemMatch": {"x": 1}}}
+        with pytest.raises(InvalidLookup):
+            lookups.Lookup().to_expr("$a", 1)
+
+    @pytest.mark.parametrize(
+        "kind,value",
+        [
+            ("int", -1),
+            ("int", True),
+            ("bool", "x"),
+            ("range", [1]),
+            ("string", 3),
+            ("regex", 3),
+            ("mod", [1]),
+            ("near", [1]),
+            ("near", ["a", "b"]),
+        ],
+    )
+    def test_prepare_kind_rejections(self, kind, value):
+        with pytest.raises(ValidationError):
+            lookups.prepare_kind(kind, "x", value, lambda v: v)
+
+    def test_regex_accepts_compiled_patterns(self):
+        import re
+
+        assert lookups.prepare_kind("regex", "regex", re.compile("^a"), str) == "^a"
+        assert lookups.prepare_kind("query", "x", {"a": 1}, str) == {"a": 1}
+
+
+class TestSchemaAndRegistry:
+    def test_scope_api(self):
+        s = Scope.for_model(Account)
+        assert s.has("age") and "age" in repr(s) and repr(Scope({}, open=True)).endswith("open)")
+        assert set(s.evolve(only=["age"]).fields) == {"age"}
+        assert s.get("_id").name == "id"
+        assert isinstance(Scope({}, open=True).get("anything"), Field)
+        assert dynamic_field("x", IntField()).name == "x"
+
+    def test_resolve_errors(self):
+        s = Scope.for_model(Account)
+        with pytest.raises(InvalidField):
+            resolve_path(s, "")
+        with pytest.raises(InvalidField):
+            resolve_path(s, "age__x__y")
+        with pytest.raises(InvalidField):
+            resolve_path(Scope.for_model(Post), "owner__username", allow_joins=False)
+        ctx = Context(Scope.for_model(Post), allow_joins=False)
+        with pytest.raises(InvalidField):
+            ctx.add_joins([aggify.schema.Join("a", Account, "x")])
+
+    def test_dict_of_embedded_documents(self):
+        class Holder(Document):
+            by_name = DictField(EmbeddedField(Address))
+
+        r = resolve_path(Scope.for_model(Holder), "by_name__home__city")
+        assert r.db_path == "by_name.home.city"
+
+    def test_registry(self):
+        class Solo(Document):
+            pass
+
+        assert registry.resolve(Solo) is Solo
+        assert registry.resolve("self", relative_to=Solo) is Solo
+        assert registry.resolve("tests.models.Account") is Account
+        with pytest.raises(aggify.ConfigurationError):
+            registry.resolve("Missing")
+
+
+class TestSecurityModule:
+    def test_raw_repr_and_unwrap(self):
+        assert repr(Raw({"a": 1})) == "Raw({'a': 1})"
+        assert security.assert_safe_value(Raw({"$x": 1})) == {"$x": 1}
+        assert security.assert_safe_value({"a": [1, (2, 3)]}) == {"a": [1, (2, 3)]}
+        with pytest.raises(aggify.UnsafeQueryError):
+            security.ensure_iterable("abc")
+
+    def test_mask_uri_without_credentials(self):
+        assert security.mask_uri("mongodb://h:1/x") == "mongodb://h:1/x"
+        assert "***" in security.mask_uri("mongodb://u:p@h/x")
+
+
+class TestIndexSpecs:
+    def test_all_options_compile(self):
+        from aggify import Index
+
+        class Spec(Document):
+            title = StringField()
+            body = StringField()
+
+        doc = Index("title", "-body", name="my_idx", collation={"locale": "en"}).to_model(Spec).document
+        assert doc["name"] == "my_idx" and doc["collation"] == {"locale": "en"}
+        text = Index("title", type="text", weights={"title": 5}, default_language="english").to_model(Spec).document
+        assert text["weights"] == {"title": 5} and text["default_language"] == "english"
+
+
+class TestFieldEdges:
+    def test_base_field(self):
+        f = Field()
+        assert f.to_python(3) == 3 and f.get_default() is None and "Field" in repr(f)
+        assert repr(NOT_PROVIDED) == "NOT_PROVIDED" and not NOT_PROVIDED
+        assert Field(default=5).get_default() == 5
+        assert f.deconstruct_kwargs()["required"] is False
+        named = StringField()
+        named.name = "x"
+        assert repr(named) == "<StringField: x>"
+        assert Account.age.__class__ is IntField  # descriptor on the class returns the field
+
+    def test_validator_raising_validation_error_is_preserved(self):
+        def v(value):
+            raise ValidationError("custom")
+
+        with pytest.raises(ValidationError, match="custom"):
+            StringField(validators=[v]).clean("a")
+
+    def test_choices_variants(self):
+        class Color(enum.Enum):
+            RED = "r"
+
+        f = StringField(choices=Color)
+        assert "r" in f.choice_values() and Color.RED in f.choice_values()
+        assert StringField().choice_values() == []
+
+    def test_raw_passthrough_in_prepare(self):
+        for f in (
+            StringField(),
+            ListField(StringField()),
+            DictField(),
+            EmbeddedField(Address),
+            GeoPointField(),
+            aggify.AnyField(),
+        ):
+            assert f.prepare_value(Raw({"$x": 1})) == {"$x": 1}
+        assert StringField().prepare_lookup("exact", Raw("z")) == "z"
+
+    def test_conversions(self):
+        assert ObjectIdField().to_python(b"123456789012") == ObjectId(b"123456789012")
+        assert StringField().to_python(b"abc") == "abc" and StringField().to_python(b"\xff") == b"\xff"
+        assert IntField().to_python(Int64(5)) == 5 and IntField().to_python("x") == "x"
+        assert FloatField().to_python("x") == "x" and FloatField().to_python(decimal.Decimal("1.5")) == 1.5
+        assert FloatField().to_python([1]) == [1]
+        with pytest.raises(ValidationError):
+            FloatField(max_value=1).clean(2)
+        assert DecimalField().to_python(1.25) == decimal.Decimal("1.25")
+        assert DecimalField().to_python("zz") == "zz" and DecimalField().to_python([1]) == [1]
+        assert DecimalField().to_python(True) is True
+        assert DecimalField().from_mongo(Decimal128("1")) == decimal.Decimal("1")
+        with pytest.raises(ValidationError):
+            DecimalField(min_value=1).clean("0.5")
+        with pytest.raises(ValidationError):
+            DecimalField(max_value=1).clean("2")
+        assert URLField().to_python("x") == "x"
+        with pytest.raises(ValidationError):
+            URLField().clean("http://[::1")
+        assert DateTimeField_to_python()
+
+    def test_date_edges(self):
+        from aggify import DateField, DateTimeField
+
+        assert DateTimeField().to_python(dt.date(2024, 1, 2)) == dt.datetime(2024, 1, 2, tzinfo=dt.timezone.utc)
+        assert DateTimeField().to_python(0).year == 1970 and DateTimeField().to_python(10**30) == 10**30
+        assert DateTimeField().to_python([1]) == [1]
+        aware = dt.datetime(2024, 1, 2, 3, tzinfo=dt.timezone(dt.timedelta(hours=2)))
+        assert DateField().to_python(aware) == dt.date(2024, 1, 2)
+        assert DateField().to_python(dt.datetime(2024, 1, 2, 3)) == dt.date(2024, 1, 2)
+        assert DateField().to_python(dt.date(2024, 1, 2)) == dt.date(2024, 1, 2)
+        assert DateField().to_python("nope") == "nope" and DateField().to_python(5) == 5
+
+    def test_binary_uuid_enum_edges(self):
+        u = uuid.uuid4()
+        assert UUIDField().to_python(Binary.from_uuid(u)) == u and UUIDField().to_python(5) == 5
+        assert BinaryField().to_python(Binary(b"x")) == b"x" and BinaryField().to_python(bytearray(b"y")) == b"y"
+        with pytest.raises(ValidationError):
+            BinaryField().to_mongo("not bytes")
+        assert EnumField(Role).python_type is Role and EnumField(Role).to_python("admin") is Role.ADMIN
+        assert EnumField(Role).to_python("zzz") == "zzz"
+
+    def test_list_edges(self):
+        f = ListField(IntField())
+        assert f.to_python((1, "2")) == [1, 2] and f.to_python({1}) == [1] and f.to_python("x") == "x"
+        assert f.from_mongo("x") == "x" and f.from_mongo([1, None]) == [1, None]
+        assert f.prepare_value([1, "2"]) == [1, 2] and f.prepare_value("3") == 3
+        assert f._prepare_exact("3") == 3 and f._prepare_exact((1, 2)) == [1, 2]
+        assert f.lookup_field("exact", [1]) is f and f.lookup_field("gt", 1) is f.field
+        aggify.configure(max_in_list=1)
+        with pytest.raises(ValidationError):
+            f.to_mongo(list(range(30)))
+
+    def test_list_item_errors_are_indexed(self):
+        class Bag(Document):
+            nums = ListField(IntField(min_value=5))
+
+        with pytest.raises(ValidationError) as ei:
+            Bag(nums=[9, 1, None]).full_clean()
+        assert any("nums[1]" in k for k in ei.value.errors)
+
+    def test_dict_edges(self):
+        f = DictField(IntField())
+        assert f.from_mongo({"a": 1, "b": None}) == {"a": 1, "b": None} and f.from_mongo(3) == 3
+        assert f.to_python(3) == 3
+        with pytest.raises(ValidationError):
+            f.to_mongo([1])
+        assert f.lookup_field("exact", {"a": 1}) is f and f.lookup_field("gt", 1) is f.field
+        assert f.prepare_value({"a": "1"}) == {"a": 1} and f.prepare_value("5") == 5
+        assert f._prepare_exact({"a": "1"}) == {"a": 1} and f._prepare_exact("5") == 5
+        with pytest.raises(ValidationError):
+            DictField(IntField(min_value=3)).clean({"a": 1})
+
+    def test_embedded_edges(self):
+        f = EmbeddedField(Address)
+        f.name = "address"
+        assert f.from_mongo(3) == 3 and f.to_python(3) == 3
+        assert f.prepare_value({"city": "x"}) == {"city": "x"}
+        assert f.prepare_value(Address(city="y")) == {"city": "y"}
+        with pytest.raises(ValidationError):
+            f.to_mongo({"city": "x"})
+        lazy = EmbeddedField("Address")
+        lazy.model = Account
+        assert lazy.document_type is Address and lazy.python_type is Address
+
+    def test_reference_edges(self):
+        f = Post._meta.fields["owner"]
+        assert f.python_type is Account and f.pk_field.name == "id"
+        assert Post.owner_id is not None and Post.__dict__["owner_id"].__get__(None, Post) is Post.__dict__["owner_id"]
+        assert f.__get__(None, Post) is f
+        with pytest.raises(ValidationError, match="Save the referenced"):
+            f.to_mongo(Account(id=None))
+        with pytest.raises(ValidationError):
+            f.to_mongo("not-an-id")
+        with pytest.raises(ValidationError):
+            f.to_mongo(Post())
+        assert f.from_mongo(ObjectId()) is not None
+
+    def test_geo_edges(self):
+        f = GeoPointField()
+        assert f.to_python({"type": "Point", "coordinates": [1, 2]}) == (1.0, 2.0)
+        assert f.to_python(["a", "b"]) == ["a", "b"] and f.to_python(5) == 5
+        assert f.prepare_value((1.0, 2.0)) == {"type": "Point", "coordinates": [1.0, 2.0]}
+        assert f.prepare_value(Raw(1)) == 1
+
+
+def DateTimeField_to_python():
+    from aggify import DateTimeField
+
+    return DateTimeField().to_python("2024-01-01T00:00:00Z").year == 2024
+
+
+class TestExpressionEdges:
+    def test_operator_variants(self):
+        assert res(2 * F("age")) == {"$multiply": [2, "$age"]}
+        assert res(8 / F("age")) == {"$divide": [8, "$age"]}
+        assert res(F("age") % 2) == {"$mod": ["$age", 2]}
+        assert res(7 % F("age")) == {"$mod": [7, "$age"]}
+        assert res(F("age") ** 2) == {"$pow": ["$age", 2]}
+        assert res(abs(F("age"))) == {"$abs": "$age"}
+        assert res(1 + F("age")) == {"$add": [1, "$age"]}
+        assert res((F("age") > 1).eq(True)) == {"$eq": [{"$gt": ["$age", 1]}, True]}
+
+    def test_boolean_with_q_defers_to_q(self):
+        combined = (F("age") > 1) & Q(username="x")
+        assert isinstance(combined, Q)
+        assert isinstance((F("age") > 1) | Q(username="x"), Q)
+        assert isinstance(Q(username="x") & (F("age") > 1), Q)
+        assert isinstance(Q(username="x") | (F("age") > 1), Q)
+        assert isinstance((F("age") > 1) & (F("age") < 3), aggify.expressions.BooleanExpression)
+
+    def test_ordering_markers(self):
+        assert F("age").asc().descending is False and F("age").desc().descending is True
+
+    def test_value_types_and_repr(self):
+        ctx = Context(Scope.for_model(Account))
+        assert repr(Value(3)) == "Value(3)" and repr(F("age")) == "F('age')"
+        for v, cls in [(True, BooleanField), (1.5, FloatField), ("s", StringField), (2, IntField)]:
+            assert isinstance(Value(v).output_field(ctx), cls)
+        assert Value(dt.datetime.now()).output_field(ctx).__class__.__name__ == "DateTimeField"
+        assert Value(object()).output_field(ctx) is None
+        assert Value(5, output_field=FloatField()).output_field(ctx).__class__ is FloatField
+        assert Value(Raw({"$x": 1})).value == {"$x": 1}
+
+    def test_f_validation_and_output_field(self):
+        ctx = Context(Scope.for_model(Account))
+        with pytest.raises(InvalidArgument):
+            F("")
+        assert F("nope").output_field(ctx) is None and F("age").output_field(ctx).__class__ is IntField
+        assert res(F("age").last()) == {"$last": "$age"} and res(F("age").min()) == {"$min": "$age"}
+        assert res(F("age").max()) == {"$max": "$age"} and res(F("age").avg()) == {"$avg": "$age"}
+        with pytest.raises(InvalidArgument):
+            Var("$bad")
+
+    def test_output_field_inference(self):
+        ctx = Context(Scope.for_model(Account))
+        assert (F("score") + 1).output_field(ctx).__class__ is FloatField
+        assert (F("age") / 2).output_field(ctx).__class__ is FloatField
+        assert (F("age") + 1).output_field(ctx).__class__ is IntField
+        assert (F("age") > 1).output_field(ctx).__class__ is BooleanField
+        assert ((F("age") > 1) & (F("age") < 2)).output_field(ctx).__class__ is BooleanField
+        assert aggify.expressions.QExpression(Q(age=1)).output_field(ctx).__class__ is BooleanField
+        assert Func("$x", "age", output_field=IntField()).output_field(ctx).__class__ is IntField
+        assert aggify.Upper("username").output_field(ctx).__class__ is StringField
+        assert Func("$x", "age").output_field(ctx) is None and "Func" in repr(Func("$x", "age"))
+        assert Cond(Q(age=1), F("age"), 0).output_field(ctx).__class__ is IntField
+        assert Case(When(Q(age=1), F("age"))).output_field(ctx).__class__ is IntField
+
+    def test_func_construction_errors(self):
+        with pytest.raises(InvalidArgument):
+            Func()
+        with pytest.raises(InvalidArgument):
+            DateToString_too_many()
+        assert res(Rand()) == {"$rand": {}}
+
+    def test_date_functions(self):
+        assert res(DateTrunc("created_at", "hour", bin_size=2, timezone="UTC", start_of_week="mon")) == {
+            "$dateTrunc": {"date": "$created_at", "unit": "hour", "binSize": 2, "timezone": "UTC", "startOfWeek": "mon"}
+        }
+        assert res(DateAdd("created_at", "day", 3, timezone="UTC")) == {
+            "$dateAdd": {"startDate": "$created_at", "unit": "day", "amount": 3, "timezone": "UTC"}
+        }
+        assert res(DateDiff("created_at", "created_at", "hour", timezone="UTC"))["$dateDiff"]["timezone"] == "UTC"
+
+    def test_string_and_higher_order(self):
+        assert res(ReplaceAll("username", "a", "b")) == {
+            "$replaceAll": {"input": "$username", "find": "a", "replacement": "b"}
+        }
+        assert res(RegexMatch("username", "^a", "i"))["$regexMatch"]["options"] == "i"
+        with pytest.raises(ValidationError):
+            RegexMatch("username", "(")
+        assert res(Trim("username", chars=" ")) == {"$trim": {"input": "$username", "chars": " "}}
+        assert res(aggify.Filter("tags", Var("item").ne("x"), limit=2))["$filter"]["limit"] == 2
+        assert res(Reduce("tags", 0, Var("value") + 1)) == {
+            "$reduce": {"input": "$tags", "initialValue": 0, "in": {"$add": ["$$value", 1]}}
+        }
+        assert res(Let({"a": F("age")}, Var("a") + 1)) == {"$let": {"vars": {"a": "$age"}, "in": {"$add": ["$$a", 1]}}}
+
+    def test_cond_forms(self):
+        assert res(Cond(Q(age=1), 1, 2))["$cond"]["if"] == {"$eq": ["$age", 1]}
+        assert res(Cond(F("age") > 1, 1, 2))["$cond"]["if"] == {"$gt": ["$age", 1]}
+        assert res(Cond(Q(age=1)))["$cond"]["then"] is None
+        with pytest.raises(InvalidArgument):
+            Cond(1, 2)
+        with pytest.raises(InvalidArgument):
+            When()
+        assert res(When(Q(age=1), then=2).condition) == {"$eq": ["$age", 1]}
+        assert When(Q(age=1), then=2, username="x").condition is not None
+
+    def test_aggregates(self):
+        assert res(Median("age")) == {"$median": {"input": "$age", "method": "approximate"}}
+        assert res(Sum("age", window={"documents": [-1, 0]})) == {"$sum": "$age", "window": {"documents": [-1, 0]}}
+        assert res(Count(window={"range": [1, 2]}))["window"] == {"range": [1, 2]}
+        assert res(Count("age", filter=Q(age__gt=1)))["$sum"]["$cond"][0]["$and"]
+        assert res(Rank()) == {"$rank": {}} and res(DocumentNumber()) == {"$documentNumber": {}}
+        assert res(Shift("age", -1, 0)) == {"$shift": {"output": "$age", "by": -1, "default": 0}}
+        with pytest.raises(InvalidArgument):
+            Sum("age", "score", filter=Q(age=1)).resolve(Context(Scope.for_model(Account)))
+        ctx = Context(Scope.for_model(Account))
+        assert Sum("age").output_field(ctx).__class__ is IntField
+        assert Sum(Value(object())).output_field(ctx).__class__ is FloatField
+        assert Max("age", output_field=FloatField()).output_field(ctx).__class__ is FloatField
+        assert Avg("age").output_field(ctx).__class__ is FloatField and Count().output_field(ctx).__class__ is IntField
+        assert aggify.expressions.Aggregate("$foo").output_field(ctx) is None
+        for cls in (Last, Min, Push, AddToSet, StdDevPop, StdDevSamp, MergeObjects):
+            assert cls("age").resolve(ctx)
+
+    def test_q_edge_cases(self):
+        with pytest.raises(TypeError):
+            Q(5)
+        assert Q(Raw({"a": 1})).resolve(Context(Scope.for_model(Account))) == {"a": 1}
+        e = Q(Raw({"a": 1}), F("age") > 1, Q(age=2)).resolve_expr(Context(Scope.for_model(Account)))
+        assert e["$and"][0] == {"a": 1}
+        assert Q().resolve_expr(Context(Scope.for_model(Account))) is True
+        assert (Q(a=1) & Q()).children == Q(a=1).children and (Q() | Q(a=1)).children == Q(a=1).children
+        assert Q(a=1).__and__(5) is NotImplemented
+        assert (F("age") > 1).__and__(Q(a=1)) is NotImplemented
+        assert (F("age") > 1).__or__(Q(a=1)) is NotImplemented
+        assert Q(age=1).__rand__(5) is NotImplemented and Q(age=1).__ror__(5) is NotImplemented
+        with pytest.raises(InvalidLookup):
+            Q(tags__all=["a"]).resolve_expr(Context(Scope.for_model(Account)))
+        assert Q(username__startswith=Value("$x")).resolve_expr(Context(Scope.for_model(Account))) is not None
+        e = Q(username="$x").resolve_expr(Context(Scope.for_model(Account)))
+        assert e == {"$eq": ["$username", {"$literal": "$x"}]}
+        with pytest.raises(InvalidArgument):
+            Q(comments__elem_match=5).resolve(Context(Scope.for_model(Post)))
+        with pytest.raises(InvalidArgument):
+            Q(tags__elem_match=Q(a=1)).resolve(Context(Scope.for_model(Account)))
+        out = Q(comments__elem_match=Raw({"text": "x"})).resolve(Context(Scope.for_model(Post)))
+        assert out == {"comments": {"$elemMatch": {"text": "x"}}}
+
+
+def DateToString_too_many():
+    return aggify.Trim("a", "b")
+
+
+class TestPipelineEdges:
+    def P(self, model=Account):
+        return Aggify(model)
+
+    def test_clone_and_protocol(self):
+        p = self.P().filter(age=1)
+        assert bool(p) and p.pipelines == p.stages
+        assert p._clone(model=Post).model is Post
+
+    def test_order_by_variants(self):
+        assert self.P().order_by(F("age")).to_list() == [{"$sort": {"age": 1}}]
+        assert self.P().order_by().to_list() == []
+        with pytest.raises(InvalidArgument):
+            self.P().order_by(F("age") + 1)
+        with pytest.raises(InvalidArgument):
+            self.P().order_by((F("age") + 1).desc())
+        with pytest.raises(InvalidArgument):
+            self.P().order_by(5)
+
+    def test_skip_limit_validation(self):
+        with pytest.raises(MongoIndexError):
+            self.P().skip(-1)
+        with pytest.raises(MongoIndexError):
+            self.P().limit(True)
+        with pytest.raises(InvalidArgument):
+            self.P().limit(0)
+        with pytest.raises(InvalidArgument):
+            self.P()[3:3]
+        assert self.P().skip(0).to_list() == []
+
+    def test_project_validation(self):
+        with pytest.raises(InvalidProjection):
+            self.P().project()
+        assert self.P().project(age=True).to_list() == [{"$project": {"age": 1}}]
+
+    def test_filter_with_only_joins(self):
+        p = self.P(Post).filter(Q(owner__username="a") | Q(owner__username="b"))
+        assert p.to_list()[-1]["$match"]["$or"]
+
+    def test_group_variants(self):
+        assert self.P().group(F("age") + 1).to_list()[0]["$group"]["_id"] == {"$add": ["$age", 1]}
+        with pytest.raises(InvalidArgument):
+            self.P().group(5)
+        with pytest.raises(InvalidArgument):
+            self.P().group("age", _id=Count())
+        assert self.P().group(None, n=Raw({"$sum": 1})).to_list()[0]["$group"]["n"] == {"$sum": 1}
+
+    def test_annotate_args(self):
+        with pytest.raises(AnnotationError):
+            self.P().annotate(5)
+        assert self.P().annotate().to_list() == []
+
+    def test_lookup_sub_pipeline_as_list_and_scan(self):
+        p = self.P().lookup(Post, "x", let={}, pipeline=[{"$limit": 1}])
+        assert p.to_list()[0]["$lookup"]["pipeline"] == [{"$limit": 1}]
+
+    def test_graph_lookup_options(self):
+        spec = (
+            self.P()
+            .graph_lookup(Account, F("age"), "age", "score", "c", depth_field="d", restrict=Q(age__gt=1))
+            .to_list()[0]["$graphLookup"]
+        )
+        assert spec["depthField"] == "d" and spec["restrictSearchWithMatch"] == {"age": {"$gt": 1}}
+
+    def test_bucket_auto(self):
+        spec = self.P().bucket_auto("age", 3, granularity="R5", n=Count()).to_list()[0]["$bucketAuto"]
+        assert spec == {"groupBy": "$age", "buckets": 3, "granularity": "R5", "output": {"n": {"$sum": 1}}}
+        assert "output" not in self.P().bucket_auto("age", 3).to_list()[0]["$bucketAuto"]
+        assert "output" not in self.P().bucket("age", [0, 1]).to_list()[0]["$bucket"]
+
+    def test_replace_root_from_list_of_embedded_and_expression(self):
+        class Team(Document):
+            members = ListField(EmbeddedField(Address))
+
+        assert self.P(Team).replace_root("members").to_list() == [{"$replaceRoot": {"newRoot": "$members"}}]
+        assert self.P().replace_root(F("address")).to_list() == [{"$replaceRoot": {"newRoot": "$address"}}]
+        with pytest.raises(InvalidArgument):
+            self.P().replace_root()
+        with pytest.raises(InvalidArgument):
+            self.P().redact(Q(age=1), value1="a")
+
+    def test_geo_near_options_and_stage_names(self):
+        spec = (
+            self.P()
+            .geo_near((1.0, 2.0), "d", max_distance=5, key="address__geo", query=Q(age__gt=1))
+            .to_list()[0]["$geoNear"]
+        )
+        assert spec["maxDistance"] == 5 and spec["key"] == "address.geo" and spec["query"] == {"age": {"$gt": 1}}
+        assert self.P().stage("sample", {"size": 1}).to_list() == [{"$sample": {"size": 1}}]
+
+    def test_default_max_time_applies(self):
+        aggify.configure(default_max_time_ms=123)
+        assert self.P()._aggregate_kwargs() == {"maxTimeMS": 123}
+
+    def test_convert_row_and_helpers(self):
+        from aggify.pipeline import _convert, convert_row
+
+        scope = Scope.for_model(Account)
+        row = convert_row(
+            scope,
+            {
+                "_id": ObjectId(),
+                "username": "a",
+                "extra": 1,
+                "_j_x": {},
+                "tags": ["a"],
+                "address": {"city": "t", "zip": "9"},
+                "role": "admin",
+            },
+        )
+        assert row["username"] == "a" and row["extra"] == 1 and "_j_x" not in row
+        assert row["address"] == {"city": "t", "zip_code": "9"} and row["role"] is Role.ADMIN
+        assert _convert(IntField(), None) is None
+        assert _convert(DecimalField(), object()) is not None  # conversion failure keeps the raw value
+
+
+class TestDocumentEdges:
+    def test_options_api(self):
+        meta = Account._meta
+        assert meta.get_field("pk") is meta.pk and len(meta.concrete_fields) == len(meta.fields)
+        assert "Account" in repr(meta) and meta.db_fields["_id"] is meta.pk
+        with pytest.raises(InvalidField):
+            meta.get_field("nope")
+
+    def test_pk_declarations(self):
+        class WithId(Document):
+            id = StringField()
+
+        assert WithId._meta.pk.db_field == "_id"
+        with pytest.raises(aggify.ConfigurationError):
+
+            class TwoPks(Document):
+                a = StringField(primary_key=True)
+                b = StringField(primary_key=True)
+
+        class ObjPk(Document):
+            id = ObjectIdField(primary_key=True)
+
+        assert ObjPk._meta.pk.has_default()
+
+    def test_pk_kwarg_and_manager_inheritance(self):
+        pk = ObjectId()
+        assert Account(pk=pk, username="ab").pk == pk
+
+        class Base(Document):
+            objects = aggify.Manager()
+
+            class Meta:
+                abstract = True
+
+        class Kid(Base):
+            pass
+
+        assert Kid.objects.model is Kid
+
+    def test_repr_hash_and_dirty(self):
+        a = Account(username="ab")
+        assert str(a.pk) in repr(a) and a.is_dirty
+        with pytest.raises(TypeError):
+            hash(Account(id=None, username="ab"))
+        e = Address(city="x")
+        assert repr(e).startswith("<Address") and e.pk is None
+
+    def test_json_primitives(self):
+        from aggify.documents import _to_primitive
+
+        assert _to_primitive(dt.datetime(2024, 1, 2, tzinfo=dt.timezone.utc)) == "2024-01-02T00:00:00Z"
+        assert _to_primitive(dt.date(2024, 1, 2)) == "2024-01-02"
+        assert _to_primitive(Decimal128("1.5")) == "1.5" and _to_primitive(decimal.Decimal("2")) == "2"
+        u = uuid.uuid4()
+        assert _to_primitive(u) == str(u) and _to_primitive(Role.ADMIN) == "admin"
+        assert _to_primitive(b"ab") == "YWI=" and _to_primitive({"a": (1, 2)}) == {"a": [1, 2]}
+
+    def test_full_clean_collects_message_only_errors(self):
+        class Odd(Document):
+            x = IntField()
+
+        o = Odd(x=1)
+        o._meta.fields["x"].validators.append(lambda v: (_ for _ in ()).throw(ValidationError("plain")))
+        try:
+            with pytest.raises(ValidationError) as ei:
+                o.full_clean()
+            assert "x" in ei.value.errors
+        finally:
+            o._meta.fields["x"].validators.clear()
+
+    def test_pk_required_operations(self):
+        a = Account(id=None, username="ab")
+        for op in (a.delete, a.refresh_from_db):
+            with pytest.raises(ValueError):
+                op()
+
+    def test_manager_details(self):
+        mgr = Account.objects
+        with pytest.raises(AttributeError):
+            mgr.__nonexistent__
+        import copy
+
+        assert copy.copy(mgr).model is Account
+        with pytest.raises(AttributeError):
+            mgr.model_
+            raise AttributeError
+        qs_manager = aggify.Manager(aggify.QuerySet)
+        assert qs_manager.queryset_class is aggify.QuerySet
+
+    def test_signal_helper_receiver_with_list(self):
+        calls = []
+        fn = aggify.receiver([aggify.pre_save, aggify.post_save])(lambda sender, **kw: calls.append(1))
+        try:
+            aggify.pre_save.send(Account, instance=None, created=True)
+            assert calls == [1]
+        finally:
+            aggify.pre_save.disconnect(fn)
+            aggify.post_save.disconnect(fn)
+
+
+class TestMonitoringUnit:
+    def test_log_helpers(self):
+        from aggify.monitoring import CommandLog, CommandRecord
+
+        log = CommandLog(
+            [CommandRecord("find", "db", {"find": "c"}, 1), CommandRecord("insert", "db", {"insert": 5}, 2)]
+        )
+        assert len(log) == 2 and [r.name for r in log] == ["find", "insert"] and log.count == 2
+        assert log.of("find")[0].collection == "c" and log.records[1].collection is None
+
+    def test_listener_failed_and_succeeded_branches(self):
+        from types import SimpleNamespace as NS
+
+        from aggify.monitoring import LISTENER, capture_commands
+
+        with capture_commands() as log:
+            LISTENER.started(NS(command_name="find", database_name="d", command={"find": "c"}, request_id=7))
+            LISTENER.failed(NS(request_id=7))
+            LISTENER.succeeded(NS(request_id=7, duration_micros=5))
+            LISTENER.started(NS(command_name="ping", database_name="d", command={}, request_id=8))
+        assert log.records[0].failed and log.records[0].duration_micros == 5 and len(log) == 1
+        LISTENER.failed(NS(request_id=1))  # no active recorder: no-op
+        LISTENER.succeeded(NS(request_id=1, duration_micros=1))
diff --git a/tests/test_pipeline.py b/tests/test_pipeline.py
index 990f717..dfb26a6 100644
--- a/tests/test_pipeline.py
+++ b/tests/test_pipeline.py
@@ -395,6 +395,6 @@ def test_nested_pipeline_is_scanned(self):
 
     def test_options(self):
         p = P().options(allow_disk_use=True, max_time_ms=1000)
-        assert p._aggregate_kwargs() == {"allow_disk_use": True, "max_time_ms": 1000}
+        assert p._aggregate_kwargs() == {"allowDiskUse": True, "maxTimeMS": 1000}
         with pytest.raises(InvalidArgument):
             P().options(bogus=1)
diff --git a/tests/test_query_compile.py b/tests/test_query_compile.py
index 24a709f..a3d6175 100644
--- a/tests/test_query_compile.py
+++ b/tests/test_query_compile.py
@@ -194,6 +194,4 @@ def test_unknown_embedded_field(self):
 
     def test_dollar_key_in_dict_path_is_refused(self):
         with pytest.raises(UnsafeQueryError):
-            q(Account, **{"attrs____set": 1}) if False else (_ for _ in ()).throw(UnsafeQueryError())
-        with pytest.raises(Exception):
             q(Account, **{"attrs__$ne": 1})