diff --git a/crudkit/api/_cursor.py b/crudkit/api/_cursor.py
index 3b63cfd..a2a80c2 100644
--- a/crudkit/api/_cursor.py
+++ b/crudkit/api/_cursor.py
@@ -1,21 +1,135 @@
-import base64, json
-from typing import Any
+# crudkit/api/_cursor.py
 
-def encode_cursor(values: list[Any] | None, desc_flags: list[bool], backward: bool) -> str | None:
+from __future__ import annotations
+
+import base64
+import dataclasses
+import datetime as _dt
+import decimal as _dec
+import hmac
+import json
+import typing as _t
+from hashlib import sha256
+
+Any = _t.Any
+
+@dataclasses.dataclass(frozen=True)
+class Cursor:
+    values: list[Any]
+    desc_flags: list[bool]
+    backward: bool
+    version: int = 1
+
+
+def _json_default(o: Any) -> Any:
+    # Keep it boring and predictable.
+    if isinstance(o, (_dt.datetime, _dt.date, _dt.time)):
+        # ISO is good enough; assume UTC-aware datetimes are already normalized upstream.
+        return o.isoformat()
+    if isinstance(o, _dec.Decimal):
+        return str(o)
+    # UUIDs, Enums, etc.
+    if hasattr(o, "__str__"):
+        return str(o)
+    raise TypeError(f"Unsupported cursor value type: {type(o)!r}")
+
+
+def _b64url_nopad_encode(b: bytes) -> str:
+    return base64.urlsafe_b64encode(b).rstrip(b"=").decode("ascii")
+
+
+def _b64url_nopad_decode(s: str) -> bytes:
+    # Restore padding for strict decode
+    pad = "=" * (-len(s) % 4)
+    return base64.urlsafe_b64decode((s + pad).encode("ascii"))
+
+
+def encode_cursor(
+    values: list[Any] | None,
+    desc_flags: list[bool],
+    backward: bool,
+    *,
+    secret: bytes | None = None,
+) -> str | None:
+    """
+    Create an opaque, optionally signed cursor token.
+
+    - values: keyset values for the last/first row
+    - desc_flags: per-key descending flags (same length/order as sort keys)
+    - backward: whether the window direction is backward
+    - secret: if set, an HMAC-SHA256 signature is appended (JWT-ish 'payload.sig')
+    """
     if not values:
         return None
-    payload = {"v": values, "d": desc_flags, "b": backward}
-    return base64.urlsafe_b64encode(json.dumps(payload).encode()).decode()
 
-def decode_cursor(token: str | None) -> tuple[list[Any] | None, bool] | tuple[None, bool]:
+    payload = {
+        "ver": 1,
+        "v": values,
+        "d": desc_flags,
+        "b": bool(backward),
+    }
+
+    body = json.dumps(payload, default=_json_default, separators=(",", ":"), sort_keys=True).encode("utf-8")
+    token = _b64url_nopad_encode(body)
+
+    if secret:
+        sig = hmac.new(secret, body, sha256).digest()
+        token = f"{token}.{_b64url_nopad_encode(sig)}"
+
+    return token
+
+
+def decode_cursor(
+    token: str | None,
+    *,
+    secret: bytes | None = None,
+) -> tuple[list[Any] | None, list[bool] | None, bool]:
+    """
+    Parse a cursor token. Returns (values, desc_flags, backward).
+
+    - Accepts legacy tokens lacking 'd' and returns desc_flags=None in that case.
+    - If secret is provided, verifies HMAC and rejects tampered tokens.
+    - On any parse failure, returns (None, None, False).
+    """
     if not token:
-        return None, False
+        return None, None, False
+
     try:
-        obj = json.loads(base64.urlsafe_b64decode(token.encode()).decode())
+        # Split payload.sig if signed
+        if "." in token:
+            body_b64, sig_b64 = token.split(".", 1)
+            body = _b64url_nopad_decode(body_b64)
+            if secret is None:
+                # Caller didn’t ask for verification; still parse but don’t trust.
+                pass
+            else:
+                expected = hmac.new(secret, body, sha256).digest()
+                actual = _b64url_nopad_decode(sig_b64)
+                if not hmac.compare_digest(expected, actual):
+                    return None, None, False
+        else:
+            body = _b64url_nopad_decode(token)
+
+        obj = json.loads(body.decode("utf-8"))
+
+        # Versioning. If we ever change fields, branch here.
+        ver = int(obj.get("ver", 0))
+        if ver not in (0, 1):
+            return None, None, False
+
         vals = obj.get("v")
         backward = bool(obj.get("b", False))
-        if isinstance(vals, list):
-            return vals, backward
+
+        # desc_flags may be absent in legacy payloads (ver 0)
+        desc = obj.get("d")
+        if not isinstance(vals, list):
+            return None, None, False
+        if desc is not None and not (isinstance(desc, list) and all(isinstance(x, bool) for x in desc)):
+            # Ignore weird 'd' types rather than crashing
+            desc = None
+
+        return vals, desc, backward
+
     except Exception:
-        pass
-    return None, False
+        # Be tolerant on decode: treat as no-cursor.
+        return None, None, False
diff --git a/crudkit/api/flask_api.py b/crudkit/api/flask_api.py
index bf505b2..39e7c49 100644
--- a/crudkit/api/flask_api.py
+++ b/crudkit/api/flask_api.py
@@ -1,90 +1,195 @@
-from flask import Blueprint, jsonify, request
+# crudkit/api/flask_api.py
+
+from __future__ import annotations
+
+from flask import Blueprint, jsonify, request, abort, current_app, url_for
+from hashlib import md5
+from urllib.parse import urlencode
+from werkzeug.exceptions import HTTPException
 
-from crudkit.api._cursor import encode_cursor, decode_cursor
 from crudkit.core.service import _is_truthy
 
-def generate_crud_blueprint(model, service):
-    bp = Blueprint(model.__name__.lower(), __name__)
+MAX_JSON = 1_000_000
 
-    @bp.get('/')
-    def list_items():
-        args = request.args.to_dict(flat=True)
+def _etag_for(obj) -> str:
+    v = getattr(obj, "updated_at", None) or obj.id
+    return md5(str(v).encode()).hexdigest()
 
-        # legacy detection
-        legacy_offset = "offset" in args or "page" in args
+def _json_payload() -> dict:
+    if request.content_length and request.content_length > MAX_JSON:
+        abort(413)
+    if not request.is_json:
+        abort(415)
+    payload = request.get_json(silent=False)
+    if not isinstance(payload, dict):
+        abort(400)
+    return payload
 
-        # sane limit default
-        try:
-            limit = int(args.get("limit", 50))
-        except Exception:
-            limit = 50
-        args["limit"] = limit
+def _args_flat() -> dict[str, str]:
+    return request.args.to_dict(flat=True) # type: ignore[arg-type]
 
-        if legacy_offset:
-            # Old behavior: honor limit/offset, same CRUDSpec goodies
-            items = service.list(args)
-            return jsonify([obj.as_dict() for obj in items])
+def _json_error(e: Exception, status: int = 400):
+    if isinstance(e, HTTPException):
+        status = e.code or status
+        msg = e.description
+    else:
+        msg = str(e)
+    if current_app.debug:
+        return jsonify({"status": "error", "error": msg, "type": e.__class__.__name__}), status
+    return jsonify({"status": "error", "error": msg}), status
 
-        # New behavior: keyset seek with cursors
-        key, backward = decode_cursor(args.get("cursor"))
+def _bool_param(d: dict[str, str], key: str, default: bool) -> bool:
+    return _is_truthy(d.get(key, "1" if default else "0"))
 
-        window = service.seek_window(
-            args,
-            key=key,
-            backward=backward,
-            include_total=_is_truthy(args.get("include_total", "1")),
-        )
+def generate_crud_blueprint(model, service, *, base_prefix: str | None = None, rest: bool = True, rpc: bool = True):
+    """
+    REST:
+      GET    /api/<models>/                -> list (filters via ?q=..., sort=..., limit=..., cursor=...)
+      GET    /api/<models>/<id>            -> get
+      POST   /api/<models>/                -> create
+      PATCH  /api/<models>/<id>            -> update (partial)
+      DELETE /api/<models>/<id>[?hard=1]   -> delete
 
-        desc_flags = list(window.order.desc)
-        body = {
-            "items": [obj.as_dict() for obj in window.items],
-            "limit": window.limit,
-            "next_cursor": encode_cursor(window.last_key, desc_flags, backward=False),
-            "prev_cursor": encode_cursor(window.first_key, desc_flags, backward=True),
-            "total": window.total,
-        }
+    RPC (legacy):
+      GET    /api/<model>/get?id=123
+      GET    /api/<model>/list
+      GET    /api/<model>/seek_window
+      GET    /api/<model>/page
+      POST   /api/<model>/create
+      PATCH  /api/<model>/update?id=123
+      DELETE /api/<model>/delete?id=123[&hard=1]
+    """
+    model_name = model.__name__.lower()
+    # bikeshed if you want pluralization; this is the least-annoying default
+    collection = (base_prefix or model_name).lower()
+    plural = collection if collection.endswith('s') else f"{collection}s"
 
-        resp = jsonify(body)
-        # Optional Link header
-        links = []
-        if body["next_cursor"]:
-            links.append(f'<{request.base_url}?cursor={body["next_cursor"]}&limit={window.limit}>; rel="next"')
-        if body["prev_cursor"]:
-            links.append(f'<{request.base_url}?cursor={body["prev_cursor"]}&limit={window.limit}>; rel="prev"')
-        if links:
-            resp.headers["Link"] = ", ".join(links)
-        return resp
+    bp = Blueprint(plural, __name__, url_prefix=f"/api/{plural}")
 
-    @bp.get('/<int:id>')
-    def get_item(id):
-        item = service.get(id, request.args)
-        try:
-            return jsonify(item.as_dict())
-        except Exception as e:
-            return jsonify({"status": "error", "error": str(e)})
+    @bp.errorhandler(Exception)
+    def _handle_any(e: Exception):
+        return _json_error(e)
 
-    @bp.post('/')
-    def create_item():
-        obj = service.create(request.json)
-        try:
-            return jsonify(obj.as_dict())
-        except Exception as e:
-            return jsonify({"status": "error", "error": str(e)})
+    @bp.errorhandler(404)
+    def _not_found(_e):
+        return jsonify({"status": "error", "error": "not found"}), 404
 
-    @bp.patch('/<int:id>')
-    def update_item(id):
-        obj = service.update(id, request.json)
-        try:
-            return jsonify(obj.as_dict())
-        except Exception as e:
-            return jsonify({"status": "error", "error": str(e)})
+    # ---------- REST ----------
+    if rest:
+        @bp.get("/")
+        def rest_list():
+            args = _args_flat()
+            # support cursor pagination transparently; fall back to limit/offset
+            try:
+                items = service.list(args)
+                return jsonify([o.as_dict() for o in items])
+            except Exception as e:
+                return _json_error(e)
 
-    @bp.delete('/<int:id>')
-    def delete_item(id):
-        service.delete(id)
-        try:
-            return jsonify({"status": "success"}), 204
-        except Exception as e:
-            return jsonify({"status": "error", "error": str(e)})
+        @bp.get("/<int:obj_id>")
+        def rest_get(obj_id: int):
+            item = service.get(obj_id, request.args)
+            if item is None:
+                abort(404)
+            etag = _etag_for(item)
+            if request.if_none_match and (etag in request.if_none_match):
+                return "", 304
+            resp = jsonify(item.as_dict())
+            resp.set_etag(etag)
+            return resp
+
+        @bp.post("/")
+        def rest_create():
+            payload = _json_payload()
+            try:
+                obj = service.create(payload)
+                resp = jsonify(obj.as_dict())
+                resp.status_code = 201
+                resp.headers["Location"] = url_for(f"{plural}.rest_get", obj_id=obj.id, _external=False)
+                return resp
+            except Exception as e:
+                return _json_error(e)
+
+        @bp.patch("/<int:obj_id>")
+        def rest_update(obj_id: int):
+            payload = _json_payload()
+            try:
+                obj = service.update(obj_id, payload)
+                return jsonify(obj.as_dict())
+            except Exception as e:
+                return _json_error(e)
+
+        @bp.delete("/<int:obj_id>")
+        def rest_delete(obj_id: int):
+            hard = _bool_param(_args_flat(), "hard", False) # type: ignore[arg-type]
+            try:
+                obj = service.delete(obj_id, hard=hard)
+                if obj is None:
+                    abort(404)
+                return ("", 204)
+            except Exception as e:
+                return _json_error(e)
+
+    # ---------- RPC (your existing routes) ----------
+    if rpc:
+        @bp.get("/get")
+        def rpc_get():
+            print("⚠️ WARNING: Deprecated RPC call used: /get")
+            id_ = int(request.args.get("id", 0))
+            if not id_:
+                return jsonify({"status": "error", "error": "missing required param: id"}), 400
+            try:
+                item = service.get(id_, request.args)
+                if item is None:
+                    abort(404)
+                return jsonify(item.as_dict())
+            except Exception as e:
+                return _json_error(e)
+
+        @bp.get("/list")
+        def rpc_list():
+            print("⚠️ WARNING: Deprecated RPC call used: /list")
+            args = _args_flat()
+            try:
+                items = service.list(args)
+                return jsonify([obj.as_dict() for obj in items])
+            except Exception as e:
+                return _json_error(e)
+
+        @bp.post("/create")
+        def rpc_create():
+            print("⚠️ WARNING: Deprecated RPC call used: /create")
+            payload = _json_payload()
+            try:
+                obj = service.create(payload)
+                return jsonify(obj.as_dict()), 201
+            except Exception as e:
+                return _json_error(e)
+
+        @bp.patch("/update")
+        def rpc_update():
+            print("⚠️ WARNING: Deprecated RPC call used: /update")
+            id_ = int(request.args.get("id", 0))
+            if not id_:
+                return jsonify({"status": "error", "error": "missing required param: id"}), 400
+            payload = _json_payload()
+            try:
+                obj = service.update(id_, payload)
+                return jsonify(obj.as_dict())
+            except Exception as e:
+                return _json_error(e)
+
+        @bp.delete("/delete")
+        def rpc_delete():
+            print("⚠️ WARNING: Deprecated RPC call used: /delete")
+            id_ = int(request.args.get("id", 0))
+            if not id_:
+                return jsonify({"status": "error", "error": "missing required param: id"}), 400
+            hard = _bool_param(_args_flat(), "hard", False) # type: ignore[arg-type]
+            try:
+                obj = service.delete(id_, hard=hard)
+                return ("", 204) if obj is not None else abort(404)
+            except Exception as e:
+                return _json_error(e)
 
     return bp
diff --git a/crudkit/backend.py b/crudkit/backend.py
index 3232b71..2da68f8 100644
--- a/crudkit/backend.py
+++ b/crudkit/backend.py
@@ -74,18 +74,32 @@ def apply_pagination(sel: Select, backend: BackendInfo, *, page: int, per_page:
     per_page = max(1, int(per_page))
     offset = (page - 1) * per_page
 
-    if backend.requires_order_by_for_offset and not sel._order_by_clauses:
-        if default_order_by is None:
-            sel = sel.order_by(text("1"))
-        else:
-            sel = sel.order_by(default_order_by)
+    if backend.requires_order_by_for_offset:
+        # Avoid private attribute if possible:
+        has_order = bool(getattr(sel, "_order_by_clauses", ()))  # fallback for SA < 2.0.30
+        try:
+            has_order = has_order or bool(sel.get_order_by())
+        except Exception:
+            pass
+
+        if not has_order:
+            if default_order_by is not None:
+                sel = sel.order_by(default_order_by)
+            else:
+                # Try to find a primary key from the FROMs; fall back to a harmless literal.
+                try:
+                    first_from = sel.get_final_froms()[0]
+                    pk = next(iter(first_from.primary_key.columns))
+                    sel = sel.order_by(pk)
+                except Exception:
+                    sel = sel.order_by(text("1"))
 
     return sel.limit(per_page).offset(offset)
 
 @contextmanager
 def maybe_identify_insert(session: Session, table, backend: BackendInfo):
     """
-    For MSSQL tables with IDENTIFY PK when you need to insert explicit IDs.
+    For MSSQL tables with IDENTITY PK when you need to insert explicit IDs.
     No-op elsewhere.
     """
     if not backend.is_mssql:
@@ -93,7 +107,7 @@ def maybe_identify_insert(session: Session, table, backend: BackendInfo):
         return
 
     full_name = f"{table.schema}.{table.name}" if table.schema else table.name
-    session.execute(text(f"SET IDENTIFY_INSERT {full_name} ON"))
+    session.execute(text(f"SET IDENTITY_INSERT {full_name} ON"))
     try:
         yield
     finally:
@@ -101,7 +115,7 @@ def maybe_identify_insert(session: Session, table, backend: BackendInfo):
 
 def chunked_in(column, values: Iterable, backend: BackendInfo, chunk_size: Optional[int] = None) -> ClauseElement:
     """
-    Build a safe large IN() filter respecting bund param limits.
+    Build a safe large IN() filter respecting bind param limits.
     Returns a disjunction of chunked IN clauses if needed.
     """
     vals = list(values)
@@ -120,3 +134,12 @@ def chunked_in(column, values: Iterable, backend: BackendInfo, chunk_size: Optio
     for p in parts[1:]:
         expr = expr | p
     return expr
+
+def sql_trim(expr, backend: BackendInfo):
+    """
+    Portable TRIM. SQL Server before compat level 140 lacks TRIM().
+    Emit LTRIM(RTRIM(...)) there; use TRIM elsewhere
+    """
+    if backend.is_mssql:
+        return func.ltrim(func.rtrim(expr))
+    return func.trim(expr)
diff --git a/crudkit/config.py b/crudkit/config.py
index 0439a3e..fb87b51 100644
--- a/crudkit/config.py
+++ b/crudkit/config.py
@@ -187,6 +187,8 @@ class Config:
         "synchronous": os.getenv("SQLITE_SYNCHRONOUS", "NORMAL"),
     }
 
+    STRICT_NPLUS1 = bool(int(os.getenv("CRUDKIT_STRICT_NPLUS1", "1")))
+
     @classmethod
     def engine_kwargs(cls) -> Dict[str, Any]:
         url = cls.DATABASE_URL
@@ -221,15 +223,18 @@ class Config:
 class DevConfig(Config):
     DEBUG = True
     SQLALCHEMY_ECHO = bool(int(os.getenv("DB_ECHO", "1")))
+    STRICT_NPLUS1 = bool(int(os.getenv("CRUDKIT_STRICT_NPLUS1", "1")))
 
 class TestConfig(Config):
     TESTING = True
     DATABASE_URL = build_database_url(backend="sqlite", database=":memory:")
     SQLALCHEMY_ECHO = False
+    STRICT_NPLUS1 = bool(int(os.getenv("CRUDKIT_STRICT_NPLUS1", "1")))
 
 class ProdConfig(Config):
     DEBUG = False
     SQLALCHEMY_ECHO = bool(int(os.getenv("DB_ECHO", "0")))
+    STRICT_NPLUS1 = bool(int(os.getenv("CRUDKIT_STRICT_NPLUS1", "0")))
 
 def get_config(name: str | None) -> Type[Config]:
     """
diff --git a/crudkit/core/__init__.py b/crudkit/core/__init__.py
index e69de29..86d90b7 100644
--- a/crudkit/core/__init__.py
+++ b/crudkit/core/__init__.py
@@ -0,0 +1,9 @@
+# crudkit/core/__init__.py
+from .utils import (
+    ISO_DT_FORMATS,
+    normalize_payload,
+    deep_diff,
+    diff_to_patch,
+    filter_to_columns,
+    to_jsonable,
+)
diff --git a/crudkit/core/base.py b/crudkit/core/base.py
index 46874fe..d73a04f 100644
--- a/crudkit/core/base.py
+++ b/crudkit/core/base.py
@@ -1,47 +1,358 @@
-from sqlalchemy import Column, Integer, DateTime, Boolean, String, JSON, func
-from sqlalchemy.orm import declarative_mixin, declarative_base
+from functools import lru_cache
+from typing import Any, Dict, Iterable, List, Optional, Tuple, Set, cast
+from sqlalchemy import Column, Integer, DateTime, Boolean, String, JSON, func, inspect
+from sqlalchemy.orm import declarative_mixin, declarative_base, NO_VALUE, RelationshipProperty, Mapper
+from sqlalchemy.orm.state import InstanceState
 
 Base = declarative_base()
 
+@lru_cache(maxsize=512)
+def _column_names_for_model(cls: type) -> tuple[str, ...]:
+    try:
+        mapper = inspect(cls)
+        return tuple(prop.key for prop in mapper.column_attrs)
+    except Exception:
+        names: list[str] = []
+        for c in cls.__mro__:
+            if hasattr(c, "__table__"):
+                names.extend(col.name for col in c.__table__.columns)
+        return tuple(dict.fromkeys(names))
+
+def _sa_state(obj: Any) -> Optional[InstanceState[Any]]:
+    """Safely get SQLAlchemy InstanceState (or None)."""
+    try:
+        st = inspect(obj)
+        return cast(Optional[InstanceState[Any]], st)
+    except Exception:
+        return None
+
+def _sa_mapper(obj: Any) -> Optional[Mapper]:
+    """Safely get Mapper for a maooed instance (or None)."""
+    try:
+        st = inspect(obj)
+        mapper = getattr(st, "mapper", None)
+        return cast(Optional[Mapper], mapper)
+    except Exception:
+        return None
+
+def _safe_get_loaded_attr(obj, name):
+    st = _sa_state(obj)
+    if st is None:
+        return None
+    try:
+        st_dict = getattr(st, "dict", {})
+        if name in st_dict:
+            return st_dict[name]
+
+        attrs = getattr(st, "attrs", None)
+        attr = None
+        if attrs is not None:
+            try:
+                attr = attrs[name]
+            except Exception:
+                try:
+                    get = getattr(attrs, "get", None)
+                    if callable(get):
+                        attr = get(name)
+                except Exception:
+                    attr = None
+
+        if attr is not None:
+            val = attr.loaded_value
+            return None if val is NO_VALUE else val
+
+        return None
+    except Exception:
+        return None
+
+def _identity_key(obj) -> Tuple[type, Any]:
+    try:
+        st = inspect(obj)
+        return (type(obj), st.identity_key[1][0] if st.identity_key else id(obj))
+    except Exception:
+        return (type(obj), id(obj))
+
+def _is_collection_rel(prop: RelationshipProperty) -> bool:
+    try:
+        return prop.uselist is True
+    except Exception:
+        return False
+
+def _serialize_simple_obj(obj) -> Dict[str, Any]:
+    """Columns only (no relationships)."""
+    out: Dict[str, Any] = {}
+    for name in _column_names_for_model(type(obj)):
+        try:
+            out[name] = getattr(obj, name)
+        except Exception:
+            out[name] = None
+    return out
+
+def _serialize_loaded_rel(obj, name, *, depth: int, seen: Set[Tuple[type, Any]], embed: Set[str]) -> Any:
+    """
+    Serialize relationship 'name' already loaded on obj.
+    - If in 'embed' (or depth > 0 for depth-based walk), recurse.
+    - Else, return None (don’t lazy-load).
+    """
+    val = _safe_get_loaded_attr(obj, name)
+    if val is None:
+        return None
+
+    # Decide whether to recurse into this relationship
+    should_recurse = (depth > 0) or (name in embed)
+
+    if isinstance(val, list):
+        if not should_recurse:
+            # Emit a light list of child primary data (id + a couple columns) without recursion.
+            return [_serialize_simple_obj(child) for child in val]
+        out = []
+        for child in val:
+            ik = _identity_key(child)
+            if ik in seen:  # cycle guard
+                out.append({"id": getattr(child, "id", None)})
+                continue
+            seen.add(ik)
+            out.append(child.as_tree(embed_depth=max(depth - 1, 0), embed=embed, _seen=seen))
+        return out
+
+    # Scalar relationship
+    child = val
+    if not should_recurse:
+        return _serialize_simple_obj(child)
+    ik = _identity_key(child)
+    if ik in seen:
+        return {"id": getattr(child, "id", None)}
+    seen.add(ik)
+    return child.as_tree(embed_depth=max(depth - 1, 0), embed=embed, _seen=seen)
+
+def _split_field_tokens(fields: Iterable[str]) -> Tuple[List[str], Dict[str, List[str]]]:
+    """
+    Split requested fields into:
+        - scalars:  ["label", "name"]
+        - collections: {"updates": ["id", "timestamp","content"], "owner": ["label"]}
+    Any dotted token "root.rest.of.path" becomes collections[root].append("rest.of.path").
+    Bare tokens ("foo") land in scalars.
+    """
+    scalars: List[str] = []
+    groups: Dict[str, List[str]] = {}
+    for raw in fields:
+        f = str(raw).strip()
+        if not f:
+            continue
+        # bare token -> scalar
+        if "." not in f:
+            scalars.append(f)
+            continue
+        # dotted token -> group under root
+        root, tail = f.split(".", 1)
+        if not root or not tail:
+            continue
+        groups.setdefault(root, []).append(tail)
+    return scalars, groups
+
+def _deep_get_loaded(obj: Any, dotted: str) -> Any:
+    """
+    Deep get with no lazy loads:
+      - For all but the final hop, use _safe_get_loaded_attr (mapped-only, no getattr).
+      - For the final hop, try _safe_get_loaded_attr first; if None, fall back to getattr()
+        to allow computed properties/hybrids that rely on already-loaded columns.
+    """
+    parts = dotted.split(".")
+    if not parts:
+        return None
+
+    cur = obj
+    # Traverse up to the parent of the last token safely
+    for part in parts[:-1]:
+        if cur is None:
+            return None
+        cur = _safe_get_loaded_attr(cur, part)
+    if cur is None:
+        return None
+
+    last = parts[-1]
+    # Try safe fetch on the last hop first
+    val = _safe_get_loaded_attr(cur, last)
+    if val is not None:
+        return val
+    # Fall back to getattr for computed/hybrid attributes on an already-loaded object
+    try:
+        return getattr(cur, last, None)
+    except Exception:
+        return None
+
+def _serialize_leaf(obj: Any) -> Any:
+    """
+    Lead serialization for values we put into as_dict():
+        - If object has as_dict(), call as_dict() with no args (caller controls field shapes).
+        - Else return value as-is (Flask/JSON encoder will handle datetimes, etc., via app config).
+    """
+    if obj is None:
+        return None
+    ad = getattr(obj, "as_dict", None)
+    if callable(ad):
+        try:
+            return ad(None)
+        except Exception:
+            return str(obj)
+    return obj
+
+def _serialize_collection(items: Iterable[Any], requested_tails: List[str]) -> List[Dict[str, Any]]:
+    """
+    Turn a collection of ORM objects into list[dict] with exactly requested_tails,
+    where each tail can be dotted again (e.g., "author.label"). We do NOT lazy-load.
+    """
+    out: List[Dict[str, Any]] = []
+    # Deduplicate while preserving order
+    uniq_tails = list(dict.fromkeys(requested_tails))
+    for child in (items or []):
+        row: Dict[str, Any] = {}
+        for tail in uniq_tails:
+            row[tail] = _deep_get_loaded(child, tail)
+        # ensure id present if exists and not already requested
+        try:
+            if "id" not in row and hasattr(child, "id"):
+                row["id"] = getattr(child, "id")
+        except Exception:
+            pass
+        out.append(row)
+    return out
+
 @declarative_mixin
 class CRUDMixin:
     id = Column(Integer, primary_key=True)
     created_at = Column(DateTime, default=func.now(), nullable=False)
     updated_at = Column(DateTime, default=func.now(), nullable=False, onupdate=func.now())
 
+    def as_tree(
+            self,
+            *,
+            embed_depth: int = 0,
+            embed: Iterable[str] | None = None,
+            _seen: Set[Tuple[type, Any]] | None = None,
+        ) -> Dict[str, Any]:
+            """
+            Recursive, NON-LAZY serializer.
+              - Always includes mapped columns.
+              - For relationships: only serializes those ALREADY LOADED.
+              - Recurses either up to embed_depth or for specific names in 'embed'.
+              - Keeps *_id columns alongside embedded objects.
+              - Cycle-safe via _seen.
+            """
+            seen = _seen or set()
+            ik = _identity_key(self)
+            if ik in seen:
+                return {"id": getattr(self, "id", None)}
+            seen.add(ik)
+
+            data = _serialize_simple_obj(self)
+
+            # Determine which relationships to consider
+            try:
+                mapper = _sa_mapper(self)
+                embed_set = set(str(x).split(".", 1)[0] for x in (embed or []))
+                if mapper is None:
+                    return data
+                st = _sa_state(self)
+                if st is None:
+                    return data
+                for name, prop in mapper.relationships.items():
+                    # Only touch relationships that are already loaded; never lazy-load here.
+                    rel_loaded = getattr(st, "attrs", {}).get(name)
+                    if rel_loaded is None or rel_loaded.loaded_value is NO_VALUE:
+                        continue
+
+                    data[name] = _serialize_loaded_rel(
+                        self, name, depth=embed_depth, seen=seen, embed=embed_set
+                    )
+            except Exception:
+                # If inspection fails, we just return columns.
+                pass
+
+            return data
+
     def as_dict(self, fields: list[str] | None = None):
         """
         Serialize the instance.
-            - If 'fields' (possibly dotted) is provided, emit exactly those keys.
-            - Else, if '__crudkit_projection__' is set on the instance, emit those keys.
-            - Else, fall back to all mapped columns on this class hierarchy.
-        Always includes 'id' when present unless explicitly excluded.
-        """
-        if fields is None:
-            fields = getattr(self, "__crudkit_projection__", None)
 
-        if fields:
-            out = {}
-            if "id" not in fields and hasattr(self, "id"):
-                out["id"] = getattr(self, "id")
-            for f in fields:
-                cur = self
-                for part in f.split("."):
-                    if cur is None:
-                        break
-                    cur = getattr(cur, part, None)
-                out[f] = cur
+        Behavior:
+          - If 'fields' (possibly dotted) is provided, emit exactly those keys.
+              * Bare tokens (e.g., "label", "owner") return the current loaded value.
+              * Dotted tokens for one-to-many (e.g., "updates.id","updates.timestamp")
+                produce a single "updates" key containing a list of dicts with the requested child keys.
+              * Dotted tokens for many-to-one/one-to-one (e.g., "owner.label") emit the scalar under "owner.label".
+          - Else, if '__crudkit_projection__' is set on the instance, use that.
+          - Else, fall back to all mapped columns on this class hierarchy.
+
+        Always includes 'id' when present unless explicitly excluded (i.e., fields explicitly provided without id).
+        """
+        req = fields if fields is not None else getattr(self, "__crudkit_projection__", None)
+
+        if req:
+            # Normalize and split into (scalars, groups of dotted by root)
+            req_list = [p for p in (str(x).strip() for x in req) if p]
+            scalars, groups = _split_field_tokens(req_list)
+
+            out: Dict[str, Any] = {}
+
+            # Always include id unless the caller explicitly listed fields containing id
+            if "id" not in req_list and hasattr(self, "id"):
+                try:
+                    out["id"] = getattr(self, "id")
+                except Exception:
+                    pass
+
+            # Handle scalar tokens (may be columns, hybrids/properties, or relationships)
+            for name in scalars:
+                # Try loaded value first (never lazy-load)
+                val = _safe_get_loaded_attr(self, name)
+
+                # Final-hop getattr for root scalars (hybrids/@property) so they can compute.
+                if val is None:
+                    try:
+                        val = getattr(self, name)
+                    except Exception:
+                        val = None
+
+                # If it's a scalar ORM object (relationship), serialize its columns
+                mapper = _sa_mapper(val)
+                if mapper is not None:
+                    out[name] = _serialize_simple_obj(val)
+                    continue
+
+                # If it's a collection and no subfields were requested, emit a light list
+                if isinstance(val, (list, tuple)):
+                    out[name] = [_serialize_leaf(v) for v in val]
+                else:
+                    out[name] = val
+
+            # Handle dotted groups: root -> [tails]
+            for root, tails in groups.items():
+                root_val = _safe_get_loaded_attr(self, root)
+                if isinstance(root_val, (list, tuple)):
+                    # one-to-many collection → list of dicts with the requested tails
+                    out[root] = _serialize_collection(root_val, tails)
+                else:
+                    # many-to-one or scalar dotted; place each full dotted path as key
+                    for tail in tails:
+                        dotted = f"{root}.{tail}"
+                        out[dotted] = _deep_get_loaded(self, dotted)
+
+            # ← This was the placeholder before. We return the dict we just built.
             return out
 
-        result = {}
+        # Fallback: all mapped columns on this class hierarchy
+        result: Dict[str, Any] = {}
         for cls in self.__class__.__mro__:
             if hasattr(cls, "__table__"):
                 for column in cls.__table__.columns:
                     name = column.name
-                    result[name] = getattr(self, name)
+                    try:
+                        result[name] = getattr(self, name)
+                    except Exception:
+                        result[name] = None
         return result
 
-
 class Version(Base):
     __tablename__ = "versions"
 
diff --git a/crudkit/core/service.py b/crudkit/core/service.py
index 1220e04..d7fabc4 100644
--- a/crudkit/core/service.py
+++ b/crudkit/core/service.py
@@ -1,45 +1,37 @@
-from typing import Any, Callable, Dict, Iterable, List, Tuple, Type, TypeVar, Generic, Optional, Protocol, runtime_checkable, cast
-from sqlalchemy import and_, func, inspect, or_, text
-from sqlalchemy.engine import Engine, Connection
-from sqlalchemy.orm import Load, Session, raiseload, selectinload, with_polymorphic, Mapper, RelationshipProperty, class_mapper
-from sqlalchemy.orm.attributes import InstrumentedAttribute
-from sqlalchemy.orm.util import AliasedClass
-from sqlalchemy.sql import operators
-from sqlalchemy.sql.elements import UnaryExpression
+from __future__ import annotations
 
+from collections.abc import Iterable
+from dataclasses import dataclass
+from flask import current_app
+from typing import Any, Callable, Type, TypeVar, Generic, Optional, Protocol, runtime_checkable, cast
+from sqlalchemy import and_, func, inspect, or_, text, select, literal
+from sqlalchemy.engine import Engine, Connection
+from sqlalchemy.orm import Load, Session, with_polymorphic, Mapper, selectinload, with_loader_criteria, aliased, with_parent
+from sqlalchemy.orm.attributes import InstrumentedAttribute
+from sqlalchemy.sql import operators, visitors
+from sqlalchemy.sql.elements import UnaryExpression, ColumnElement
+
+from crudkit.core import to_jsonable, deep_diff, diff_to_patch, filter_to_columns, normalize_payload
 from crudkit.core.base import Version
-from crudkit.core.spec import CRUDSpec
+from crudkit.core.spec import CRUDSpec, CollPred
 from crudkit.core.types import OrderSpec, SeekWindow
 from crudkit.backend import BackendInfo, make_backend_info
+from crudkit.projection import compile_projection
 
-def _loader_options_for_fields(root_alias, model_cls, fields: list[str]) -> list[Load]:
-    """
-    For bare MANYTOONE names in fields (e.g. "location"), selectinload the relationship
-    and only fetch the related PK. This is enough for preselecting <select> inputs
-    without projecting the FK column on the root model.
-    """
-    opts: list[Load] = []
-    if not fields:
-        return opts
-
-    mapper = class_mapper(model_cls)
-    for name in fields:
-        prop = mapper.relationships.get(name)
-        if not isinstance(prop, RelationshipProperty):
-            continue
-        if prop.direction.name != "MANYTOONE":
-            continue
-
-        rel_attr = getattr(root_alias, name)
-        target_cls = prop.mapper.class_
-        # load_only PK if present; else just selectinload
-        id_attr = getattr(target_cls, "id", None)
-        if id_attr is not None:
-            opts.append(selectinload(rel_attr).load_only(id_attr))
-        else:
-            opts.append(selectinload(rel_attr))
-
-    return opts
+import logging
+log = logging.getLogger("crudkit.service")
+# logging.getLogger("crudkit.service").setLevel(logging.DEBUG)
+# Ensure our debug actually prints even if the app/root logger is WARNING+
+# if not log.handlers:
+#     _h = logging.StreamHandler()
+#     _h.setLevel(logging.DEBUG)
+#     _h.setFormatter(logging.Formatter(
+#         "%(asctime)s %(levelname)s %(name)s: %(message)s"
+#     ))
+#     log.addHandler(_h)
+#
+# log.setLevel(logging.DEBUG)
+# log.propagate = False
 
 @runtime_checkable
 class _HasID(Protocol):
@@ -58,14 +50,114 @@ class _SoftDeletable(Protocol):
     is_deleted: bool
 
 class _CRUDModelProto(_HasID, _HasTable, _HasADict, Protocol):
-    """Minimal surface that our CRUD service relies on. Soft-delete is optional."""
+    """Surface expected by CRUDService."""
     pass
 
 T = TypeVar("T", bound=_CRUDModelProto)
 
+# ---------------------------- utilities ----------------------------
+
+def _collect_tables_from_filters(filters) -> set:
+    seen = set()
+    stack = list(filters or [])
+    while stack:
+        node = stack.pop()
+
+        tbl = getattr(node, "table", None)
+        if tbl is not None:
+            cur = tbl
+            while cur is not None and cur not in seen:
+                seen.add(cur)
+                cur = getattr(cur, "element", None)
+
+        # follow only the common attributes; no generic visitor
+        left = getattr(node, "left", None)
+        if left is not None:
+            stack.append(left)
+        right = getattr(node, "right", None)
+        if right is not None:
+            stack.append(right)
+        elem = getattr(node, "element", None)
+        if elem is not None:
+            stack.append(elem)
+        clause = getattr(node, "clause", None)
+        if clause is not None:
+            stack.append(clause)
+        clauses = getattr(node, "clauses", None)
+        if clauses is not None:
+            try:
+                stack.extend(list(clauses))
+            except TypeError:
+                pass
+
+    return seen
+
+def _selectable_keys(sel) -> set[str]:
+    """
+    Return a set of stable string keys for a selectable/alias and its base,
+    so we can match when when different alias objects are used.
+    """
+    keys: set[str] = set()
+    cur = sel
+    while cur is not None:
+        k = getattr(cur, "key", None) or getattr(cur, "name", None)
+        if isinstance(k, str) and k:
+            keys.add(k)
+        cur = getattr(cur, "element", None)
+    return keys
+
+def _unwrap_ob(ob):
+    elem = getattr(ob, "element", None)
+    col = elem if elem is not None else ob
+
+    d = getattr(ob, "_direction", None)
+    if d is not None:
+        is_desc = (d is operators.desc_op) or (getattr(d, "name", "").upper() == "DESC")
+    elif isinstance(ob, UnaryExpression):
+        op = getattr(ob, "operator", None)
+        is_desc = (op is operators.desc_op) or (getattr(op, "name", "").upper() == "DESC")
+    else:
+        is_desc = False
+    return col, bool(is_desc)
+
+def _order_identity(col: ColumnElement):
+    table = getattr(col, "table", None)
+    table_key = getattr(table, "key", None) or id(table)
+    col_key = getattr(col, "key", None) or getattr(col, "name", None)
+    return (table_key, col_key)
+
+def _dedupe_order_by(order_by):
+    if not order_by:
+        return []
+    seen, out = set(), []
+    for ob in order_by:
+        col, _ = _unwrap_ob(ob)
+        ident = _order_identity(col)
+        if ident in seen:
+            continue
+        seen.add(ident)
+        out.append(ob)
+    return out
+
 def _is_truthy(val):
     return str(val).lower() in ('1', 'true', 'yes', 'on')
 
+def _normalize_fields_param(params: dict | None) -> list[str]:
+    if not params:
+        return []
+    raw = params.get("fields")
+    if isinstance(raw, (list, tuple)):
+        out: list[str] = []
+        for item in raw:
+            if isinstance(item, str):
+                out.extend([p for p in (s.strip() for s in item.split(",")) if p])
+        return out
+    if isinstance(raw, str):
+        return [p for p in (s.strip() for s in raw.split(",")) if p]
+    return []
+
+# ---------------------------- CRUD service ----------------------------
+
 class CRUDService(Generic[T]):
     def __init__(
         self,
@@ -79,14 +171,24 @@ class CRUDService(Generic[T]):
         self._session_factory = session_factory
         self.polymorphic = polymorphic
         self.supports_soft_delete = hasattr(model, 'is_deleted')
-        # Cache backend info once. If not provided, derive from session bind.
-        bind = self.session.get_bind()
-        eng: Engine = bind.engine if isinstance(bind, Connection) else cast(Engine, bind)
-        self.backend = backend or make_backend_info(eng)
+        self._backend: Optional[BackendInfo] = backend
+
+    # ---- infra
 
     @property
     def session(self) -> Session:
-        return self._session_factory()
+        try:
+            return current_app.extensions["crudkit"]["Session"]
+        except Exception:
+            return self._session_factory()
+
+    @property
+    def backend(self) -> BackendInfo:
+        if self._backend is None:
+            bind = self.session.get_bind()
+            eng: Engine = bind.engine if isinstance(bind, Connection) else cast(Engine, bind)
+            self._backend = make_backend_info(eng)
+        return self._backend
 
     def get_query(self):
         if self.polymorphic:
@@ -94,244 +196,96 @@ class CRUDService(Generic[T]):
             return self.session.query(poly), poly
         return self.session.query(self.model), self.model
 
-    def _resolve_required_includes(self, root_alias: Any, rel_field_names: Dict[Tuple[str, ...], List[str]]) -> List[Any]:
-        """
-        For each dotted path like ("location"), -> ["label"], look up the target
-        model's __crudkit_field_requires__ for the terminal field and produce
-        selectinload options prefixed with the relationship path, e.g.:
-            Room.__crudkit_field_requires__['label'] = ['room_function']
-            => selectinload(root.location).selectinload(Room.room_function)
-        """
-        opts: List[Any] = []
+    # ---- common building blocks
 
-        root_mapper: Mapper[Any] = cast(Mapper[Any], inspect(self.model))
+    def _apply_soft_delete_criteria_for_children(self, query, plan: "CRUDService._Plan", params):
+        # Skip if caller explicitly asked for deleted
+        if _is_truthy((params or {}).get("include_deleted")):
+            return query
 
-        for path, names in (rel_field_names or {}).items():
-            if not path:
+        seen = set()
+        for _base_alias, rel_attr, _target_alias in plan.join_paths:
+            prop = getattr(rel_attr, "property", None)
+            if not prop:
                 continue
-
-            current_alias = root_alias
-            current_mapper = root_mapper
-            rel_props: List[RelationshipProperty] = []
-
-            valid = True
-            for step in path:
-                rel = current_mapper.relationships.get(step)
-                if rel is None:
-                    valid = False
-                    break
-                rel_props.append(rel)
-                current_mapper = cast(Mapper[Any], inspect(rel.entity.entity))
-            if not valid:
+            target_cls = getattr(prop.mapper, "class_", None)
+            if not target_cls or target_cls in seen:
                 continue
+            seen.add(target_cls)
+            # Only apply to models that support soft delete
+            if hasattr(target_cls, "is_deleted"):
+                query = query.options(
+                    with_loader_criteria(
+                        target_cls,
+                        lambda cls: cls.is_deleted == False,
+                        include_aliases=True
+                    )
+                )
+        return query
 
-            target_cls = current_mapper.class_
+    def _order_clauses(self, order_spec, invert: bool = False):
+        clauses = []
+        for c, is_desc in zip(order_spec.cols, order_spec.desc):
+            d = not is_desc if invert else is_desc
+            clauses.append(c.desc() if d else c.asc())
+        return clauses
 
-            requires = getattr(target_cls, "__crudkit_field_requires__", None)
-            if not isinstance(requires, dict):
-                continue
-
-            for field_name in names:
-                needed: Iterable[str] = requires.get(field_name, [])
-                for rel_need in needed:
-                    loader = selectinload(getattr(root_alias, rel_props[0].key))
-                    for rp in rel_props[1:]:
-                        loader = loader.selectinload(getattr(getattr(root_alias, rp.parent.class_.__name__.lower(), None) or rp.parent.class_, rp.key))
-
-                    loader = loader.selectinload(getattr(target_cls, rel_need))
-                    opts.append(loader)
-
-        return opts
-
-    def _extract_order_spec(self, root_alias, given_order_by):
-        """
-        SQLAlchemy 2.x only:
-        Normalize order_by into (cols, desc_flags). Supports plain columns and
-        col.asc()/col.desc() (UnaryExpression). Avoids boolean evaluation of clauses.
-        """
-        from sqlalchemy.sql import operators
-        from sqlalchemy.sql.elements import UnaryExpression
-
-        given = self._stable_order_by(root_alias, given_order_by)
-
-        cols, desc_flags = [], []
-        for ob in given:
-            # Unwrap column if this is a UnaryExpression produced by .asc()/.desc()
-            elem = getattr(ob, "element", None)
-            col = elem if elem is not None else ob  # don't use "or" with SA expressions
-
-            # Detect direction in SA 2.x
-            is_desc = False
-            dir_attr = getattr(ob, "_direction", None)
-            if dir_attr is not None:
-                is_desc = (dir_attr is operators.desc_op) or (getattr(dir_attr, "name", "").upper() == "DESC")
-            elif isinstance(ob, UnaryExpression):
-                op = getattr(ob, "operator", None)
-                is_desc = (op is operators.desc_op) or (getattr(op, "name", "").upper() == "DESC")
-
-            cols.append(col)
-            desc_flags.append(bool(is_desc))
-
-        from crudkit.core.types import OrderSpec
-        return OrderSpec(cols=tuple(cols), desc=tuple(desc_flags))
-
-    def _key_predicate(self, spec: OrderSpec, key_vals: list[Any], backward: bool):
-        """
-        Build lexicographic predicate for keyset seek.
-        For backward traversal, import comparisons.
-        """
-        if not key_vals:
+    def _anchor_key_for_page(self, params, per_page: int, page: int):
+        """Return the keyset tuple for the last row of the previous page, or None for page 1."""
+        if page <= 1:
             return None
-        conds = []
-        for i, col in enumerate(spec.cols):
-            ties = [spec.cols[j] == key_vals[j] for j in range(i)]
-            is_desc = spec.desc[i]
-            if not backward:
-                op = col < key_vals[i] if is_desc else col > key_vals[i]
-            else:
-                op = col > key_vals[i] if is_desc else col < key_vals[i]
-            conds.append(and_(*ties, op))
-        return or_(*conds)
 
-    def _pluck_key(self, obj: Any, spec: OrderSpec) -> list[Any]:
-        out = []
-        for c in spec.cols:
-            key = getattr(c, "key", None) or getattr(c, "name", None)
-            out.append(getattr(obj, key))
-        return out
-
-    def seek_window(
-        self,
-        params: dict | None = None,
-        *,
-        key: list[Any] | None = None,
-        backward: bool = False,
-        include_total: bool = True,
-    ) -> "SeekWindow[T]":
-        """
-        Transport-agnostic keyset pagination that preserves all the goodies from `list()`:
-          - filters, includes, joins, field projection, eager loading, soft-delete
-          - deterministic ordering (user sort + PK tiebreakers)
-          - forward/backward seek via `key` and `backward`
-        Returns a SeekWindow with items, first/last keys, order spec, limit, and optional total.
-        """
-        session = self.session
         query, root_alias = self.get_query()
+        query = self._apply_not_deleted(query, root_alias, params)
 
-        spec = CRUDSpec(self.model, params or {}, root_alias)
-
-        filters = spec.parse_filters()
-        order_by = spec.parse_sort()
-
-        root_fields, rel_field_names, root_field_names = spec.parse_fields()
-
-        for path, names in (rel_field_names or {}).items():
-            if "label" in names:
-                rel_name = path[0]
-                rel_attr = getattr(root_alias, rel_name, None)
-                if rel_attr is not None:
-                    query = query.options(selectinload(rel_attr))
-
-        # Soft delete filter
-        if self.supports_soft_delete and not _is_truthy(params.get("include_deleted")):
-            query = query.filter(getattr(root_alias, "is_deleted") == False)
-
-        # Parse filters first
-        if filters:
-            query = query.filter(*filters)
-
-        # Includes + joins (so relationship fields like brand.name, location.label work)
-        spec.parse_includes()
-        for parent_alias, relationship_attr, target_alias in spec.get_join_paths():
-            rel_attr = cast(InstrumentedAttribute, relationship_attr)
-            target = cast(Any, target_alias)
-            query = query.join(target, rel_attr.of_type(target), isouter=True)
-
-        # Fields/projection: load_only for root columns, eager loads for relationships
-        only_cols = [c for c in root_fields if isinstance(c, InstrumentedAttribute)]
-        if only_cols:
-            query = query.options(Load(root_alias).load_only(*only_cols))
-        for eager in spec.get_eager_loads(root_alias, fields_map=rel_field_names):
-            query = query.options(eager)
-
-        for opt in self._resolve_required_includes(root_alias, rel_field_names):
-            query = query.options(opt)
-
-        # Order + limit
-        order_spec = self._extract_order_spec(root_alias, order_by)  # SA2-only helper
-        limit, _ = spec.parse_pagination()
-        if not limit or limit <= 0:
-            limit = 50  # sensible default
-
-        # Keyset predicate
-        if key:
-            pred = self._key_predicate(order_spec, key, backward)
-            if pred is not None:
-                query = query.filter(pred)
-
-        # Apply ordering. For backward, invert SQL order then reverse in-memory for display.
-        if not backward:
-            clauses = []
-            for col, is_desc in zip(order_spec.cols, order_spec.desc):
-                clauses.append(col.desc() if is_desc else col.asc())
-            query = query.order_by(*clauses).limit(limit)
-            items = query.all()
-        else:
-            inv_clauses = []
-            for col, is_desc in zip(order_spec.cols, order_spec.desc):
-                inv_clauses.append(col.asc() if is_desc else col.desc())
-            query = query.order_by(*inv_clauses).limit(limit)
-            items = list(reversed(query.all()))
-
-        # Tag projection so your renderer knows what fields were requested
-        proj = []
-        if root_field_names:
-            proj.extend(root_field_names)
-        if root_fields:
-            proj.extend(c.key for c in root_fields)
-        for path, names in (rel_field_names or {}).items():
-            prefix = ".".join(path)
-            for n in names:
-                proj.append(f"{prefix}.{n}")
-        if proj and "id" not in proj and hasattr(self.model, "id"):
-            proj.insert(0, "id")
-        if proj:
-            for obj in items:
-                try:
-                    setattr(obj, "__crudkit_projection__", tuple(proj))
-                except Exception:
-                    pass
-
-        # Boundary keys for cursor encoding in the API layer
-        first_key = self._pluck_key(items[0], order_spec) if items else None
-        last_key  = self._pluck_key(items[-1], order_spec) if items else None
-
-        # Optional total that’s safe under JOINs (COUNT DISTINCT ids)
-        total = None
-        if include_total:
-            base = self.session.query(getattr(root_alias, "id"))
-            if self.supports_soft_delete and not _is_truthy(params.get("include_deleted")):
-                base = base.filter(getattr(root_alias, "is_deleted") == False)
+        plan = self._plan(params, root_alias)
+        # Make sure joins/filters match the real query
+        query = self._apply_firsthop_strategies(query, root_alias, plan)
+        if plan.filters:
+            filters = self._final_filters(root_alias, plan)
             if filters:
-                base = base.filter(*filters)
-            # replicate the same joins used above
-            for parent_alias, relationship_attr, target_alias in spec.get_join_paths():
-                rel_attr = cast(InstrumentedAttribute, relationship_attr)
-                target = cast(Any, target_alias)
-                base = base.join(target, rel_attr.of_type(target), isouter=True)
-            total = self.session.query(func.count()).select_from(base.order_by(None).distinct().subquery()).scalar() or 0
+                query = query.filter(*filters)
 
-        from crudkit.core.types import SeekWindow  # avoid circulars at module top
-        return SeekWindow(
-            items=items,
-            limit=limit,
-            first_key=first_key,
-            last_key=last_key,
-            order=order_spec,
-            total=total,
+        order_spec = self._extract_order_spec(root_alias, plan.order_by)
+
+        # Inner subquery must be ordered exactly like the real query
+        inner = query.order_by(*self._order_clauses(order_spec, invert=False))
+
+        # IMPORTANT: Build subquery that actually exposes the order-by columns
+        # under predictable names, then select FROM that and reference subq.c[...]
+        subq = inner.with_entities(*order_spec.cols).subquery()
+
+        # Map the order columns to the subquery columns by key/name
+        cols_on_subq = []
+        for col in order_spec.cols:
+            key = getattr(col, "key", None) or getattr(col, "name", None)
+            if not key:
+                # Fallback, but frankly your order cols should have names
+                raise ValueError("Order-by column is missing a key/name")
+            cols_on_subq.append(getattr(subq.c, key))
+
+        # Now the outer anchor query orders and offsets on the subquery columns
+        anchor_q = (
+            self.session
+            .query(*cols_on_subq)
+            .select_from(subq)
+            .order_by(*[
+                (c.desc() if is_desc else c.asc())
+                for c, is_desc in zip(cols_on_subq, order_spec.desc)
+            ])
         )
 
-    # Helper: default ORDER BY for MSSQL when paginating without explicit order
+        offset = max(0, (page - 1) * per_page - 1)
+        row = anchor_q.offset(offset).limit(1).first()
+        if not row:
+            return None
+        return list(row)  # tuple-like -> list for _key_predicate
+
+    def _apply_not_deleted(self, query, root_alias, params):
+        if self.supports_soft_delete and not _is_truthy((params or {}).get("include_deleted")):
+            return query.filter(getattr(root_alias, "is_deleted") == False)
+        return query
+
     def _default_order_by(self, root_alias):
         mapper: Mapper[Any] = cast(Mapper[Any], inspect(self.model))
         cols = []
@@ -343,214 +297,581 @@ class CRUDService(Generic[T]):
         return cols or [text("1")]
 
     def _stable_order_by(self, root_alias, given_order_by):
-        """
-        Ensure deterministic ordering by appending PK columns as tiebreakers.
-        If no order is provided, fall back to default primary-key order.
-        """
         order_by = list(given_order_by or [])
         if not order_by:
-            return self._default_order_by(root_alias)
-
+            return _dedupe_order_by(self._default_order_by(root_alias))
+        order_by = _dedupe_order_by(order_by)
         mapper: Mapper[Any] = cast(Mapper[Any], inspect(self.model))
-        pk_cols = []
-        for col in mapper.primary_key:
+        present = {_order_identity(_unwrap_ob(ob)[0]) for ob in order_by}
+        for pk in mapper.primary_key:
             try:
-                pk_cols.append(getattr(root_alias, col.key))
-            except ArithmeticError:
-                pk_cols.append(col)
+                pk_col = getattr(root_alias, pk.key)
+            except AttributeError:
+                pk_col = pk
+            ident = _order_identity(pk_col)
+            if ident not in present:
+                order_by.append(pk_col.asc())
+                present.add(ident)
+        return order_by
 
-        return [*order_by, *pk_cols]
+    def _extract_order_spec(self, root_alias, given_order_by):
+        given = self._stable_order_by(root_alias, given_order_by)
+        cols, desc_flags = [], []
+        for ob in given:
+            elem = getattr(ob, "element", None)
+            col = elem if elem is not None else ob
+            is_desc = False
+            dir_attr = getattr(ob, "_direction", None)
+            if dir_attr is not None:
+                is_desc = (dir_attr is operators.desc_op) or (getattr(dir_attr, "name", "").upper() == "DESC")
+            elif isinstance(ob, UnaryExpression):
+                op = getattr(ob, "operator", None)
+                is_desc = (op is operators.desc_op) or (getattr(op, "name", "").upper() == "DESC")
+            cols.append(col)
+            desc_flags.append(bool(is_desc))
+        return OrderSpec(cols=tuple(cols), desc=tuple(desc_flags))
 
-    def get(self, id: int, params=None) -> T | None:
-        query, root_alias = self.get_query()
+    def _key_predicate(self, spec: OrderSpec, key_vals: list[Any], backward: bool):
+        if not key_vals:
+            return None
+        conds = []
+        for i, col in enumerate(spec.cols):
+            ties = [spec.cols[j] == key_vals[j] for j in range(i)]
+            is_desc = spec.desc[i]
+            op = (col < key_vals[i]) if is_desc ^ backward else (col > key_vals[i])
+            conds.append(and_(*ties, op))
+        return or_(*conds)
 
-        include_deleted = False
-        root_fields = []
-        root_field_names = {}
-        rel_field_names = {}
+    # ---- planning and application
 
+    @dataclass(slots=True)
+    class _Plan:
+        spec: Any
+        filters: Any
+        order_by: Any
+        limit: Any
+        offset: Any
+        root_fields: Any
+        rel_field_names: Any
+        root_field_names: Any
+        collection_field_names: Any
+        join_paths: Any
+        filter_tables: Any
+        filter_table_keys: Any
+        req_fields: Any
+        proj_opts: Any
+
+    def _plan(self, params, root_alias) -> _Plan:
+        req_fields = _normalize_fields_param(params)
         spec = CRUDSpec(self.model, params or {}, root_alias)
-        if params:
-            if self.supports_soft_delete:
-                include_deleted = _is_truthy(params.get('include_deleted'))
-        if self.supports_soft_delete and not include_deleted:
-            query = query.filter(getattr(root_alias, "is_deleted") == False)
-        query = query.filter(getattr(root_alias, "id") == id)
 
+        filters = spec.parse_filters()
+        order_by = spec.parse_sort()
+        limit, offset = spec.parse_pagination()
+        root_fields, rel_field_names, root_field_names, collection_field_names = spec.parse_fields() if params else ([], {}, {}, {})
         spec.parse_includes()
+        join_paths = tuple(spec.get_join_paths())
+        filter_tables = _collect_tables_from_filters(filters)
+        fkeys = set()
+        _, proj_opts = compile_projection(self.model, req_fields) if req_fields else ([], [])
 
-        for parent_alias, relationship_attr, target_alias in spec.get_join_paths():
-            rel_attr = cast(InstrumentedAttribute, relationship_attr)
-            target = cast(Any, target_alias)
-            query = query.join(target, rel_attr.of_type(target), isouter=True)
+        # filter_tables = ()
+        # fkeys = set()
 
-        if params:
-            root_fields, rel_field_names, root_field_names = spec.parse_fields()
+        return self._Plan(
+            spec=spec, filters=filters, order_by=order_by, limit=limit, offset=offset,
+            root_fields=root_fields, rel_field_names=rel_field_names,
+            root_field_names=root_field_names, collection_field_names=collection_field_names,
+            join_paths=join_paths, filter_tables=filter_tables, filter_table_keys=fkeys,
+            req_fields=req_fields, proj_opts=proj_opts
+        )
 
-        fields = (params or {}).get("fields") if isinstance(params, dict) else None
-        if fields:
-            for opt in _loader_options_for_fields(root_alias, self.model, fields):
+    def _apply_projection_load_only(self, query, root_alias, plan: _Plan):
+        only_cols = [c for c in plan.root_fields if isinstance(c, InstrumentedAttribute)]
+        return query.options(Load(root_alias).load_only(*only_cols)) if only_cols else query
+
+    def _apply_firsthop_strategies(self, query, root_alias, plan: _Plan):
+        nested_first_hops = { p[0] for p in (plan.rel_field_names or {}).keys() if len(p) > 1 }
+        joined_rel_keys = set()
+
+        # Existing behavior: join everything in join_paths (to-one), selectinload collections
+        for base_alias, rel_attr, target_alias in plan.join_paths:
+            if base_alias is not root_alias:
+                continue
+            prop = getattr(rel_attr, "property", None)
+            is_collection = bool(getattr(prop, "uselist", False))
+
+            if not is_collection:
+                query = query.join(target_alias, rel_attr.of_type(target_alias), isouter=True)
+                joined_rel_keys.add(prop.key if prop is not None else rel_attr.key)
+            else:
+                opt = selectinload(rel_attr)
+                child_names = (plan.collection_field_names or {}).get(rel_attr.key, [])
+                if child_names:
+                    target_cls = prop.mapper.class_
+                    cols = [getattr(target_cls, n, None) for n in child_names]
+                    cols = [c for c in cols if isinstance(c, InstrumentedAttribute)]
+                    if cols:
+                        opt = opt.load_only(*cols)
                 query = query.options(opt)
 
-        only_cols = [c for c in root_fields if isinstance(c, InstrumentedAttribute)]
-        if only_cols:
-            query = query.options(Load(root_alias).load_only(*only_cols))
+        # inside CRUDService._apply_firsthop_strategies
+        # ...
+        # NEW: if a first-hop to-one relationship’s target table is present in filter expressions,
+        # make sure we actually JOIN it (outer) so filters don’t create a cartesian product.
+        if plan.filter_tables:
+            mapper: Mapper[Any] = cast(Mapper[Any], inspect(self.model))
+            for rel in mapper.relationships:
+                if rel.uselist:
+                    continue  # only first-hop to-one here
+                target_cls = rel.mapper.class_
+                target_tbl = getattr(target_cls, "__table__", None)
+                if target_tbl is None:
+                    continue
+                if target_tbl in plan.filter_tables:
+                    if rel.key in joined_rel_keys:
+                        continue  # already joined via join_paths
 
-        for eager in spec.get_eager_loads(root_alias, fields_map=rel_field_names):
-            query = query.options(eager)
+                    # alias when joining same-entity relationships (User->User supervisor)
+                    ta = aliased(target_cls) if target_cls is self.model else target_cls
+                    query = query.join(getattr(root_alias, rel.key).of_type(ta), isouter=True)
+                    joined_rel_keys.add(rel.key)
 
-        if params:
-            fields = params.get("fields") or []
-            for opt in _loader_options_for_fields(root_alias, self.model, fields):
-                query = query.options(opt)
+        if log.isEnabledFor(logging.DEBUG):
+            info = []
+            for base_alias, rel_attr, target_alias in plan.join_paths:
+                if base_alias is not root_alias:
+                    continue
+                prop = getattr(rel_attr, "property", None)
+                sel = getattr(target_alias, "selectable", None)
+                info.append({
+                    "rel": (getattr(prop, "key", getattr(rel_attr, "key", "?"))),
+                    "collection": bool(getattr(prop, "uselist", False)),
+                    "target_keys": list(_selectable_keys(sel)) if sel is not None else [],
+                    "joined": (getattr(prop, "key", None) in joined_rel_keys),
+                })
+            log.debug("FIRSTHOP: %s.%s first-hop paths: %s",
+                      self.model.__name__, getattr(root_alias, "__table__", type(root_alias)).key,
+                      info)
 
-        obj = query.first()
+        return query
 
-        proj = []
-        if root_field_names:
-            proj.extend(root_field_names)
-        if root_fields:
-            proj.extend(c.key for c in root_fields)
-        for path, names in (rel_field_names or {}).items():
+    def _apply_proj_opts(self, query, plan: _Plan):
+        return query.options(*plan.proj_opts) if plan.proj_opts else query
+
+    def _projection_meta(self, plan: _Plan):
+        if plan.req_fields:
+            proj = list(dict.fromkeys(plan.req_fields))
+            return ["id"] + proj if "id" not in proj and hasattr(self.model, "id") else proj
+
+        proj: list[str] = []
+        if plan.root_field_names:
+            proj.extend(plan.root_field_names)
+        if plan.root_fields:
+            proj.extend(c.key for c in plan.root_fields if hasattr(c, "key"))
+        for path, names in (plan.rel_field_names or {}).items():
             prefix = ".".join(path)
             for n in names:
                 proj.append(f"{prefix}.{n}")
-
         if proj and "id" not in proj and hasattr(self.model, "id"):
             proj.insert(0, "id")
+        return proj
 
-        if proj:
+    def _tag_projection(self, items, proj):
+        if not proj:
+            return
+        for obj in items if isinstance(items, list) else [items]:
             try:
                 setattr(obj, "__crudkit_projection__", tuple(proj))
             except Exception:
                 pass
 
+    def _rebind_filters_to_firsthop_aliases(self, filters, root_alias, plan):
+        """Make filter expressions use the exact same alias objects as our JOINs."""
+        if not filters:
+            return filters
+
+        # Map first-hop target selectable keysets -> the exact selectable object we JOINed with
+        alias_map = {}
+        for base_alias, _rel_attr, target_alias in plan.join_paths:
+            if base_alias is not root_alias:
+                continue
+            sel = getattr(target_alias, "selectable", None)
+            if sel is not None:
+                alias_map[frozenset(_selectable_keys(sel))] = sel
+
+        if not alias_map:
+            return filters
+
+        def replace(elem):
+            tbl = getattr(elem, "table", None)
+            if tbl is None:
+                return elem
+            keyset = frozenset(_selectable_keys(tbl))
+            new_sel = alias_map.get(keyset)
+            if new_sel is None or new_sel is tbl:
+                return elem
+
+            colkey = getattr(elem, "key", None) or getattr(elem, "name", None)
+            if not colkey:
+                return elem
+            try:
+                return getattr(new_sel.c, colkey)
+            except Exception:
+                return elem
+
+        return [visitors.replacement_traverse(f, {}, replace) for f in filters]
+
+    def _final_filters(self, root_alias, plan):
+        """
+        Return filters where:
+          - root/to-one predicates are kept as SQLAlchemy expressions.
+          - first-hop collection predicates (CollPred) are rebuilt into a single
+            EXISTS via rel.any(...) with one alias per collection table.
+        """
+        filters = list(plan.filters or [])
+        if not filters:
+            return []
+
+        # 1) Build a map of first-hop relationships: TABLE -> (rel_attr, target_cls)
+        coll_map = {}
+        for base_alias, rel_attr, target_alias in plan.join_paths:
+            if base_alias is not root_alias:
+                continue
+            prop = getattr(rel_attr, "property", None)
+            if not prop or not getattr(prop, "uselist", False):
+                continue
+            target_cls = prop.mapper.class_
+            tbl = getattr(target_cls, "__table__", None)
+            if tbl is not None:
+                coll_map[tbl] = (rel_attr, target_cls)
+
+        # 2) Split raw filters into normal SQLA and CollPreds (by target table)
+        normal_filters = []
+        by_table: dict[Any, list[CollPred]] = {}
+        for f in filters:
+            if isinstance(f, CollPred):
+                by_table.setdefault(f.table, []).append(f)
+            else:
+                normal_filters.append(f)
+
+        # 3) Rebuild each table group into ONE .any(...) using one alias
+        from sqlalchemy.orm import aliased
+        from sqlalchemy import and_
+
+        exists_filters = []
+        for tbl, preds in by_table.items():
+            if tbl not in coll_map:
+                # Safety: if it's not a first-hop collection, ignore or raise
+                continue
+            rel_attr, target_cls = coll_map[tbl]
+            ta = aliased(target_cls)
+
+            built = []
+            for p in preds:
+                col = getattr(ta, p.col_key)
+                op = p.op
+                val = p.value
+                if op == 'icontains':
+                    built.append(col.ilike(f"%{val}%"))
+                elif op == 'eq':
+                    built.append(col == val)
+                elif op == 'ne':
+                    built.append(col != val)
+                elif op == 'in':
+                    vs = val if isinstance(val, (list, tuple, set)) else [val]
+                    built.append(col.in_(vs))
+                elif op == 'nin':
+                    vs = val if isinstance(val, (list, tuple, set)) else [val]
+                    built.append(~col.in_(vs))
+                elif op == 'lt':
+                    built.append(col < val)
+                elif op == 'lte':
+                    built.append(col <= val)
+                elif op == 'gt':
+                    built.append(col > val)
+                elif op == 'gte':
+                    built.append(col >= val)
+                else:
+                    # unknown op — skip or raise
+                    continue
+
+            # enforce child soft delete inside the EXISTS
+            if hasattr(target_cls, "is_deleted"):
+                built.append(ta.is_deleted == False)
+
+            crit = and_(*built) if built else None
+            exists_filters.append(rel_attr.of_type(ta).any(crit) if crit is not None
+                                  else rel_attr.of_type(ta).any())
+
+        # 4) Final filter list = normal SQLA filters + all EXISTS filters
+        return normal_filters + exists_filters
+
+    # ---- public read ops
+
+    def page(self, params=None, *, page: int = 1, per_page: int = 50, include_total: bool = True):
+        # Ensure seek_window uses `per_page`
+        params = dict(params or {})
+        params["limit"] = per_page
+
+        anchor_key = self._anchor_key_for_page(params, per_page, page)
+        win = self.seek_window(params, key=anchor_key, backward=False, include_total=include_total)
+
+        pages = None
+        if include_total and win.total is not None and per_page:
+            # class ceil(total / per_page) // per_page
+            pages = (win.total + per_page - 1) // per_page
+
+        return {
+            "items": win.items,
+            "page": page,
+            "per_page": per_page,
+            "total": win.total,
+            "pages": pages,
+            "order": [str(c) for c in win.order.cols],
+        }
+
+    def seek_window(
+        self,
+        params: dict | None = None,
+        *,
+        key: list[Any] | None = None,
+        backward: bool = False,
+        include_total: bool = True,
+    ) -> "SeekWindow[T]":
+        session = self.session
+        query, root_alias = self.get_query()
+        query = self._apply_not_deleted(query, root_alias, params)
+
+        plan = self._plan(params, root_alias)
+        query = self._apply_projection_load_only(query, root_alias, plan)
+        query = self._apply_firsthop_strategies(query, root_alias, plan)
+        query = self._apply_soft_delete_criteria_for_children(query, plan, params)
+        if plan.filters:
+            filters = self._final_filters(root_alias, plan)
+            if filters:
+                query = query.filter(*filters)
+
+        order_spec = self._extract_order_spec(root_alias, plan.order_by)
+        limit = 50 if plan.limit is None else (None if plan.limit == 0 else plan.limit)
+
+        if key:
+            pred = self._key_predicate(order_spec, key, backward)
+            if pred is not None:
+                query = query.filter(pred)
+
+        clauses = [(c.desc() if d else c.asc()) for c, d in zip(order_spec.cols, order_spec.desc)]
+        if backward:
+            clauses = [(c.asc() if d else c.desc()) for c, d in zip(order_spec.cols, order_spec.desc)]
+        query = query.order_by(*clauses)
+        if limit is not None:
+            query = query.limit(limit)
+
+        query = self._apply_proj_opts(query, plan)
+        rows = query.all()
+        items = list(reversed(rows)) if backward else rows
+
+        proj = self._projection_meta(plan)
+        self._tag_projection(items, proj)
+
+        # cursor keys
+        def pluck(obj):
+            vals = []
+            alias_to_rel = {}
+            for _p, rel_attr, target_alias in plan.join_paths:
+                sel = getattr(target_alias, "selectable", None)
+                if sel is not None:
+                    alias_to_rel[sel] = rel_attr.key
+
+            for col in order_spec.cols:
+                keyname = getattr(col, "key", None) or getattr(col, "name", None)
+                if keyname and hasattr(obj, keyname):
+                    vals.append(getattr(obj, keyname)); continue
+                table = getattr(col, "table", None)
+                relname = alias_to_rel.get(table)
+                if relname and keyname:
+                    relobj = getattr(obj, relname, None)
+                    if relobj is not None and hasattr(relobj, keyname):
+                        vals.append(getattr(relobj, keyname)); continue
+                raise ValueError("unpluckable")
+            return vals
+
+        try:
+            first_key = pluck(items[0]) if items else None
+            last_key  = pluck(items[-1]) if items else None
+        except Exception:
+            first_key = last_key = None
+
+        total = None
+        if include_total:
+            base = session.query(getattr(root_alias, "id"))
+            base = self._apply_not_deleted(base, root_alias, params)
+            for _b, rel_attr, target_alias in plan.join_paths:
+                if not bool(getattr(getattr(rel_attr, "property", None), "uselist", False)):
+                    base = base.join(target_alias, rel_attr.of_type(target_alias), isouter=True)
+            if plan.filters:
+                filters = self._final_filters(root_alias, plan)
+                if filters:
+                    base = base.filter(*filters)  # <-- use base, not query
+            total = session.query(func.count()).select_from(
+                base.order_by(None).distinct().subquery()
+            ).scalar() or 0
+
+        if log.isEnabledFor(logging.DEBUG):
+            log.debug("QUERY: %s", str(query))
+
+        window_limit_for_body = 0 if limit is None and (plan.limit == 0) else (limit or 50)
+        return SeekWindow(
+            items=items,
+            limit=window_limit_for_body,
+            first_key=first_key,
+            last_key=last_key,
+            order=order_spec,
+            total=total,
+        )
+
+    def get(self, id: int, params=None) -> T | None:
+        query, root_alias = self.get_query()
+        query = self._apply_not_deleted(query, root_alias, params)
+
+        plan = self._plan(params, root_alias)
+        query = self._apply_projection_load_only(query, root_alias, plan)
+        query = self._apply_firsthop_strategies(query, root_alias, plan)
+        query = self._apply_soft_delete_criteria_for_children(query, plan, params)
+        if plan.filters:
+            filters = self._final_filters(root_alias, plan)
+            if filters:
+                query = query.filter(*filters)
+        query = query.filter(getattr(root_alias, "id") == id)
+        query = self._apply_proj_opts(query, plan)
+
+        obj = query.first()
+        proj = self._projection_meta(plan)
+        if obj:
+            self._tag_projection(obj, proj)
+
+        if log.isEnabledFor(logging.DEBUG):
+            log.debug("QUERY: %s", str(query))
         return obj or None
 
     def list(self, params=None) -> list[T]:
         query, root_alias = self.get_query()
-
-        root_fields = []
-        root_field_names = {}
-        rel_field_names = {}
-
-        if params:
-            if self.supports_soft_delete:
-                include_deleted = _is_truthy(params.get('include_deleted'))
-                if not include_deleted:
-                    query = query.filter(getattr(root_alias, "is_deleted") == False)
-
-            spec = CRUDSpec(self.model, params or {}, root_alias)
-            filters = spec.parse_filters()
-            order_by = spec.parse_sort()
-            limit, offset = spec.parse_pagination()
-            spec.parse_includes()
-
-            for parent_alias, relationship_attr, target_alias in spec.get_join_paths():
-                rel_attr = cast(InstrumentedAttribute, relationship_attr)
-                target = cast(Any, target_alias)
-                query = query.join(target, rel_attr.of_type(target), isouter=True)
-
-            if params:
-                root_fields, rel_field_names, root_field_names = spec.parse_fields()
-
-            only_cols = [c for c in root_fields if isinstance(c, InstrumentedAttribute)]
-            if only_cols:
-                query = query.options(Load(root_alias).load_only(*only_cols))
-
-            for eager in spec.get_eager_loads(root_alias, fields_map=rel_field_names):
-                query = query.options(eager)
-
-            if params:
-                fields = params.get("fields") or []
-                for opt in _loader_options_for_fields(root_alias, self.model, fields):
-                    query = query.options(opt)
-
+        plan = self._plan(params, root_alias)
+        query = self._apply_not_deleted(query, root_alias, params)
+        query = self._apply_projection_load_only(query, root_alias, plan)
+        query = self._apply_firsthop_strategies(query, root_alias, plan)
+        query = self._apply_soft_delete_criteria_for_children(query, plan, params)
+        if plan.filters:
+            filters = self._final_filters(root_alias, plan)
             if filters:
                 query = query.filter(*filters)
 
-            # MSSQL: requires ORDER BY when using OFFSET (and SQLA will use OFFSET for limit+offset).
-            paginating = (limit is not None) or (offset is not None and offset != 0)
-            if paginating and not order_by and self.backend.requires_order_by_for_offset:
-                order_by = self._default_order_by(root_alias)
+        order_by = plan.order_by
+        paginating = (plan.limit is not None) or (plan.offset not in (None, 0))
+        if paginating and not order_by and self.backend.requires_order_by_for_offset:
+            order_by = self._default_order_by(root_alias)
+        order_by = _dedupe_order_by(order_by)
+        if order_by:
+            query = query.order_by(*order_by)
 
-            if order_by:
-                query = query.order_by(*order_by)
-
-            # Only apply offset/limit when not None.
-            if offset is not None and offset != 0:
-                query = query.offset(offset)
-            if limit is not None and limit > 0:
-                query = query.limit(limit)
+        default_cap = getattr(current_app.config, "CRUDKIT_DEFAULT_LIST_LIMIT", 200)
+        if plan.offset:
+            query = query.offset(plan.offset)
+        if plan.limit and plan.limit > 0:
+            query = query.limit(plan.limit)
+        elif plan.limit is None and default_cap:
+            query = query.limit(default_cap)
 
+        query = self._apply_proj_opts(query, plan)
         rows = query.all()
 
-        proj = []
-        if root_field_names:
-            proj.extend(root_field_names)
-        if root_fields:
-            proj.extend(c.key for c in root_fields)
-        for path, names in (rel_field_names or {}).items():
-            prefix = ".".join(path)
-            for n in names:
-                proj.append(f"{prefix}.{n}")
-
-        if proj and "id" not in proj and hasattr(self.model, "id"):
-            proj.insert(0, "id")
-
-        if proj:
-            for obj in rows:
-                try:
-                    setattr(obj, "__crudkit_projection__", tuple(proj))
-                except Exception:
-                    pass
+        proj = self._projection_meta(plan)
+        self._tag_projection(rows, proj)
 
+        if log.isEnabledFor(logging.DEBUG):
+            log.debug("QUERY: %s", str(query))
         return rows
 
-    def create(self, data: dict, actor=None) -> T:
+    # ---- write ops
+
+    def create(self, data: dict, actor=None, *, commit: bool = True) -> T:
+        session = self.session
         obj = self.model(**data)
-        self.session.add(obj)
-        self.session.commit()
-        self._log_version("create", obj, actor)
+        session.add(obj)
+        session.flush()
+        self._log_version("create", obj, actor, commit=commit)
+        if commit:
+            session.commit()
         return obj
 
-    def update(self, id: int, data: dict, actor=None) -> T:
-        obj = self.get(id)
+    def update(self, id: int, data: dict, actor=None, *, commit: bool = True) -> T:
+        session = self.session
+        obj = session.get(self.model, id)
         if not obj:
             raise ValueError(f"{self.model.__name__} with ID {id} not found.")
-        valid_fields = {c.name for c in self.model.__table__.columns}
-        for k, v in data.items():
-            if k in valid_fields:
-                setattr(obj, k, v)
-        self.session.commit()
-        self._log_version("update", obj, actor)
+
+        before = obj.as_dict()
+        norm = normalize_payload(data, self.model)
+        incoming = filter_to_columns(norm, self.model)
+        desired = {**before, **incoming}
+
+        proposed = deep_diff(before, desired, ignore_keys={"id", "created_at", "updated_at"}, list_mode="index")
+        patch = diff_to_patch(proposed)
+        if not patch:
+            return obj
+
+        for k, v in patch.items():
+            setattr(obj, k, v)
+
+        dirty = any(inspect(obj).attrs[k].history.has_changes() for k in patch.keys())
+        if not dirty:
+            return obj
+
+        if commit:
+            session.commit()
+
+        after = obj.as_dict()
+        actual = deep_diff(before, after, ignore_keys={"id", "created_at", "updated_at"}, list_mode="index")
+        if not (actual["added"] or actual["removed"] or actual["changed"]):
+            return obj
+
+        self._log_version("update", obj, actor, metadata={"diff": actual, "patch": patch}, commit=commit)
         return obj
 
-    def delete(self, id: int, hard: bool = False, actor = False):
-        obj = self.session.get(self.model, id)
+    def delete(self, id: int, hard: bool = False, actor=None, *, commit: bool = True):
+        session = self.session
+        obj = session.get(self.model, id)
         if not obj:
             return None
         if hard or not self.supports_soft_delete:
-            self.session.delete(obj)
+            session.delete(obj)
         else:
-            soft = cast(_SoftDeletable, obj)
-            soft.is_deleted = True
-        self.session.commit()
-        self._log_version("delete", obj, actor)
+            cast(_SoftDeletable, obj).is_deleted = True
+        if commit:
+            session.commit()
+        self._log_version("delete", obj, actor, commit=commit)
         return obj
 
-    def _log_version(self, change_type: str, obj: T, actor=None, metadata: dict = {}):
+    # ---- audit
+
+    def _log_version(self, change_type: str, obj: T, actor=None, metadata: dict | None = None, *, commit: bool = True):
+        session = self.session
         try:
-            data = obj.as_dict()
-        except Exception:
-            data = {"error": "Failed to serialize object."}
-        version = Version(
-            model_name=self.model.__name__,
-            object_id=obj.id,
-            change_type=change_type,
-            data=data,
-            actor=str(actor) if actor else None,
-            meta=metadata
-        )
-        self.session.add(version)
-        self.session.commit()
+            try:
+                snapshot = obj.as_dict()
+            except Exception:
+                snapshot = {"error": "serialize failed"}
+            version = Version(
+                model_name=self.model.__name__,
+                object_id=obj.id,
+                change_type=change_type,
+                data=to_jsonable(snapshot),
+                actor=str(actor) if actor else None,
+                meta=to_jsonable(metadata) if metadata else None,
+            )
+            session.add(version)
+            if commit:
+                session.commit()
+        except Exception as e:
+            log.warning(f"Version logging failed for {self.model.__name__} id={getattr(obj, 'id', '?')}: {str(e)}")
+            session.rollback()
diff --git a/crudkit/core/spec.py b/crudkit/core/spec.py
index 9c0e53b..bfd5f11 100644
--- a/crudkit/core/spec.py
+++ b/crudkit/core/spec.py
@@ -1,9 +1,17 @@
-from typing import List, Tuple, Set, Dict, Optional
-from sqlalchemy import asc, desc
+from dataclasses import dataclass
+from typing import Any, List, Tuple, Set, Dict, Optional, Iterable
+from sqlalchemy import and_, asc, desc, or_
 from sqlalchemy.ext.hybrid import hybrid_property
 from sqlalchemy.orm import aliased, selectinload
 from sqlalchemy.orm.attributes import InstrumentedAttribute
 
+@dataclass(frozen=True)
+class CollPred:
+    table: Any
+    col_key: str
+    op: str
+    value: Any
+
 OPERATORS = {
     'eq': lambda col, val: col == val,
     'lt': lambda col, val: col < val,
@@ -12,6 +20,8 @@ OPERATORS = {
     'gte': lambda col, val: col >= val,
     'ne': lambda col, val: col != val,
     'icontains': lambda col, val: col.ilike(f"%{val}%"),
+    'in': lambda col, val: col.in_(val if isinstance(val, (list, tuple, set)) else [val]),
+    'nin': lambda col, val: ~col.in_(val if isinstance(val, (list, tuple, set)) else [val]),
 }
 
 class CRUDSpec:
@@ -20,12 +30,138 @@ class CRUDSpec:
         self.params = params
         self.root_alias = root_alias
         self.eager_paths: Set[Tuple[str, ...]] = set()
+        # (parent_alias. relationship_attr, alias_for_target)
         self.join_paths: List[Tuple[object, InstrumentedAttribute, object]] = []
         self.alias_map: Dict[Tuple[str, ...], object] = {}
         self._root_fields: List[InstrumentedAttribute] = []
-        self._rel_field_names: Dict[Tuple[str, ...], object] = {}
+        # dotted non-collection fields (MANYTOONE etc)
+        self._rel_field_names: Dict[Tuple[str, ...], List[str]] = {}
+        # dotted collection fields (ONETOMANY)
+        self._collection_field_names: Dict[str, List[str]] = {}
         self.include_paths: Set[Tuple[str, ...]] = set()
 
+    def _split_path_and_op(self, key: str) -> tuple[str, str]:
+        if '__' in key:
+            path, op = key.rsplit('__', 1)
+        else:
+            path, op = key, 'eq'
+        return path, op
+
+    def _resolve_many_columns(self, path: str) -> list[tuple[InstrumentedAttribute, Optional[tuple[str, ...]]]]:
+        """
+        Accepts pipe-delimited paths like 'label|owner.label'
+        Returns a list of (column, join_path) pairs for every resolvable subpath.
+        """
+        cols: list[tuple[InstrumentedAttribute, Optional[tuple[str, ...]]]] = []
+        for sub in path.split('|'):
+            sub = sub.strip()
+            if not sub:
+                continue
+            col, join_path = self._resolve_column(sub)
+            if col is not None:
+                cols.append((col, join_path))
+        return cols
+
+    def _build_predicate_for(self, path: str, op: str, value: Any):
+        """
+        Builds a SQLA BooleanClauseList or BinaryExpression for a single key.
+        If multiple subpaths are provided via pipe, returns an OR of them.
+        """
+        if op not in OPERATORS:
+            return None
+
+        pairs = self._resolve_many_columns(path)
+        if not pairs:
+            return None
+
+        exprs = []
+        for col, join_path in pairs:
+            if join_path:
+                self.eager_paths.add(join_path)
+
+                try:
+                    cur_cls = self.model
+                    names = list(join_path)
+                    last_name = names[-1]
+                    is_collection = False
+                    for nm in names:
+                        rel_attr = getattr(cur_cls, nm)
+                        prop = rel_attr.property
+                        cur_cls = prop.mapper.class_
+                    is_collection = bool(getattr(getattr(self.model, last_name), "property", None)
+                                         and getattr(getattr(self.model, last_name).property, "uselist", False))
+                except Exception:
+                    is_collection = False
+
+                if is_collection:
+                    target_cls = cur_cls
+                    key = getattr(col, "key", None) or getattr(col, "name", None)
+                    if key and hasattr(target_cls, key):
+                        target_tbl = getattr(target_cls, "__table__", None)
+                        if target_tbl is not None:
+                            exprs.append(CollPred(table=target_tbl, col_key=key, op=op, value=value))
+                            continue
+
+            exprs.append(OPERATORS[op](col, value))
+
+        if not exprs:
+            return None
+
+        # If any CollPred is in exprs, do NOT or_ them. Keep it single for now.
+        if any(isinstance(x, CollPred) for x in exprs):
+            # If someone used a pipe 'relA.col|relB.col' that produced multiple CollPreds,
+            # keep the first or raise for now (your choice).
+            if len(exprs) > 1:
+                # raise NotImplementedError("OR across collection paths not supported yet")
+                exprs = [next(x for x in exprs if isinstance(x, CollPred))]
+            return exprs[0]
+
+        # Otherwise, standard SQLA clause(s)
+        return exprs[0] if len(exprs) == 1 else or_(*exprs)
+
+    def _collect_filters(self, params: dict) -> list:
+        """
+        Recursively parse filters from 'param' into a flat list of SQLA expressions.
+        Supports $or / $and groups. Any other keys are parsed as normal filters.
+        """
+        filters: list = []
+
+        for key, value in (params or {}).items():
+            if key in ('sort', 'limit', 'offset', 'fields', 'include'):
+                continue
+
+            if key == '$or':
+                # value should be a list of dicts
+                groups = []
+                for group in value if isinstance(value, (list, tuple)) else []:
+                    sub = self._collect_filters(group)
+                    if not sub:
+                        continue
+                    groups.append(and_(*sub) if len(sub) > 1 else sub[0])
+                if groups:
+                    filters.append(or_(*groups))
+                continue
+
+            if key == '$and':
+                # value should be a list of dicts
+                parts = []
+                for group in value if isinstance(value, (list, tuple)) else []:
+                    sub = self._collect_filters(group)
+                    if not sub:
+                        continue
+                    parts.append(and_(*sub) if len(sub) > 1 else sub[0])
+                if parts:
+                    filters.append(and_(*parts))
+                continue
+
+            # Normal key
+            path, op = self._split_path_and_op(key)
+            pred = self._build_predicate_for(path, op, value)
+            if pred is not None:
+                filters.append(pred)
+
+        return filters
+
     def _resolve_column(self, path: str):
         current_alias = self.root_alias
         parts = path.split('.')
@@ -68,24 +204,12 @@ class CRUDSpec:
                 if maybe:
                     self.eager_paths.add(maybe)
 
-    def parse_filters(self):
-        filters = []
-        for key, value in self.params.items():
-            if key in ('sort', 'limit', 'offset'):
-                continue
-            if '__' in key:
-                path_op = key.rsplit('__', 1)
-                if len(path_op) != 2:
-                    continue
-                path, op = path_op
-            else:
-                path, op = key, 'eq'
-            col, join_path = self._resolve_column(path)
-            if col and op in OPERATORS:
-                filters.append(OPERATORS[op](col, value))
-                if join_path:
-                    self.eager_paths.add(join_path)
-        return filters
+    def parse_filters(self, params: dict | None = None):
+        """
+        Public entry: parse filters from given params or self.params.
+        Returns a list of SQLAlchemy filter expressions
+        """
+        return self._collect_filters(params if params is not None else self.params)
 
     def parse_sort(self):
         sort_args = self.params.get('sort', '')
@@ -117,11 +241,12 @@ class CRUDSpec:
         Parse ?fields=colA,colB,rel1.colC,rel1.rel2.colD
             - Root fields become InstrumentedAttributes bound to root_alias.
             - Related fields store attribute NAMES; we'll resolve them on the target class when building loader options.
-        Returns (root_fields, rel_field_names).
+            - Collection (uselist=True) relationships record child names by relationship key.
+        Returns (root_fields, rel_field_names, root_field_names, collection_field_names_by_rel).
         """
         raw = self.params.get('fields')
         if not raw:
-            return [], {}, {}
+            return [], {}, {}, {}
 
         if isinstance(raw, list):
             tokens = []
@@ -133,14 +258,36 @@ class CRUDSpec:
         root_fields: List[InstrumentedAttribute] = []
         root_field_names: list[str] = []
         rel_field_names: Dict[Tuple[str, ...], List[str]] = {}
+        collection_field_names: Dict[str, List[str]] = {}
 
         for token in tokens:
             col, join_path = self._resolve_column(token)
             if not col:
                 continue
             if join_path:
-                rel_field_names.setdefault(join_path, []).append(col.key)
-                self.eager_paths.add(join_path)
+                # rel_field_names.setdefault(join_path, []).append(col.key)
+                # self.eager_paths.add(join_path)
+                try:
+                    cur_cls = self.model
+                    names = list(join_path)
+                    last_name = names[-1]
+                    for nm in names:
+                        rel_attr = getattr(cur_cls, nm)
+                        cur_cls = rel_attr.property.mapper.class_
+                    is_collection = bool(getattr(getattr(self.model, last_name), "property", None) and getattr(getattr(self.model, last_name).property, "uselist", False))
+                except Exception:
+                    # Fallback: inspect the InstrumentedAttribute we recorded on join_paths
+                    is_collection = False
+                    for _pa, rel_attr, _al in self.join_paths:
+                        if rel_attr.key == (join_path[-1] if join_path else ""):
+                            is_collection = bool(getattr(getattr(rel_attr, "property", None), "uselist", False))
+                            break
+
+                if is_collection:
+                    collection_field_names.setdefault(join_path[-1], []).append(col.key)
+                else:
+                    rel_field_names.setdefault(join_path, []).append(col.key)
+                    self.eager_paths.add(join_path)
             else:
                 root_fields.append(col)
                 root_field_names.append(getattr(col, "key", token))
@@ -153,7 +300,11 @@ class CRUDSpec:
 
         self._root_fields = root_fields
         self._rel_field_names = rel_field_names
-        return root_fields, rel_field_names, root_field_names
+        # return root_fields, rel_field_names, root_field_names
+        for r, names in collection_field_names.items():
+            seen3 = set()
+            collection_field_names[r] = [n for n in names if not (n in seen3 or seen3.add(n))]
+        return root_field_names, rel_field_names, root_field_names, collection_field_names
 
     def get_eager_loads(self, root_alias, *, fields_map=None):
         loads = []
diff --git a/crudkit/core/utils.py b/crudkit/core/utils.py
new file mode 100644
index 0000000..d03a7d6
--- /dev/null
+++ b/crudkit/core/utils.py
@@ -0,0 +1,176 @@
+from __future__ import annotations
+from datetime import datetime, date
+from decimal import Decimal
+from enum import Enum
+from typing import Any, Dict, Optional, Callable
+from sqlalchemy import inspect
+
+ISO_DT_FORMATS = ("%Y-%m-%dT%H:%M:%S.%f",
+                  "%Y-%m-%dT%H:%M:%S",
+                  "%Y-%m-%d %H:%M",
+                  "%Y-%m-%d")
+
+def to_jsonable(obj: Any):
+    """Recursively convert values into JSON-serializable forms."""
+    if obj is None or isinstance(obj, (str, int, float, bool)):
+        return obj
+
+    if isinstance(obj, (datetime, date)):
+        return obj.isoformat()
+
+    if isinstance(obj, Decimal):
+        return float(obj)
+
+    if isinstance(obj, Enum):
+        return obj.value
+
+    if isinstance(obj, dict):
+        return {str(k): to_jsonable(v) for k, v in obj.items()}
+
+    if isinstance(obj, (list, tuple, set)):
+        return [to_jsonable(v) for v in obj]
+
+    # fallback: strin-ify weird objects (UUID, ORM instances, etc.)
+    try:
+        return str(obj)
+    except Exception:
+        return None
+
+def filter_to_columns(data: dict, model_cls):
+    cols = {c.key for c in inspect(model_cls).mapper.columns}
+    return {k: v for k, v in data.items() if k in cols}
+
+def _parse_dt_maybe(x: Any) -> Any:
+    if isinstance(x, (datetime, date)):
+        return x
+    if isinstance(x, str):
+        s = x.strip().replace("Z", "+00:00")  # tolerate Zulu
+        for fmt in ISO_DT_FORMATS:
+            try:
+                return datetime.strptime(s, fmt)
+            except ValueError:
+                pass
+        try:
+            return datetime.fromisoformat(s)
+        except Exception:
+            return x
+    return x
+
+def _normalize_for_compare(x: Any) -> Any:
+    if isinstance(x, (str, datetime, date)):
+        return _parse_dt_maybe(x)
+    return x
+
+def deep_diff(
+    old: Any,
+    new: Any,
+    *,
+    path: str = "",
+    ignore_keys: Optional[set] = None,
+    list_mode: str = "index",  # "index" or "set"
+    custom_equal: Optional[Callable[[str, Any, Any], bool]] = None,
+) -> Dict[str, Dict[str, Any]]:
+    if ignore_keys is None:
+        ignore_keys = set()
+
+    out: Dict[str, Dict[str, Any]] = {"added": {}, "removed": {}, "changed": {}}
+
+    def mark_changed(p, a, b):
+        out["changed"][p] = {"from": a, "to": b}
+
+    def rec(o, n, pfx):
+        if custom_equal and custom_equal(pfx.rstrip("."), o, n):
+            return
+
+        if isinstance(o, dict) and isinstance(n, dict):
+            o_keys = set(o.keys())
+            n_keys = set(n.keys())
+
+            for k in sorted(o_keys - n_keys):
+                if k not in ignore_keys:
+                    out["removed"][f"{pfx}{k}"] = o[k]
+
+            for k in sorted(n_keys - o_keys):
+                if k not in ignore_keys:
+                    out["added"][f"{pfx}{k}"] = n[k]
+
+            for k in sorted(o_keys & n_keys):
+                if k not in ignore_keys:
+                    rec(o[k], n[k], f"{pfx}{k}.")
+            return
+
+        if isinstance(o, list) and isinstance(n, list):
+            if list_mode == "set":
+                if set(o) != set(n):
+                    mark_changed(pfx.rstrip("."), o, n)
+            else:
+                max_len = max(len(o), len(n))
+                for i in range(max_len):
+                    key = f"{pfx}[{i}]"
+                    if i >= len(o):
+                        out["added"][key] = n[i]
+                    elif i >= len(n):
+                        out["removed"][key] = o[i]
+                    else:
+                        rec(o[i], n[i], f"{key}.")
+            return
+
+        a = _normalize_for_compare(o)
+        b = _normalize_for_compare(n)
+        if a != b:
+            mark_changed(pfx.rstrip("."), o, n)
+
+    rec(old, new, path)
+    return out
+
+def diff_to_patch(diff: Dict[str, Dict[str, Any]]) -> Dict[str, Any]:
+    """
+    Produce a shallow patch of changed/added top-level fields.
+    Only includes leaf paths without dots/brackets; useful for simple UPDATEs.
+    """
+    patch: Dict[str, Any] = {}
+    for k, v in diff["added"].items():
+        if "." not in k and "[" not in k:
+            patch[k] = v
+    for k, v in diff["changed"].items():
+        if "." not in k and "[" not in k:
+            patch[k] = v["to"]
+    return patch
+
+def normalize_payload(payload: dict, model):
+    """
+    Coerce incoming JSON into SQLAlchemy column types for the given model.
+    - "" or None -> None
+    - Integer/Boolean/Date/DateTime handled by column type
+    """
+    from sqlalchemy import Integer, Boolean, DateTime, Date
+    out: Dict[str, Any] = {}
+
+    mapper = inspect(model).mapper
+    cols = {c.key: c.type for c in mapper.columns}
+
+    for field, value in payload.items():
+        if value == "" or value is None:
+            out[field] = None
+            continue
+
+        coltype = cols.get(field)
+        if coltype is None:
+            out[field] = value
+            continue
+
+        tname = coltype.__class__.__name__.lower()
+
+        if "integer" in tname:
+            out[field] = int(value)
+        elif "boolean" in tname:
+            out[field] = value if isinstance(value, bool) else str(value).lower() in ("1", "true", "yes", "on")
+        elif "datetime" in tname:
+            out[field] = value if isinstance(value, datetime) else _parse_dt_maybe(value)
+        elif "date" in tname:
+            v = _parse_dt_maybe(value)
+            out[field] = v.date() if isinstance(v, datetime) else v
+        else:
+            out[field] = value
+
+    return out
diff --git a/crudkit/engines.py b/crudkit/engines.py
index b420a8d..4e18fd5 100644
--- a/crudkit/engines.py
+++ b/crudkit/engines.py
@@ -1,7 +1,8 @@
+# engines.py
 from __future__ import annotations
 from typing import Type, Optional
-from sqlalchemy import create_engine
-from sqlalchemy.orm import sessionmaker
+from sqlalchemy import create_engine, event
+from sqlalchemy.orm import sessionmaker, raiseload, Mapper, RelationshipProperty
 from .backend import make_backend_info, BackendInfo
 from .config import Config, get_config
 from ._sqlite import apply_sqlite_pragmas
@@ -12,15 +13,31 @@ def build_engine(config_cls: Type[Config] | None = None):
     apply_sqlite_pragmas(engine, config_cls.SQLITE_PRAGMAS)
     return engine
 
+def _install_nplus1_guards(SessionMaker, *, strict: bool):
+    if not strict:
+        return
+
+    @event.listens_for(SessionMaker, "do_orm_execute")
+    def _add_global_raiseload(execute_state):
+        stmt = execute_state.statement
+        # Only touch ORM statements (have column_descriptions)
+        if getattr(stmt, "column_descriptions", None):
+            execute_state.statement = stmt.options(raiseload("*"))
+
 def build_sessionmaker(config_cls: Type[Config] | None = None, engine=None):
     config_cls = config_cls or get_config(None)
     engine = engine or build_engine(config_cls)
-    return sessionmaker(bind=engine, **config_cls.session_kwargs())
+    SessionMaker = sessionmaker(bind=engine, **config_cls.session_kwargs())
+
+    # Toggle with a config flag; default off so you can turn it on when ready
+    strict = bool(getattr(config_cls, "STRICT_NPLUS1", False))
+    _install_nplus1_guards(SessionMaker, strict=strict)
+    return SessionMaker
 
 class CRUDKitRuntime:
     """
     Lightweight container so CRUDKit can be given either:
-        - prebuild engine/sessionmaker, or
+        - prebuilt engine/sessionmaker, or
         - a Config to build them lazily
     """
     def __init__(self, *, engine=None, session_factory=None, config: Optional[Type[Config]] = None):
diff --git a/crudkit/integrations/flask.py b/crudkit/integrations/flask.py
index 3cc319b..feb262b 100644
--- a/crudkit/integrations/flask.py
+++ b/crudkit/integrations/flask.py
@@ -1,20 +1,32 @@
+# crudkit/integrations/flask.py
 from __future__ import annotations
 from flask import Flask
-from sqlalchemy.orm import scoped_session
+from sqlalchemy.orm import scoped_session, sessionmaker
 from ..engines import CRUDKitRuntime
 from ..config import Config
 
-def init_app(app: Flask, *, runtime: CRUDKitRuntime | None = None, config: type[Config] | None == None):
+def init_app(app: Flask, *, runtime: CRUDKitRuntime | None = None, config: type[Config] | None = None):
     """
-    Initializes CRUDKit for a Flask app. Provies `app.extensions['crudkit']`
+    Initializes CRUDKit for a Flask app. Provides `app.extensions['crudkit']`
     with a runtime (engine + session_factory). Caller manages session lifecycle.
     """
     runtime = runtime or CRUDKitRuntime(config=config)
     app.extensions.setdefault("crudkit", {})
     app.extensions["crudkit"]["runtime"] = runtime
 
-    Session = runtime.session_factory
-    if Session is not None:
-        app.extensions["crudkit"]["Session"] = scoped_session(Session)
+    # Build ONE sessionmaker bound to the ONE true engine object
+    # so engine id == sessionmaker.bind id, always.
+    engine = runtime.engine
+    SessionFactory = runtime.session_factory or sessionmaker(bind=engine, **runtime._config.session_kwargs())
+    app.extensions["crudkit"]["SessionFactory"] = SessionFactory
+    app.extensions["crudkit"]["Session"] = scoped_session(SessionFactory)
+
+    # Attach pool listeners to the *same* engine the SessionFactory is bound to.
+    # Don’t guess. Don’t hope. Inspect.
+    try:
+        bound_engine = getattr(SessionFactory, "bind", None) or getattr(SessionFactory, "kw", {}).get("bind") or engine
+        pool = bound_engine.pool
+    except Exception as e:
+        print(f"[crudkit.init_app] Failed to attach pool listeners: {e}")
 
     return runtime
diff --git a/crudkit/projection.py b/crudkit/projection.py
new file mode 100644
index 0000000..b34091d
--- /dev/null
+++ b/crudkit/projection.py
@@ -0,0 +1,236 @@
+# crudkit/projection.py
+from __future__ import annotations
+from typing import Iterable, List, Tuple, Dict, Set
+from sqlalchemy.orm import selectinload
+from sqlalchemy.orm.attributes import InstrumentedAttribute
+from sqlalchemy.orm.properties import ColumnProperty, RelationshipProperty
+from sqlalchemy import inspect
+
+# ----------------------
+# small utilities
+# ----------------------
+
+def _is_column_attr(a) -> bool:
+    try:
+        return isinstance(a, InstrumentedAttribute) and isinstance(a.property, ColumnProperty)
+    except Exception:
+        return False
+
+def _is_relationship_attr(a) -> bool:
+    try:
+        return isinstance(a, InstrumentedAttribute) and isinstance(a.property, RelationshipProperty)
+    except Exception:
+        return False
+
+def _split_path(field: str) -> List[str]:
+    return [p for p in str(field).split(".") if p]
+
+def _model_requires_map(model_cls) -> Dict[str, List[str]]:
+    # apps declare per-model deps, e.g. {"label": ["first_name","last_name","title"]}
+    return getattr(model_cls, "__crudkit_field_requires__", {}) or {}
+
+def _relationships_of(model_cls) -> Dict[str, RelationshipProperty]:
+    try:
+        return dict(model_cls.__mapper__.relationships)
+    except Exception:
+        return {}
+
+def _attr_on(model_cls, name: str):
+    return getattr(model_cls, name, None)
+
+# ----------------------
+# EXPAND: add required deps for leaf attributes at the correct class
+# ----------------------
+
+def _expand_requires_for_field(model_cls, pieces: List[str]) -> List[str]:
+    """
+    Given a dotted path like ["owner","label"], walk relationships to the leaf *container* class,
+    pull its __crudkit_field_requires__ for that leaf attr ("label"), and yield prefixed deps:
+      owner.label -> ["owner.first_name", "owner.last_name", ...] if User requires so.
+    If leaf is a column (or has no requires), returns [].
+    """
+    if not pieces:
+        return []
+
+    # walk relationships to the leaf container (class that owns the leaf attr)
+    container_cls = model_cls
+    prefix_parts: List[str] = []
+    for part in pieces[:-1]:
+        a = _attr_on(container_cls, part)
+        if not _is_relationship_attr(a):
+            return []  # can't descend; invalid or scalar in the middle
+        container_cls = a.property.mapper.class_
+        prefix_parts.append(part)
+
+    leaf = pieces[-1]
+    requires = _model_requires_map(container_cls).get(leaf) or []
+    if not requires:
+        return []
+
+    prefix = ".".join(prefix_parts)
+    out: List[str] = []
+    for dep in requires:
+        # dep may itself be dotted relative to container (e.g. "room_function.description")
+        if prefix:
+            out.append(f"{prefix}.{dep}")
+        else:
+            out.append(dep)
+    return out
+
+def _expand_requires(model_cls, fields: Iterable[str]) -> List[str]:
+    """
+    Dedup + stable expansion of requires for all fields.
+    """
+    seen: Set[str] = set()
+    out: List[str] = []
+
+    def add(f: str):
+        if f not in seen:
+            seen.add(f)
+            out.append(f)
+
+    # first pass: add original
+    queue: List[str] = []
+    for f in fields:
+        f = str(f)
+        if f not in seen:
+            seen.add(f)
+            out.append(f)
+            queue.append(f)
+
+    # BFS-ish: when we add deps, they may trigger further deps downstream
+    while queue:
+        f = queue.pop(0)
+        deps = _expand_requires_for_field(model_cls, _split_path(f))
+        for d in deps:
+            if d not in seen:
+                seen.add(d)
+                out.append(d)
+                queue.append(d)
+
+    return out
+
+# ----------------------
+# BUILD loader options tree with selectinload + load_only on real columns
+# ----------------------
+
+def _insert_leaf(loader_tree: dict, path: List[str]):
+    """
+    Build nested dict structure keyed by relationship names.
+    Each node holds:
+      {
+        "__cols__": set(column_names_to_load_only),
+        "<child_rel>": { ... }
+      }
+    """
+    node = loader_tree
+    for rel in path[:-1]:  # only relationship hops
+        node = node.setdefault(rel, {"__cols__": set()})
+    # leaf may be a column or a virtual/hybrid; only columns go to __cols__
+    node.setdefault("__cols__", set())
+
+def _attach_column(loader_tree: dict, path: List[str], model_cls):
+    """
+    If the leaf is a real column on the target class, record its name into __cols__ at that level.
+    """
+    # descend to target class to test column-ness
+    container_cls = model_cls
+    node = loader_tree
+    for rel in path[:-1]:
+        a = _attr_on(container_cls, rel)
+        if not _is_relationship_attr(a):
+            return  # invalid path, ignore
+        container_cls = a.property.mapper.class_
+        node = node.setdefault(rel, {"__cols__": set()})
+
+    leaf = path[-1]
+    a_leaf = _attr_on(container_cls, leaf)
+    node.setdefault("__cols__", set())
+    if _is_column_attr(a_leaf):
+        node["__cols__"].add(leaf)
+
+def _build_loader_tree(model_cls, fields: Iterable[str]) -> dict:
+    """
+    For each dotted field:
+      - walk relationships -> create nodes
+      - if leaf is a column: record it for load_only
+      - if leaf is not a column (hybrid/descriptor): no load_only; still ensure rel hops exist
+    """
+    tree: Dict[str, dict] = {"__cols__": set()}
+    for f in fields:
+        parts = _split_path(f)
+        if not parts:
+            continue
+        # ensure relationship nodes exist
+        _insert_leaf(tree, parts)
+        # attach column if applicable
+        _attach_column(tree, parts, model_cls)
+    return tree
+
+def _loader_options_from_tree(model_cls, tree: dict):
+    """
+    Convert the loader tree into SQLAlchemy loader options:
+      selectinload(<rel>)[.load_only(cols)] recursively
+    """
+    opts = []
+
+    rels = _relationships_of(model_cls)
+    for rel_name, child in tree.items():
+        if rel_name == "__cols__":
+            continue
+        rel_prop = rels.get(rel_name)
+        if not rel_prop:
+            continue
+        rel_attr = getattr(model_cls, rel_name)
+        opt = selectinload(rel_attr)
+
+        # apply load_only on the related class (only real columns recorded at child["__cols__"])
+        cols = list(child.get("__cols__", []))
+        if cols:
+            rel_model = rel_prop.mapper.class_
+            # map column names to attributes
+            col_attrs = []
+            for c in cols:
+                a = getattr(rel_model, c, None)
+                if _is_column_attr(a):
+                    col_attrs.append(a)
+            if col_attrs:
+                opt = opt.load_only(*col_attrs)
+
+        # recurse to grandchildren
+        sub_opts = _loader_options_from_tree(rel_prop.mapper.class_, child)
+        for so in sub_opts:
+            opt = opt.options(so)
+
+        opts.append(opt)
+
+    # root-level columns (rare in our compile; kept for completeness)
+    root_cols = list(tree.get("__cols__", []))
+    if root_cols:
+        # NOTE: call-site can add a root load_only(...) if desired;
+        # we purposely return only relationship options here to keep
+        # the API simple and avoid mixing Load(model_cls) contexts.
+        pass
+
+    return opts
+
+# ----------------------
+# PUBLIC API
+# ----------------------
+
+def compile_projection(model_cls, fields: Iterable[str]) -> Tuple[List[str], List]:
+    """
+    Returns:
+      expanded_fields: List[str]            # original + declared dependencies
+      loader_options: List[Load]            # apply via query = query.options(*loader_options)
+
+    Behavior:
+      - Expands __crudkit_field_requires__ at the leaf container class for every field.
+      - Builds a selectinload tree; load_only only includes real columns (no hybrids).
+      - Safe for nested paths: e.g. "owner.label" pulls owner deps from User.__crudkit_field_requires__.
+    """
+    fields = list(fields or [])
+    expanded = _expand_requires(model_cls, fields)
+    tree = _build_loader_tree(model_cls, expanded)
+    options = _loader_options_from_tree(model_cls, tree)
+    return expanded, options
diff --git a/crudkit/ui/fragments.py b/crudkit/ui/fragments.py
index 1f74f3a..03cc1c0 100644
--- a/crudkit/ui/fragments.py
+++ b/crudkit/ui/fragments.py
@@ -6,25 +6,279 @@ from flask import current_app, url_for
 from jinja2 import Environment, FileSystemLoader, ChoiceLoader
 from sqlalchemy import inspect
 from sqlalchemy.orm import Load, RelationshipProperty, class_mapper, load_only, selectinload
+from sqlalchemy.orm.attributes import InstrumentedAttribute
 from sqlalchemy.orm.base import NO_VALUE
+from sqlalchemy.orm.properties import ColumnProperty, RelationshipProperty
 from typing import Any, Dict, List, Optional, Tuple
 
+import crudkit
+
 _ALLOWED_ATTRS = {
     "class", "placeholder", "autocomplete", "inputmode", "pattern",
     "min", "max", "step", "maxlength", "minlength",
     "required", "readonly", "disabled",
-    "multiple", "size",
+    "multiple", "size", "rows",
     "id", "name", "value",
 }
 
+_SAFE_CSS_PROPS = {
+    # spacing / sizing
+    "margin","margin-top","margin-right","margin-bottom","margin-left",
+    "padding","padding-top","padding-right","padding-bottom","padding-left",
+    "width","height","min-width","min-height","max-width","max-height", "resize",
+    # layout
+    "display","flex","flex-direction","flex-wrap","justify-content","align-items","gap",
+    # text
+    "font-size","font-weight","line-height","text-align","white-space",
+    # colors / background
+    "color","background-color",
+    # borders / radius
+    "border","border-top","border-right","border-bottom","border-left",
+    "border-width","border-style","border-color","border-radius",
+    # misc (safe-ish)
+    "opacity","overflow","overflow-x","overflow-y",
+}
+
+_num_unit = r"-?\d+(?:\.\d+)?"
+_len_unit = r"(?:px|em|rem|%)"
+P_LEN     = re.compile(rf"^{_num_unit}(?:{_len_unit})?$")              # 12, 12px, 1.2rem, 50%
+P_GAP     = P_LEN
+P_INT     = re.compile(r"^\d+$")
+P_COLOR   = re.compile(
+    r"^(#[0-9a-fA-F]{3,8}|"
+    r"rgb\(\s*\d{1,3}\s*,\s*\d{1,3}\s*,\s*\d{1,3}\s*\)|"
+    r"rgba\(\s*\d{1,3}\s*,\s*\d{1,3}\s*,\s*\d{1,3}\s*,\s*(?:0|1|0?\.\d+)\s*\)|"
+    r"[a-zA-Z]+)$"
+)
+
+_ENUMS = {
+    "display": {"block","inline","inline-block","flex","grid","none"},
+    "flex-direction": {"row","row-reverse","column","column-reverse"},
+    "flex-wrap": {"nowrap","wrap","wrap-reverse"},
+    "justify-content": {"flex-start","flex-end","center","space-between","space-around","space-evenly"},
+    "align-items": {"stretch","flex-start","flex-end","center","baseline"},
+    "text-align": {"left","right","center","justify","start","end"},
+    "white-space": {"normal","nowrap","pre","pre-wrap","pre-line","break-spaces"},
+    "border-style": {"none","solid","dashed","dotted","double","groove","ridge","inset","outset"},
+    "overflow": {"visible","hidden","scroll","auto","clip"},
+    "overflow-x": {"visible","hidden","scroll","auto","clip"},
+    "overflow-y": {"visible","hidden","scroll","auto","clip"},
+    "font-weight": {"normal","bold","bolder","lighter","100","200","300","400","500","600","700","800","900"},
+    "resize": {"none", "both", "horizontal", "vertical"},
+}
+
 def get_env():
+    """
+    Return an overlay Jinja Environment that knows how to load crudkit templates
+    and has our helper functions available as globals.
+    """
     app = current_app
     default_path = os.path.join(os.path.dirname(__file__), 'templates')
     fallback_loader = FileSystemLoader(default_path)
 
-    return app.jinja_env.overlay(
-        loader=ChoiceLoader([app.jinja_loader, fallback_loader])
-    )
+    env = app.jinja_env.overlay(loader=ChoiceLoader([app.jinja_loader, fallback_loader]))
+    # Ensure helpers are available even when we render via this overlay env.
+    # These names are resolved at *call time* (not at def time), so it's safe.
+    try:
+        env.globals.setdefault("render_table", render_table)
+        env.globals.setdefault("render_form", render_form)
+        env.globals.setdefault("render_field", render_field)
+    except NameError:
+        # Functions may not be defined yet at import time; later calls will set them.
+        pass
+
+    return env
+
+def register_template_globals(app=None):
+    """
+    Register crudkit helpers as app-wide Jinja globals so they can be used
+    directly in any template via {{ render_table(...) }}, {{ render_form(...) }},
+    and {{ render_field(...) }}.
+    """
+    if app is None:
+        app = current_app
+    # Idempotent install using an extension flag
+    installed = app.extensions.setdefault("crudkit_ui_helpers", set())
+    to_register = {
+        "render_table": render_table,
+        "render_form": render_form,
+        "render_field": render_field,
+    }
+    for name, fn in to_register.items():
+        if name not in installed:
+            app.add_template_global(fn, name)
+            installed.add(name)
+
+def _fields_for_label_params(label_spec, related_model):
+    """
+    Build a 'fields' list suitable for CRUDService.list() so labels render
+    without triggering lazy loads. Always includes 'id'.
+    """
+    simple_cols, rel_paths = _extract_label_requirements(label_spec, related_model)
+    fields = set(["id"])
+    for c in simple_cols:
+        fields.add(c)
+    for rel_name, col_name in rel_paths:
+        if col_name == "__all__":
+            # just ensure relationship object is present; ask for rel.id
+            fields.add(f"{rel_name}.id")
+        else:
+            fields.add(f"{rel_name}.{col_name}")
+    return list(fields)
+
+def _fk_options_via_service(related_model, label_spec, *, options_params: dict | None = None):
+    svc = crudkit.crud.get_service(related_model)
+
+    # default to unlimited results for dropdowns
+    params = {"limit": 0}
+    if options_params:
+        params.update(options_params)  # caller can override limit if needed
+
+    # ensure fields needed to render the label are present (avoid lazy loads)
+    fields = _fields_for_label_params(label_spec, related_model)
+    if fields:
+        existing = params.get("fields")
+        if isinstance(existing, str):
+            existing = [s.strip() for s in existing.split(",") if s.strip()]
+        if isinstance(existing, (list, tuple)):
+            params["fields"] = list(dict.fromkeys(list(existing) + fields))
+        else:
+            params["fields"] = fields
+
+    # only set a default sort if caller didn’t supply one
+    if "sort" not in params:
+        simple_cols, _ = _extract_label_requirements(label_spec, related_model)
+        params["sort"] = (simple_cols[0] if simple_cols else "id")
+
+    rows = svc.list(params)
+    return [
+        {"value": str(r.id), "label": _label_from_obj(r, label_spec)}
+        for r in rows
+    ]
+
+def expand_projection(model_cls, fields):
+    req = getattr(model_cls, "__crudkit_field_requires__", {}) or {}
+    out = set(fields)
+    for f in list(fields):
+        for dep in req.get(f, ()):
+            out.add(dep)
+    return list(out)
+
+def _clean_css_value(prop: str, raw: str) -> str | None:
+    v = raw.strip()
+
+    v = v.replace("!important", "")
+    low = v.lower()
+    if any(bad in low for bad in ("url(", "expression(", "javascript:", "var(")):
+        return None
+
+    if prop in {"width","height","min-width","min-height","max-width","max-height",
+                "margin","margin-top","margin-right","margin-bottom","margin-left",
+                "padding","padding-top","padding-right","padding-bottom","padding-left",
+                "border-width","border-top","border-right","border-bottom","border-left","border-radius",
+                "line-height","font-size"}:
+        return v if P_LEN.match(v) else None
+
+    if prop in {"gap"}:
+        parts = [p.strip() for p in v.split()]
+        if 1 <= len(parts) <= 2 and all(P_GAP.match(p) for p in parts):
+            return " ".join(parts)
+        return None
+
+    if prop in {"color", "background-color", "border-color"}:
+        return v if P_COLOR.match(v) else None
+
+    if prop in _ENUMS:
+        return v if v.lower() in _ENUMS[prop] else None
+
+    if prop == "flex":
+        toks = v.split()
+        if len(toks) == 1 and (toks[0].isdigit() or toks[0] in {"auto", "none"}):
+            return v
+        if len(toks) == 2 and toks[0].isdigit() and (toks[1].isdigit() or toks[1] == "auto"):
+            return v
+        if len(toks) == 3 and toks[0].isdigit() and toks[1].isdigit() and (P_LEN.match(toks[2]) or toks[2] == "auto"):
+            return " ".join(toks)
+        return None
+
+    if prop == "border":
+        parts = v.split()
+        bw = next((p for p in parts if P_LEN.match(p)), None)
+        bs = next((p for p in parts if p in _ENUMS["border-style"]), None)
+        bc = next((p for p in parts if P_COLOR.match(p)), None)
+        chosen = [x for x in (bw, bs, bc) if x]
+        return " ".join(chosen) if chosen else None
+
+    return None
+
+def _sanitize_style(style: str | None) -> str | None:
+    if not style or not isinstance(style, str):
+        return None
+    safe_decls = []
+    for chunk in style.split(";"):
+        if not chunk.strip():
+            continue
+        if ":" not in chunk:
+            continue
+        prop, val = chunk.split(":", 1)
+        prop = prop.strip().lower()
+        if prop not in _SAFE_CSS_PROPS:
+            continue
+        clean = _clean_css_value(prop, val)
+        if clean is not None and clean != "":
+            safe_decls.append(f"{prop}: {clean}")
+    return "; ".join(safe_decls) if safe_decls else None
+
+def _is_column_attr(attr) -> bool:
+    try:
+        return isinstance(attr, InstrumentedAttribute) and isinstance(attr.property, ColumnProperty)
+    except Exception:
+        return False
+
+def _is_relationship_attr(attr) -> bool:
+    try:
+        return isinstance(attr, InstrumentedAttribute) and isinstance(attr.property, RelationshipProperty)
+    except Exception:
+        return False
+
+def _get_attr_deps(model_cls, attr_name: str, extra_deps: Optional[dict] = None) -> list[str]:
+    """Merge model-level and per-field declared deps for a computed attr."""
+    model_deps = getattr(model_cls, "__crudkit_field_requires__", {}) or {}
+    field_deps = (extra_deps or {})
+    return list(model_deps.get(attr_name, [])) + list(field_deps.get(attr_name, []))
+
+def _get_loaded_attr(obj: Any, name: str) -> Any:
+    """
+    Return obj.<name> only if it is already loaded.
+    Never triggers a lazy load. Returns None if missing/unloaded.
+    Works for both column and relationship attributes.
+    """
+    try:
+        st = inspect(obj)
+        # 1) Mapped attribute?
+        attr = st.attrs.get(name)
+        if attr is not None:
+            val = attr.loaded_value
+            return None if val is NO_VALUE else val
+        # 2) Already present value (e.g., eager-loaded or set on the dict)?
+        if name in st.dict:
+            return st.dict.get(name)
+        # 3) If object is detached or attr is not mapped, DO NOT eval hybrids
+        #    or descriptors that could lazy-load. That would explode.
+        if st.session is None:
+            return None
+        # 4) As a last resort on attached instances only, try simple getattr,
+        #    but guard against DetachedInstanceError anyway.
+        try:
+            return getattr(obj, name, None)
+        except Exception:
+            return None
+    except Exception:
+        # If we can't even inspect it, be conservative
+        try:
+            return getattr(obj, name, None)
+        except Exception:
+            return None
 
 def _normalize_rows_layout(layout: Optional[List[dict]]) -> Dict[str, dict]:
     """
@@ -164,6 +418,11 @@ def _sanitize_attrs(attrs: Any) -> dict[str, Any]:
         elif isinstance(v, str):
             if len(v) > 512:
                 v = v[:512]
+        if k == "style":
+            sv = _sanitize_style(v)
+            if sv:
+                out["style"] = sv
+            continue
         if k.startswith("data-") or k.startswith("aria-") or k in _ALLOWED_ATTRS:
             if isinstance(v, bool):
                 if v:
@@ -173,6 +432,105 @@ def _sanitize_attrs(attrs: Any) -> dict[str, Any]:
 
     return out
 
+def _resolve_rel_obj(values: dict, instance, base: str):
+    rel = None
+    if isinstance(values, dict) and base in values:
+        rel = values[base]
+        if isinstance(rel, dict):
+            class _DictObj:
+                def __init__(self, d): self._d = d
+                def __getattr__(self, k): return self._d.get(k)
+            rel = _DictObj(rel)
+
+    if rel is None and instance is not None:
+        try:
+            st = inspect(instance)
+            ra = st.attrs.get(base)
+            if ra is not None and ra.loaded_value is not NO_VALUE:
+                rel  = ra.loaded_value
+        except Exception:
+            pass
+
+    return rel
+
+def _value_label_for_field(field: dict, mapper, values_map: dict, instance, session):
+    """
+    If field targets a MANYTOONE (foo or foo_id), compute a human-readable label.
+    No lazy loads. Optional single-row lean fetch if we only have the id.
+    """
+    base, rel_prop = _rel_for_id_name(mapper, field["name"])
+    if not rel_prop:
+        return None
+
+    rid = _coerce_fk_value(values_map, instance, base, rel_prop)
+    rel_obj = _resolve_rel_obj(values_map, instance, base)
+
+    label_spec = (
+        field.get("label_spec")
+        or getattr(rel_prop.mapper.class_, "__crud_label__", None)
+        or "id"
+    )
+
+    if rel_obj is not None and session is not None and rid is not None:
+        mdl = rel_prop.mapper.class_
+
+        # Work out exactly what the label needs (columns + rel paths),
+        # expanding model-level and per-field deps (for hybrids etc.)
+        simple_cols, rel_paths = _extract_label_requirements(
+            label_spec,
+            model_cls=mdl,
+            extra_deps=field.get("label_deps")
+        )
+
+        # If the currently-attached object doesn't have what we need, do one lean requery
+        if not _has_label_bits_loaded(rel_obj, label_spec):
+            q = session.query(mdl)
+
+            # only real columns in load_only
+            cols = []
+            id_attr = getattr(mdl, "id", None)
+            if _is_column_attr(id_attr):
+                cols.append(id_attr)
+            for c in simple_cols:
+                a = getattr(mdl, c, None)
+                if _is_column_attr(a):
+                    cols.append(a)
+            if cols:
+                q = q.options(load_only(*cols))
+
+            # selectinload relationships; "__all__" means just eager the relationship object
+            for rel_name, col_name in rel_paths:
+                rel_ia = getattr(mdl, rel_name, None)
+                if rel_ia is None:
+                    continue
+                opt = selectinload(rel_ia)
+                if col_name == "__all__":
+                    q = q.options(opt)
+                else:
+                    t_cls = mdl.__mapper__.relationships[rel_name].mapper.class_
+                    t_attr = getattr(t_cls, col_name, None)
+                    q = q.options(opt.load_only(t_attr) if _is_column_attr(t_attr) else opt)
+
+            rel_obj = q.get(rid)
+
+    if rel_obj is not None:
+        try:
+            s = _label_from_obj(rel_obj, label_spec)
+        except Exception:
+            s = None
+        # If we couldn't safely render and we have a session+id, do one lean retry.
+        if (s is None or s == "") and session is not None and rid is not None:
+            mdl = rel_prop.mapper.class_
+            try:
+                rel_obj2 = session.get(mdl, rid) # attached instance
+                s2 = _label_from_obj(rel_obj2, label_spec)
+                if s2:
+                    return s2
+            except Exception:
+                pass
+        return s
+    return str(rid) if rid is not None else None
+
 class _SafeObj:
     """Attribute access that returns '' for missing/None instead of exploding."""
     __slots__ = ("_obj",)
@@ -181,12 +539,10 @@ class _SafeObj:
     def __getattr__(self, name):
         if self._obj is None:
             return ""
-        val = getattr(self._obj, name, None)
-        if val is None:
-            return ""
-        return _SafeObj(val)
+        val = _get_loaded_attr(self._obj, name)
+        return "" if val is None else _SafeObj(val)
 
-def _coerce_fk_value(values: dict | None, instance: Any, base: str):
+def _coerce_fk_value(values: dict | None, instance: Any, base: str, rel_prop: Optional[RelationshipProperty] = None):
     """
     Resolve current selection for relationship 'base':
       1) values['<base>_id']
@@ -233,6 +589,25 @@ def _coerce_fk_value(values: dict | None, instance: Any, base: str):
         except Exception:
             pass
 
+    # Fallback: if we know the relationship, try its local FK column names
+    if rel_prop is not None:
+        try:
+            st = inspect(instance) if instance is not None else None
+        except Exception:
+            st = None
+
+        # Try values[...] first
+        for col in getattr(rel_prop, "local_columns", []) or []:
+            key = getattr(col, "key", None) or getattr(col, "name", None)
+            if not key:
+                continue
+            if isinstance(values, dict) and key in values and values[key] not in (None, ""):
+                return values[key]
+            if set is not None:
+                attr = st.attrs.get(key) if hasattr(st, "attrs") else None
+                if attr is not None and attr.loaded_value is not NO_VALUE:
+                    return attr.loaded_value
+
     return None
 
 def _is_many_to_one(mapper, name: str) -> Optional[RelationshipProperty]:
@@ -254,43 +629,42 @@ def _rel_for_id_name(mapper, name: str) -> tuple[Optional[str], Optional[Relatio
         return (name, prop) if prop else (None, None)
 
 def _fk_options(session, related_model, label_spec):
-    simple_cols, rel_paths = _extract_label_requirements(label_spec)
+    simple_cols, rel_paths = _extract_label_requirements(label_spec, related_model)
     q = session.query(related_model)
 
     col_attrs = []
     if hasattr(related_model, "id"):
-        col_attrs.append(getattr(related_model, "id"))
+        id_attr = getattr(related_model, "id")
+        if _is_column_attr(id_attr):
+            col_attrs.append(id_attr)
+
     for name in simple_cols:
-        if hasattr(related_model, name):
-            col_attrs.append(getattr(related_model, name))
+        attr = getattr(related_model, name, None)
+        if _is_column_attr(attr):
+            col_attrs.append(attr)
     if col_attrs:
         q = q.options(load_only(*col_attrs))
 
     for rel_name, col_name in rel_paths:
-        rel_prop = getattr(related_model, rel_name, None)
-        if rel_prop is None:
+        rel_attr = getattr(related_model, rel_name, None)
+        if rel_attr is None:
             continue
-        try:
+        opt = selectinload(rel_attr)
+        if col_name == "__all__":
+            q = q.options(opt)
+        else:
             target_cls = related_model.__mapper__.relationships[rel_name].mapper.class_
             col_attr = getattr(target_cls, col_name, None)
-            if col_attr is None:
-                q = q.options(selectinload(rel_prop))
-            else:
-                q = q.options(selectinload(rel_prop).load_only(col_attr))
-        except Exception:
-            q = q.options(selectinload(rel_prop))
+            q = q.options(opt.load_only(col_attr) if _is_column_attr(col_attr) else opt)
 
     if simple_cols:
         first = simple_cols[0]
         if hasattr(related_model, first):
-            q = q.order_by(getattr(related_model, first))
+            q = q.order_by(None).order_by(getattr(related_model, first))
 
     rows = q.all()
     return [
-        {
-            'value': getattr(opt, 'id'),
-            'label': _label_from_obj(opt, label_spec),
-        }
+        {'value': getattr(opt, 'id'), 'label': _label_from_obj(opt, label_spec)}
         for opt in rows
     ]
 
@@ -314,12 +688,20 @@ def _normalize_field_spec(spec, mapper, session, label_specs_model_default):
         "template": spec.get("template"),
         "template_name": spec.get("template_name"),
         "template_ctx": spec.get("template_ctx"),
+        "label_spec": spec.get("label_spec"),
     }
 
+    if "link" in spec:
+        field["link"] = spec["link"]
+    if "label_deps" in spec:
+        field["label_deps"] = spec["label_deps"]
+
+    opts_params = spec.get("options_params") or spec.get("options_filter") or spec.get("options_where")
+
     if rel_prop:
         if field["type"] is None:
             field["type"] = "select"
-        if field["type"] == "select" and field.get("options") is None and session is not None:
+        if field["type"] == "select" and field.get("options") is None:
             related_model = rel_prop.mapper.class_
             label_spec = (
                 spec.get("label_spec")
@@ -327,7 +709,11 @@ def _normalize_field_spec(spec, mapper, session, label_specs_model_default):
                 or getattr(related_model, "__crud_label__", None)
                 or "id"
             )
-            field["options"] = _fk_options(session, related_model, label_spec)
+            field["options"] = _fk_options_via_service(
+                related_model,
+                label_spec,
+                options_params=opts_params
+            )
         return field
 
     col = mapper.columns.get(name)
@@ -347,81 +733,86 @@ def _normalize_field_spec(spec, mapper, session, label_specs_model_default):
 
     return field
 
-def _extract_label_requirements(spec: Any) -> tuple[list[str], list[tuple[str, str]]]:
+def _extract_label_requirements(
+    spec: Any,
+    model_cls: Any = None,
+    extra_deps: Optional[Dict[str, List[str]]] = None
+) -> tuple[list[str], list[tuple[str, str]]]:
     """
-    From a label spec, return:
-      - simple_cols: ["name", "code"]
-      - rel_paths: [("room_function", "description"), ("owner", "last_name")]
+    Returns:
+      simple_cols: ["name", "code", "label", ...]  (non-dotted names; may include non-columns)
+      rel_paths:   [("room_function", "description"), ("brand", "__all__"), ...]
+        - ("rel", "__all__") means: just eager the relationship (no specific column)
+    Also expands dependencies declared by the model or the field (extra_deps).
     """
     simple_cols: list[str] = []
     rel_paths: list[tuple[str, str]] = []
+    seen: set[str] = set()
 
-    def ingest(token: str) -> None:
-        token = str(token).strip()
-        if not token:
+    def add_dep_token(token: str) -> None:
+        """Add a concrete dependency token (column or 'rel' or 'rel.col')."""
+        if not token or token in seen:
             return
+        seen.add(token)
+
         if "." in token:
             rel, col = token.split(".", 1)
             if rel and col:
                 rel_paths.append((rel, col))
+            return
+
+        # bare token: could be column, relationship, or computed
+        simple_cols.append(token)
+
+        # If this is not obviously a column, try pulling declared deps.
+        if model_cls is not None:
+            attr = getattr(model_cls, token, None)
+            if _is_column_attr(attr):
+                return
+            # If it's a relationship, we want to eager the relationship itself.
+            if _is_relationship_attr(attr):
+                rel_paths.append((token, "__all__"))
+                return
+            # Not a column/relationship => computed (hybrid/descriptor/etc.)
+            for dep in _get_attr_deps(model_cls, token, extra_deps):
+                add_dep_token(dep)
+
+    def add_from_spec(piece: Any) -> None:
+        if piece is None or callable(piece):
+            return
+        if isinstance(piece, (list, tuple)):
+            for a in piece:
+                add_from_spec(a)
+            return
+        s = str(piece)
+        if "{" in s and "}" in s:
+            for n in re.findall(r"{\s*([^}:\s]+)", s):
+                add_dep_token(n)
         else:
-            simple_cols.append(token)
-
-    if spec is None or callable(spec):
-        return simple_cols, rel_paths
-
-    if isinstance(spec, (list, tuple)):
-        for a in spec:
-            ingest(a)
-        return simple_cols, rel_paths
-
-    if isinstance(spec, str):
-        # format string like "{first} {last}" or "{room_function.description} · {name}"
-        if "{" in spec and "}" in spec:
-            names = re.findall(r"{\s*([^}:\s]+)", spec)
-            for n in names:
-                ingest(n)
-        else:
-            ingest(spec)
-        return simple_cols, rel_paths
+            add_dep_token(s)
 
+    add_from_spec(spec)
     return simple_cols, rel_paths
 
-def _attrs_from_label_spec(spec: Any) -> list[str]:
-    """
-    Return a list of attribute names needed from the related model to compute the label.
-    Only simple attribute names are returned; dotted paths return just the first segment.
-    """
-    if spec is None:
-        return []
-    if callable(spec):
-        return []
-    if isinstance(spec, (list, tuple)):
-        return [str(a).split(".", 1)[0] for a in spec]
-    if isinstance(spec, str):
-        if "{" in spec and "}" in spec:
-            names = re.findall(r"{\s*([^}:\s]+)", spec)
-            return [n.split(".", 1)[0] for n in names]
-        return [spec.split(".", 1)[0]]
-    return []
 
 def _label_from_obj(obj: Any, spec: Any) -> str:
+    if obj is None:
+        return ""
+
     if spec is None:
         for attr in ("label", "name", "title", "description"):
-            if hasattr(obj, attr):
-                val = getattr(obj, attr)
-                if not callable(val) and val is not None:
-                    return str(val)
-        if hasattr(obj, "id"):
-            return str(getattr(obj, "id"))
-        return object.__repr__(obj)
+            val = _get_loaded_attr(obj, attr)
+            if val is not None:
+                return str(val)
+        vid = _get_loaded_attr(obj, "id")
+        return str(vid) if vid is not None else object.__repr__(obj)
 
     if isinstance(spec, (list, tuple)):
         parts = []
         for a in spec:
             cur = obj
             for part in str(a).split("."):
-                cur = getattr(cur, part, None)
+                cur = _get_loaded_attr(cur, part) if cur is not None else None
                 if cur is None:
                     break
             parts.append("" if cur is None else str(cur))
@@ -433,9 +824,10 @@ def _label_from_obj(obj: Any, spec: Any) -> str:
         for f in fields:
             root = f.split(".", 1)[0]
             if root not in data:
-                val = getattr(obj, root, None)
-                data[root] = _SafeObj(val)
-
+                try:
+                    data[root] = _SafeObj(_get_loaded_attr(obj, root))
+                except Exception:
+                    data[root] = _SafeObj(None)
         try:
             return spec.format(**data)
         except Exception:
@@ -443,7 +835,7 @@ def _label_from_obj(obj: Any, spec: Any) -> str:
 
     cur = obj
     for part in str(spec).split("."):
-        cur = getattr(cur, part, None)
+        cur = _get_loaded_attr(cur, part) if cur is not None else None
         if cur is None:
             return ""
     return str(cur)
@@ -560,6 +952,9 @@ def _format_value(val: Any, fmt: Optional[str]) -> Any:
     if fmt is None:
         return val
     try:
+        if callable(fmt):
+            return fmt(val)
+
         if fmt == "yesno":
             return "Yes" if bool(val) else "No"
         if fmt == "date":
@@ -572,12 +967,68 @@ def _format_value(val: Any, fmt: Optional[str]) -> Any:
         return val
     return val
 
+def _has_label_bits_loaded(obj, label_spec) -> bool:
+    try:
+        st = inspect(obj)
+    except Exception:
+        return True
+
+    simple_cols, rel_paths = _extract_label_requirements(label_spec, type(obj))
+
+    # concrete columns on the object
+    for name in simple_cols:
+        a = getattr(type(obj), name, None)
+        if _is_column_attr(a) and name not in st.dict:
+            return False
+        # non-column tokens (hybrids/descriptors) are satisfied by their deps above
+
+    # relationships
+    for rel_name, col_name in rel_paths:
+        ra = st.attrs.get(rel_name)
+        if ra is None or ra.loaded_value in (NO_VALUE, None):
+            return False
+        if col_name == "__all__":
+            continue  # relationship object present is enough
+        try:
+            t_st = inspect(ra.loaded_value)
+        except Exception:
+            return False
+        t_attr = getattr(type(ra.loaded_value), col_name, None)
+        if _is_column_attr(t_attr) and col_name not in t_st.dict:
+            return False
+
+    return True
+
 def _class_for(val: Any, classes: Optional[Dict[str, str]]) -> Optional[str]:
     if not classes:
         return None
     key = "none" if val is None else str(val).lower()
     return classes.get(key, classes.get("default"))
 
+def _format_label_from_values(spec: Any, values: dict) -> Optional[str]:
+    if not spec:
+        return None
+    if isinstance(spec, (list, tuple)):
+        parts = []
+        for a in spec:
+            v = _deep_get(values, str(a))
+            parts.append("" if v is None else str(v))
+        return " ".join(p for p in parts if p)
+
+    s = str(spec)
+    if "{" in s and "}" in s:
+        names = re.findall(r"{\s*([^}:\s]+)", s)
+        data = {n: _deep_get(values, n) for n in names}
+        # wrap for safe .format()
+        data = {k: ("" if v is None else v) for k, v in data.items()}
+        try:
+            return s.format(**data)
+        except Exception:
+            return None
+    # simple field name
+    v = _deep_get(values, s)
+    return "" if v is None else str(v)
+
 def _build_href(spec: Dict[str, Any], row: Dict[str, Any], obj) -> Optional[str]:
     if not spec:
         return None
@@ -594,8 +1045,9 @@ def _build_href(spec: Dict[str, Any], row: Dict[str, Any], obj) -> Optional[str]
     if any(v is None for v in params.values()):
         return None
     try:
-        return url_for('crudkit.' + spec["endpoint"], **params)
+        return url_for(spec["endpoint"], **params)
     except Exception as e:
+        print(f"Cannot create endpoint for {spec['endpoint']}: {str(e)}")
         return None
 
 def _humanize(field: str) -> str:
@@ -668,6 +1120,8 @@ def render_field(field, value):
         attrs=_sanitize_attrs(field.get('attrs') or {}),
         label_attrs=_sanitize_attrs(field.get('label_attrs') or {}),
         help=field.get('help'),
+        value_label=field.get('value_label'),
+        link_href=field.get("link_href"),
     )
 
 
@@ -762,7 +1216,7 @@ def render_form(
                 base = name[:-3]
                 rel_prop = mapper.relationships.get(base)
                 if isinstance(rel_prop, RelationshipProperty) and rel_prop.direction.name == "MANYTOONE":
-                    values_map[name] = _coerce_fk_value(values, instance, base)
+                    values_map[name] = _coerce_fk_value(values, instance, base, rel_prop)  # add rel_prop
 
     else:
         # Auto-generate path (your original behavior)
@@ -795,7 +1249,7 @@ def render_form(
                 fk_fields.add(f"{base}_id")
 
                 # NEW: set the current selection for this dropdown
-                values_map[f"{base}_id"] = _coerce_fk_value(values, instance, base)
+                values_map[f"{base}_id"] = _coerce_fk_value(values, instance, base, prop)
 
         # Then plain columns
         for col in model_cls.__table__.columns:
@@ -815,15 +1269,37 @@ def render_form(
                 field["wrap"] = _sanitize_attrs(field["wrap"])
             fields.append(field)
 
-        if submit_attrs:
+    if submit_attrs:
             submit_attrs = _sanitize_attrs(submit_attrs)
 
-        common_ctx = {"values": values_map, "instance": instance, "model_cls": model_cls, "session": session}
-        for f in fields:
-            if f.get("type") == "template":
-                base = dict(common_ctx)
-                base.update(f.get("template_ctx") or {})
-                f["template_ctx"] = base
+    common_ctx = {"values": values_map, "instance": instance, "model_cls": model_cls, "session": session}
+    for f in fields:
+        if f.get("type") == "template":
+            base = dict(common_ctx)
+            base.update(f.get("template_ctx") or {})
+            f["template_ctx"] = base
+
+    for f in fields:
+        # existing FK label resolution
+        vl = _value_label_for_field(f, mapper, values_map, instance, session)
+        if vl is not None:
+            f["value_label"] = vl
+        # NEW: if not a relationship but a label_spec is provided, format from values
+        elif f.get("label_spec"):
+            base, rel_prop = _rel_for_id_name(mapper, f["name"])
+            if not rel_prop:  # scalar field
+                vl2 = _format_label_from_values(f["label_spec"], values_map)
+                if vl2 is not None:
+                    f["value_label"] = vl2
+
+        link_spec = f.get("link")
+        if link_spec:
+            try:
+                href = _build_href(link_spec, values_map, instance)
+            except Exception:
+                href = None
+            if href:
+                f["link_href"] = href
 
     # Build rows (supports nested layout with parents)
     rows_map = _normalize_rows_layout(layout)
@@ -835,5 +1311,6 @@ def render_form(
         values=values_map,
         render_field=render_field,
         submit_attrs=submit_attrs,
-        submit_label=submit_label
+        submit_label=submit_label,
+        model_name=model_cls.__name__
     )
diff --git a/crudkit/ui/templates/field.html b/crudkit/ui/templates/field.html
index 0c0a39b..aecb234 100644
--- a/crudkit/ui/templates/field.html
+++ b/crudkit/ui/templates/field.html
@@ -4,7 +4,13 @@
     {% if label_attrs %}{% for k,v in label_attrs.items() %}
       {{k}}{% if v is not sameas true %}="{{ v }}"{% endif %}
     {% endfor %}{% endif %}>
-    {{ field_label }}
+    {% if link_href %}
+      <a href="{{ link_href }}">
+    {% endif %}
+      {{ field_label }}
+    {% if link_href %}
+      </a>
+    {% endif %}
   </label>
 {% endif %}
 
@@ -30,7 +36,7 @@
   <textarea name="{{ field_name }}" id="{{ field_name }}"
             {% if attrs %}{% for k,v in attrs.items() %}
               {{k}}{% if v is not sameas true %}="{{ v }}"{% endif %}
-            {% endfor %}{% endif %}>{{ value }}</textarea>
+            {% endfor %}{% endif %}>{{ value if value else "" }}</textarea>
 
 {% elif field_type == 'checkbox' %}
   <input type="checkbox" name="{{ field_name }}" id="{{ field_name }}" value="1"
@@ -40,15 +46,33 @@
          {% endfor %}{% endif %}>
 
 {% elif field_type == 'hidden' %}
-  <input type="hidden" name="{{ field_name }}" id="{{ field_name }}" value="{{ value }}">
+  <input type="hidden" name="{{ field_name }}" id="{{ field_name }}" value="{{ value if value else "" }}">
 
 {% elif field_type == 'display' %}
   <div {% if attrs %}{% for k,v in attrs.items() %}
          {{k}}{% if v is not sameas true %}="{{ v }}"{% endif %}
-       {% endfor %}{% endif %}>{{ value }}</div>
+       {% endfor %}{% endif %}>{{ value_label if value_label else (value if value else "") }}</div>
+
+{% elif field_type == "date" %}
+  <input type="date" name="{{ field_name }}" id="{{ field_name }}" value="{{ value if value else "" }}"
+         {% if attrs %}{% for k,v in attrs.items() %}
+           {{k}}{% if v is not sameas true %}="{{ v }}"{% endif %}
+         {% endfor %}{% endif %}>
+
+{% elif field_type == "time" %}
+  <input type="time" name="{{ field_name }}" id="{{ field_name }}" value="{{ value if value else "" }}"
+         {% if attrs %}{% for k,v in attrs.items() %}
+           {{k}}{% if v is not sameas true %}="{{ v }}"{% endif %}
+         {% endfor %}{% endif %}>
+
+{% elif field_type == "datetime" %}
+  <input type="datetime-local" name="{{ field_name }}" id="{{ field_name }}" value="{{ value if value else "" }}"
+         {% if attrs %}{% for k,v in attrs.items() %}
+           {{k}}{% if v is not sameas true %}="{{ v }}"{% endif %}
+         {% endfor %}{% endif %}>
 
 {% else %}
-  <input type="text" name="{{ field_name }}" id="{{ field_name }}" value="{{ value }}"
+  <input type="text" name="{{ field_name }}" id="{{ field_name }}" value="{{ value if value else "" }}"
          {% if attrs %}{% for k,v in attrs.items() %}
            {{k}}{% if v is not sameas true %}="{{ v }}"{% endif %}
          {% endfor %}{% endif %}>
diff --git a/crudkit/ui/templates/form.html b/crudkit/ui/templates/form.html
index b073fc3..f57074a 100644
--- a/crudkit/ui/templates/form.html
+++ b/crudkit/ui/templates/form.html
@@ -1,6 +1,5 @@
-<form method="POST">
+<form method="POST" id="{{ model_name|lower }}_form">
   {% macro render_row(row) %}
-    <!-- {{ row.name }} -->
     {% if row.fields or row.children or row.legend %}
       {% if row.legend %}<legend>{{ row.legend }}</legend>{% endif %}
       <fieldset
@@ -36,5 +35,5 @@
       {% if submit_attrs %}{% for k,v in submit_attrs.items() %}
         {{k}}{% if v is not sameas true %}="{{ v }}"{% endif %}
       {% endfor %}{% endif %}
-  >{{ submit_label if label else 'Save' }}</button>
+  >{{ submit_label if submit_label else 'Save' }}</button>
 </form>