crudkit/core/service.py

@@ -1,7 +1,6 @@
 from __future__ import annotations
 
 from collections.abc import Iterable
-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
 from sqlalchemy.engine import Engine, Connection
@@ -134,25 +133,24 @@ class CRUDService(Generic[T]):
         self.polymorphic = polymorphic
         self.supports_soft_delete = hasattr(model, 'is_deleted')
 
-        self._backend: Optional[BackendInfo] = backend
+        # Derive engine WITHOUT leaking a session/connection
+        bind = getattr(session_factory, "bind", None)
+        if bind is None:
+            tmp_sess = session_factory()
+            try:
+                bind = tmp_sess.get_bind()
+            finally:
+                try:
+                    tmp_sess.close()
+                except Exception:
+                    pass
+
+        eng: Engine = bind.engine if isinstance(bind, Connection) else cast(Engine, bind)
+        self.backend = backend or make_backend_info(eng)
 
     @property
     def session(self) -> Session:
-        """Always return the Flask-scoped Session if available; otherwise the provided factory."""
+        return self._session_factory()
-        try:
-            sess = current_app.extensions["crudkit"]["Session"]
-            return sess
-        except Exception:
-            return self._session_factory()
-
-    @property
-    def backend(self) -> BackendInfo:
-        """Resolve backend info lazily against the active session's engine."""
-        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:
@@ -239,46 +237,85 @@ class CRUDService(Generic[T]):
         include_total: bool = True,
     ) -> "SeekWindow[T]":
         """
-        Keyset pagination with relationship-safe filtering/sorting.
+        Transport-agnostic keyset pagination that preserves all the goodies from `list()`:
-        Always JOIN all CRUDSpec-discovered paths first; then apply filters, sort, seek.
+          - 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.
         """
+        self._debug_bind("seek_window")
         session = self.session
         query, root_alias = self.get_query()
 
-        # Requested fields → projection + optional loaders
+        # Normalize requested fields and compile projection (may skip later to avoid conflicts)
         fields = _normalize_fields_param(params)
         expanded_fields, proj_opts = compile_projection(self.model, fields) if fields else ([], [])
 
         spec = CRUDSpec(self.model, params or {}, root_alias)
 
-        # Parse all inputs so join_paths are populated
         filters = spec.parse_filters()
         order_by = spec.parse_sort()
-        root_fields, rel_field_names, root_field_names = spec.parse_fields()
-        spec.parse_includes()
-        join_paths = tuple(spec.get_join_paths())
 
-        # Soft delete
+        # Field parsing for root load_only fallback
+        root_fields, rel_field_names, root_field_names = spec.parse_fields()
+
+        # Soft delete filter
         query = self._apply_not_deleted(query, root_alias, params)
 
-        # Root column projection (load_only)
+        # Apply filters first
+        if filters:
+            query = query.filter(*filters)
+
+        # Includes + join paths (dotted fields etc.)
+        spec.parse_includes()
+        join_paths = tuple(spec.get_join_paths())  # iterable of (path, relationship_attr, target_alias)
+
+        # Relationship names required by ORDER BY / WHERE
+        sql_hops: set[str] = _paths_needed_for_sql(order_by, filters, join_paths)
+        # Also include relationships mentioned directly in the sort spec
+        sql_hops |= _hops_from_sort(params)
+
+        # First-hop relationship names implied by dotted projection fields
+        proj_hops: set[str] = _paths_from_fields(fields)
+
+        # Root column projection
         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))
 
-        # JOIN all resolved paths, hydrate from the join
+        # Relationship handling per path (avoid loader strategy conflicts)
         used_contains_eager = False
-        for _base_alias, rel_attr, target_alias in join_paths:
+        joined_names: set[str] = set()
-            query = query.join(target_alias, rel_attr.of_type(target_alias), isouter=True)
+
-            query = query.options(contains_eager(rel_attr, alias=target_alias))
+        for _path, relationship_attr, target_alias in join_paths:
+            rel_attr = cast(InstrumentedAttribute, relationship_attr)
+            name = relationship_attr.key
+            if name in sql_hops:
+                # Needed for WHERE/ORDER BY: join + hydrate from that join
+                query = query.join(target_alias, rel_attr.of_type(target_alias), isouter=True)
+                query = query.options(contains_eager(rel_attr, alias=target_alias))
+                used_contains_eager = True
+                joined_names.add(name)
+            elif name in proj_hops:
+                # Display-only: bulk-load efficiently, no join
+                query = query.options(selectinload(rel_attr))
+                joined_names.add(name)
+
+        # Force-join any SQL-needed relationships that weren't in join_paths
+        missing_sql = sql_hops - joined_names
+        for name in missing_sql:
+            rel_attr = cast(InstrumentedAttribute, getattr(root_alias, name))
+            query = query.join(rel_attr, isouter=True)
+            query = query.options(contains_eager(rel_attr))
             used_contains_eager = True
+            joined_names.add(name)
 
-        # Filters AFTER joins → no cartesian products
+        # Apply projection loader options only if they won't conflict with contains_eager
-        if filters:
+        if proj_opts and not used_contains_eager:
-            query = query.filter(*filters)
+            query = query.options(*proj_opts)
 
-        # Order spec (with PK tie-breakers for stability)
+        # Order + limit
-        order_spec = self._extract_order_spec(root_alias, order_by)
+        order_spec = self._extract_order_spec(root_alias, order_by)  # SA 2.x helper
         limit, _ = spec.parse_pagination()
         if limit is None:
             effective_limit = 50
@@ -287,13 +324,13 @@ class CRUDService(Generic[T]):
         else:
             effective_limit = limit
 
-        # Seek predicate from cursor key (if any)
+        # Keyset predicate
         if key:
             pred = self._key_predicate(order_spec, key, backward)
             if pred is not None:
                 query = query.filter(pred)
 
-        # Apply ORDER and LIMIT. Backward is SQL-inverted + reverse in-memory.
+        # Apply ordering. For backward, invert SQL order then reverse in-memory for display.
         if not backward:
             clauses = [(c.desc() if is_desc else c.asc()) for c, is_desc in zip(order_spec.cols, order_spec.desc)]
             query = query.order_by(*clauses)
@@ -307,9 +344,9 @@ class CRUDService(Generic[T]):
                 query = query.limit(effective_limit)
             items = list(reversed(query.all()))
 
-        # Projection meta tag for renderers
+        # Tag projection so your renderer knows what fields were requested
         if fields:
-            proj = list(dict.fromkeys(fields))
+            proj = list(dict.fromkeys(fields))  # dedupe, preserve order
             if "id" not in proj and hasattr(self.model, "id"):
                 proj.insert(0, "id")
         else:
@@ -332,9 +369,12 @@ class CRUDService(Generic[T]):
                 except Exception:
                     pass
 
-        # Cursor key pluck: support related columns we hydrated via contains_eager
+        # Boundary keys for cursor encoding in the API layer
+        # When ORDER BY includes related columns (e.g., owner.first_name),
+        # pluck values from the related object we hydrated with contains_eager/selectinload.
         def _pluck_key_from_obj(obj: Any) -> list[Any]:
             vals: list[Any] = []
+            # Build a quick map: selectable -> relationship name
             alias_to_rel: dict[Any, str] = {}
             for _p, relationship_attr, target_alias in join_paths:
                 sel = getattr(target_alias, "selectable", None)
@@ -342,17 +382,20 @@ class CRUDService(Generic[T]):
                     alias_to_rel[sel] = relationship_attr.key
 
             for col in order_spec.cols:
-                keyname = getattr(col, "key", None) or getattr(col, "name", None)
+                key = getattr(col, "key", None) or getattr(col, "name", None)
-                if keyname and hasattr(obj, keyname):
+                # Try root attribute first
-                    vals.append(getattr(obj, keyname))
+                if key and hasattr(obj, key):
+                    vals.append(getattr(obj, key))
                     continue
+                # Try relationship hop by matching the column's table/selectable
                 table = getattr(col, "table", None)
                 relname = alias_to_rel.get(table)
-                if relname and keyname:
+                if relname and key:
                     relobj = getattr(obj, relname, None)
-                    if relobj is not None and hasattr(relobj, keyname):
+                    if relobj is not None and hasattr(relobj, key):
-                        vals.append(getattr(relobj, keyname))
+                        vals.append(getattr(relobj, key))
                         continue
+                # Give up: unsupported expression for cursor purposes
                 raise ValueError("unpluckable")
             return vals
 
@@ -360,24 +403,33 @@ class CRUDService(Generic[T]):
             first_key = _pluck_key_from_obj(items[0]) if items else None
             last_key  = _pluck_key_from_obj(items[-1]) if items else None
         except Exception:
+            # If we can't derive cursor keys (e.g., ORDER BY expression/aggregate),
+            # disable cursors for this response rather than exploding.
             first_key = None
             last_key = None
 
-        # Count DISTINCT ids with mirrored joins
+        # Optional total that’s safe under JOINs (COUNT DISTINCT ids)
         total = None
         if include_total:
             base = session.query(getattr(root_alias, "id"))
             base = self._apply_not_deleted(base, root_alias, params)
-            # same joins as above for correctness
-            for _base_alias, rel_attr, target_alias in join_paths:
-                base = base.join(target_alias, rel_attr.of_type(target_alias), isouter=True)
             if filters:
                 base = base.filter(*filters)
+            # Mirror join structure for any SQL-needed relationships
+            for _path, relationship_attr, target_alias in join_paths:
+                if relationship_attr.key in sql_hops:
+                    rel_attr = cast(InstrumentedAttribute, relationship_attr)
+                    base = base.join(target_alias, rel_attr.of_type(target_alias), isouter=True)
+            # Also mirror any forced joins
+            for name in (sql_hops - {ra.key for _p, ra, _a in join_paths}):
+                rel_attr = cast(InstrumentedAttribute, getattr(root_alias, name))
+                base = base.join(rel_attr, isouter=True)
+
             total = session.query(func.count()).select_from(
                 base.order_by(None).distinct().subquery()
             ).scalar() or 0
 
-        window_limit_for_body = 0 if effective_limit is None and (limit == 0) else (effective_limit or 50)
+        window_limit_for_body = 0 if effective_limit is None and limit == 0 else (effective_limit or 50)
 
         if log.isEnabledFor(logging.DEBUG):
             log.debug("QUERY: %s", str(query))
@@ -422,53 +474,65 @@ class CRUDService(Generic[T]):
         return [*order_by, *pk_cols]
 
     def get(self, id: int, params=None) -> T | None:
-        """
+        """Fetch a single row by id with conflict-free eager loading and clean projection."""
-        Fetch a single row by id with conflict-free eager loading and clean projection.
+        self._debug_bind("get")
-        Always JOIN any paths that CRUDSpec resolved for filters/fields/includes so
-        related-column filters never create cartesian products.
-        """
         query, root_alias = self.get_query()
 
         # Defaults so we can build a projection even if params is None
-        req_fields: list[str] = _normalize_fields_param(params)
         root_fields: list[Any] = []
         root_field_names: dict[str, str] = {}
         rel_field_names: dict[tuple[str, ...], list[str]] = {}
+        req_fields: list[str] = _normalize_fields_param(params)
 
-        # Soft-delete guard first
+        # Soft-delete guard
         query = self._apply_not_deleted(query, root_alias, params)
 
         spec = CRUDSpec(self.model, params or {}, root_alias)
 
-        # Parse everything so CRUDSpec records any join paths it needed to resolve
+        # Optional extra filters (in addition to id); keep parity with list()
         filters = spec.parse_filters()
-        # no ORDER BY for get()
+        if filters:
-        if params:
+            query = query.filter(*filters)
-            root_fields, rel_field_names, root_field_names = spec.parse_fields()
+
+        # Always filter by id
+        query = query.filter(getattr(root_alias, "id") == id)
+
+        # Includes + join paths we may need
         spec.parse_includes()
         join_paths = tuple(spec.get_join_paths())
 
-        # Root-column projection (load_only)
+        # Field parsing to enable root load_only
+        if params:
+            root_fields, rel_field_names, root_field_names = spec.parse_fields()
+
+        # Decide which relationship paths are needed for SQL vs display-only
+        # For get(), there is no ORDER BY; only filters might force SQL use.
+        sql_hops = _paths_needed_for_sql(order_by=None, filters=filters, join_paths=join_paths)
+        proj_hops = _paths_from_fields(req_fields)
+
+        # Root column projection
         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))
 
-        # JOIN all discovered paths up front; hydrate via contains_eager
+        # Relationship handling per path: avoid loader strategy conflicts
         used_contains_eager = False
-        for _base_alias, rel_attr, target_alias in join_paths:
+        for _path, relationship_attr, target_alias in join_paths:
-            query = query.join(target_alias, rel_attr.of_type(target_alias), isouter=True)
+            rel_attr = cast(InstrumentedAttribute, relationship_attr)
-            query = query.options(contains_eager(rel_attr, alias=target_alias))
+            name = relationship_attr.key
-            used_contains_eager = True
+            if name in sql_hops:
-
+                # Needed in WHERE: join + hydrate from the join
-        # Apply filters (joins are in place → no cartesian products)
+                query = query.join(target_alias, rel_attr.of_type(target_alias), isouter=True)
-        if filters:
+                query = query.options(contains_eager(rel_attr, alias=target_alias))
-            query = query.filter(*filters)
+                used_contains_eager = True
-
+            elif name in proj_hops:
-        # And the id filter
+                # Display-only: bulk-load efficiently
-        query = query.filter(getattr(root_alias, "id") == id)
+                query = query.options(selectinload(rel_attr))
+            else:
+                pass
 
         # Projection loader options compiled from requested fields.
-        # Skip if we used contains_eager to avoid loader-strategy conflicts.
+        # Skip if we used contains_eager to avoid strategy conflicts.
         expanded_fields, proj_opts = compile_projection(self.model, req_fields) if req_fields else ([], [])
         if proj_opts and not used_contains_eager:
             query = query.options(*proj_opts)
@@ -477,7 +541,7 @@ class CRUDService(Generic[T]):
 
         # Emit exactly what the client requested (plus id), or a reasonable fallback
         if req_fields:
-            proj = list(dict.fromkeys(req_fields))
+            proj = list(dict.fromkeys(req_fields))  # dedupe, preserve order
             if "id" not in proj and hasattr(self.model, "id"):
                 proj.insert(0, "id")
         else:
@@ -505,20 +569,17 @@ class CRUDService(Generic[T]):
         return obj or None
 
     def list(self, params=None) -> list[T]:
-        """
+        """Offset/limit listing with smart relationship loading and clean projection."""
-        Offset/limit listing with relationship-safe filtering.
+        self._debug_bind("list")
-        We always JOIN every CRUDSpec-discovered path before applying filters/sorts.
-        """
         query, root_alias = self.get_query()
 
         # Defaults so we can reference them later even if params is None
-        req_fields: list[str] = _normalize_fields_param(params)
         root_fields: list[Any] = []
         root_field_names: dict[str, str] = {}
         rel_field_names: dict[tuple[str, ...], list[str]] = {}
+        req_fields: list[str] = _normalize_fields_param(params)
 
         if params:
-            # Soft delete
             query = self._apply_not_deleted(query, root_alias, params)
 
             spec = CRUDSpec(self.model, params or {}, root_alias)
@@ -526,56 +587,84 @@ class CRUDService(Generic[T]):
             order_by = spec.parse_sort()
             limit, offset = spec.parse_pagination()
 
-            # Includes / fields (populates join_paths)
+            # Includes + join paths we might need
-            root_fields, rel_field_names, root_field_names = spec.parse_fields()
             spec.parse_includes()
             join_paths = tuple(spec.get_join_paths())
 
-            # Root column projection (load_only)
+            # Field parsing for load_only on root columns
+            root_fields, rel_field_names, root_field_names = spec.parse_fields()
+
+            if filters:
+                query = query.filter(*filters)
+
+            # Determine which relationship paths are needed for SQL vs display-only
+            sql_hops = _paths_needed_for_sql(order_by, filters, join_paths)
+            sql_hops |= _hops_from_sort(params)  # ensure sort-driven joins exist
+            proj_hops = _paths_from_fields(req_fields)
+
+            # Root column projection
             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))
 
-            # JOIN all paths we resolved and hydrate them from the join
+            # Relationship handling per path
             used_contains_eager = False
-            for _base_alias, rel_attr, target_alias in join_paths:
+            joined_names: set[str] = set()
-                query = query.join(target_alias, rel_attr.of_type(target_alias), isouter=True)
+
-                query = query.options(contains_eager(rel_attr, alias=target_alias))
+            for _path, relationship_attr, target_alias in join_paths:
+                rel_attr = cast(InstrumentedAttribute, relationship_attr)
+                name = relationship_attr.key
+                if name in sql_hops:
+                    # Needed for WHERE/ORDER BY: join + hydrate from the join
+                    query = query.join(target_alias, rel_attr.of_type(target_alias), isouter=True)
+                    query = query.options(contains_eager(rel_attr, alias=target_alias))
+                    used_contains_eager = True
+                    joined_names.add(name)
+                elif name in proj_hops:
+                    # Display-only: no join, bulk-load efficiently
+                    query = query.options(selectinload(rel_attr))
+                    joined_names.add(name)
+
+            # Force-join any SQL-needed relationships that weren't in join_paths
+            missing_sql = sql_hops - joined_names
+            for name in missing_sql:
+                rel_attr = cast(InstrumentedAttribute, getattr(root_alias, name))
+                query = query.join(rel_attr, isouter=True)
+                query = query.options(contains_eager(rel_attr))
                 used_contains_eager = True
+                joined_names.add(name)
 
-            # Filters AFTER joins → no cartesian products
+            # MSSQL requires ORDER BY when OFFSET is used (SQLA uses OFFSET for limit/offset)
-            if filters:
-                query = query.filter(*filters)
-
-            # MSSQL requires ORDER BY when OFFSET is used; ensure stable PK tie-breakers
             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)
-            if order_by:
-                query = query.order_by(*self._stable_order_by(root_alias, order_by))
 
-            # Offset/limit
+            if order_by:
+                query = query.order_by(*order_by)
+
+            # Only apply offset/limit when not None and not zero
             if offset is not None and offset != 0:
                 query = query.offset(offset)
             if limit is not None and limit > 0:
                 query = query.limit(limit)
 
-            # Projection loaders only if we didn’t use contains_eager
+            # Projection loader options compiled from requested fields.
+            # Skip if we used contains_eager to avoid loader-strategy conflicts.
             expanded_fields, proj_opts = compile_projection(self.model, req_fields) if req_fields else ([], [])
             if proj_opts and not used_contains_eager:
                 query = query.options(*proj_opts)
 
         else:
-            # No params; still honor projection loaders if any
+            # No params means no filters/sorts/limits; still honor projection loaders if any
             expanded_fields, proj_opts = compile_projection(self.model, req_fields) if req_fields else ([], [])
             if proj_opts:
                 query = query.options(*proj_opts)
 
         rows = query.all()
 
-        # Build projection meta for renderers
+        # Emit exactly what the client requested (plus id), or a reasonable fallback
         if req_fields:
-            proj = list(dict.fromkeys(req_fields))
+            proj = list(dict.fromkeys(req_fields))  # dedupe while preserving order
             if "id" not in proj and hasattr(self.model, "id"):
                 proj.insert(0, "id")
         else:
@@ -603,6 +692,7 @@ class CRUDService(Generic[T]):
 
         return rows
 
+
     def create(self, data: dict, actor=None) -> T:
         session = self.session
         obj = self.model(**data)

crudkit/integrations/flask.py

@@ -26,6 +26,32 @@ def init_app(app: Flask, *, runtime: CRUDKitRuntime | None = None, config: type[
     try:
         bound_engine = getattr(SessionFactory, "bind", None) or getattr(SessionFactory, "kw", {}).get("bind") or engine
         pool = bound_engine.pool
+
+        from sqlalchemy import event
+
+        @event.listens_for(pool, "checkout")
+        def _on_checkout(dbapi_conn, conn_record, conn_proxy):
+            sz = pool.size()
+            chk = pool.checkedout()
+            try:
+                conns_in_pool = pool.checkedin()
+            except Exception:
+                conns_in_pool = "?"
+            print(f"POOL CHECKOUT: Pool size: {sz}  Connections in pool: {conns_in_pool} "
+                  f"Current Overflow: {pool.overflow()} Current Checked out connections: {chk} "
+                  f"engine id= {id(bound_engine)}")
+
+        @event.listens_for(pool, "checkin")
+        def _on_checkin(dbapi_conn, conn_record):
+            sz = pool.size()
+            chk = pool.checkedout()
+            try:
+                conns_in_pool = pool.checkedin()
+            except Exception:
+                conns_in_pool = "?"
+            print(f"POOL CHECKIN: Pool size: {sz}  Connections in pool: {conns_in_pool} "
+                  f"Current Overflow: {pool.overflow()} Current Checked out connections: {chk} "
+                  f"engine id= {id(bound_engine)}")
     except Exception as e:
         print(f"[crudkit.init_app] Failed to attach pool listeners: {e}")
 

inventory/__init__.py

@@ -15,6 +15,25 @@ from .routes.index import init_index_routes
 from .routes.listing import init_listing_routes
 from .routes.entry import init_entry_routes
 
+def _bind_pool_debug(engine):
+    pool = engine.pool
+    eid = id(engine)
+
+    @event.listens_for(pool, "checkout")
+    def _on_checkout(dbapi_con, con_record, con_proxy):
+        try:
+            print(f"POOL CHECKOUT: {pool.status()} engine id= {eid}")
+        except Exception:
+            # pool.status is safe on SQLA 2.x, but let's be defensive
+            print(f"POOL CHECKOUT (no status) engine id= {eid}")
+
+    @event.listens_for(pool, "checkin")
+    def _on_checkin(dbapi_con, con_record):
+        try:
+            print(f"POOL CHECKIN: {pool.status()} engine id= {eid}")
+        except Exception:
+            print(f"POOL CHECKIN (no status) engine id= {eid}")
+
 def create_app(config_cls=crudkit.DevConfig) -> Flask:
     app = Flask(__name__)
 
@@ -24,6 +43,8 @@ def create_app(config_cls=crudkit.DevConfig) -> Flask:
     from sqlalchemy import event
 
     engine = runtime.engine
+    print(f"CRUDKit engine id={id(runtime.engine)} url={runtime.engine.url}")
+    _bind_pool_debug(runtime.engine)   # ← attach to the real engine’s pool
 
     # quick status endpoint you can hit while clicking around
     @app.get("/_db_status")

inventory/routes/index.py

@@ -22,7 +22,8 @@ def init_index_routes(app):
             "fields": [
                 "start_time",
                 "contact.label",
-                "work_item.label"
+                "work_item.label",
+                "work_item.device_type.description"
             ],
             "sort": "start_time"
         })
@@ -64,7 +65,8 @@ def init_index_routes(app):
             {"field": "contact.label", "label": "Contact",
              "link": {"endpoint": "entry.entry", "params": {"id": "{contact.id}", "model": "user"}}},
             {"field": "work_item.label", "label": "Work Item",
-             "link": {"endpoint": "entry.entry", "params": {"id": "{work_item.id}", "model": "inventory"}}}
+             "link": {"endpoint": "entry.entry", "params": {"id": "{work_item.id}", "model": "inventory"}}},
+            {"field": "work_item.device_type.description", "label": "Device Type"}
         ]
 
         logs = render_table(work_logs, columns=columns, opts={"object_class": "worklog"})

inventory/routes/listing.py

@@ -18,11 +18,10 @@ def init_listing_routes(app):
             abort(404)
 
         # read query args
-        limit  = request.args.get("limit", None)
+        limit = int(request.args.get("limit", 15))
-        limit  = int(limit) if (limit is not None and str(limit).isdigit()) else 15
+        sort  = request.args.get("sort")  # <- capture sort from URL
-        sort   = request.args.get("sort")
-        fields_qs = request.args.get("fields")
         cursor = request.args.get("cursor")
+        # your decode returns (key, _desc, backward) in this project
         key, _desc, backward = decode_cursor(cursor)
 
         # base spec per model
@@ -103,42 +102,48 @@ def init_listing_routes(app):
                 {"when": {"field": "complete", "is": False}, "class": "table-danger"}
             ]
 
-        # Build params to feed CRUDService (flat dict; parse_filters expects flat keys)
+        # overlay URL-provided sort if present
-        params = dict(spec)
-
-        # overlay fields from query (?fields=...)
-        if fields_qs:
-            params["fields"] = [p.strip() for p in fields_qs.split(",") if p.strip()]
-
-        # overlay sort from query (?sort=...)
         if sort:
-            params["sort"] = sort
+            spec["sort"] = sort
-
-        # limit semantics: 0 means "unlimited" in your service layer
-        params["limit"] = limit
-
-        # forward *all other* query params as filters (flat), excluding known control keys
-        CONTROL_KEYS = {"limit", "cursor", "sort", "fields"}
-        for k, v in request.args.items():
-            if k in CONTROL_KEYS:
-                continue
-            if v is None or v == "":
-                continue
-            params[k] = v
 
         service = crudkit.crud.get_service(cls)
+        try:
+            rt = app.extensions["crudkit"]["runtime"]
+            rt_engine = rt.engine
+        except Exception:
+            rt_engine = None
 
-        window = service.seek_window(params, key=key, backward=backward, include_total=True)
+        try:
+            SessionFactory = app.extensions["crudkit"].get("SessionFactory")
+            sf_engine = getattr(SessionFactory, "bind", None) or getattr(SessionFactory, "kw", {}).get("bind")
+        except Exception:
+            sf_engine = None
+
+        try:
+            bind = service.session.get_bind()
+            svc_engine = getattr(bind, "engine", bind)
+        except Exception:
+            svc_engine = None
+
+        print(
+            "LISTING ENGINES: "
+            f"runtime={id(rt_engine) if rt_engine else None}  "
+            f"session_factory.bind={id(sf_engine) if sf_engine else None}  "
+            f"service.bind={id(svc_engine) if svc_engine else None}"
+        )
+        # include limit and go
+        window = service.seek_window(spec | {"limit": limit}, key=key, backward=backward, include_total=True)
 
         table = render_table(window.items, columns=columns,
                              opts={"object_class": model, "row_classes": row_classes})
 
+        # pass sort through so templates can preserve it on pager links, if they care
         pagination_ctx = {
             "limit": window.limit,
             "total": window.total,
             "next_cursor": encode_cursor(window.last_key, list(window.order.desc), backward=False),
             "prev_cursor": encode_cursor(window.first_key, list(window.order.desc), backward=True),
-            "sort": params.get("sort")  # expose current sort to the template
+            "sort": sort or spec.get("sort")  # expose current sort to the template
         }
 
         return render_template("listing.html", model=model, table=table, pagination=pagination_ctx)