diff --git a/src/utils/filter.py b/src/utils/filter.py
index 850fb70d..3d035e06 100644
--- a/src/utils/filter.py
+++ b/src/utils/filter.py
@@ -65,6 +65,48 @@ ALLOWED_EXTERNAL_TO_INTERNAL_COLUMN_MAPPING_DOCUMENTS = {
 
 MAX_SESSION_ALLOWLIST_ENTRIES = 1000
 
+# Values that can be bound to a non-JSONB column. Anything else (dict, list,
+# bytes, arbitrary objects) compiles into a valid statement and then fails in
+# psycopg at execute time as an unhandled 500, so it is rejected up front.
+SCALAR_OPERAND_TYPES = (
+    str,
+    bool,
+    int,
+    float,
+    Decimal,
+    datetime.datetime,
+    datetime.date,
+)
+
+
+def _require_bindable_operand(
+    column_name: str, op_value: Any, operator: str = ""
+) -> None:
+    """Reject an operand that cannot be bound to a scalar column.
+
+    For ``in``, each element is checked: a dict nested in the list is bound the
+    same way a bare dict operand would be, and fails identically.
+
+    Args:
+        column_name: Internal column name, for the error message.
+        op_value: The operand to check.
+        operator: The comparison operator, when the operand came from one.
+
+    Raises:
+        FilterError: If a value is neither None nor a scalar.
+    """
+    values: Sequence[Any] = (
+        typing_cast("Sequence[Any]", op_value)
+        if operator == "in" and isinstance(op_value, list | tuple | set)
+        else (op_value,)
+    )
+    for value in values:
+        if value is None or isinstance(value, SCALAR_OPERAND_TYPES):
+            continue
+        raise FilterError(
+            f"Invalid value for column '{column_name}': expected a scalar, got {type(value).__name__}"
+        )
+
 
 def extract_session_allowlist(
     filters: dict[str, Any] | None,
@@ -183,9 +225,21 @@ def apply_filter(
     if filters is None:
         return stmt
 
-    conditions = _build_filter_conditions(filters, model_class)
-    if conditions is not None:
-        stmt = stmt.where(conditions)
+    # Fail closed. The filter body is arbitrary client JSON, so any shape the
+    # DSL doesn't recognize must become a 422, not an unhandled 500 from
+    # somewhere deep in SQLAlchemy. The exception is still logged in full so a
+    # genuine bug in the builder stays visible rather than being swallowed.
+    try:
+        conditions = _build_filter_conditions(filters, model_class)
+        if conditions is not None:
+            stmt = stmt.where(conditions)
+    except FilterError:
+        raise
+    except Exception:
+        logger.exception(
+            "Unexpected error building filter for %s: %r", model_class.__name__, filters
+        )
+        raise FilterError("Invalid filter configuration") from None
 
     return stmt
 
@@ -361,6 +415,7 @@ def _build_field_condition(
         if column_name in JSONB_COLUMNS:
             return column.contains(value)
         else:
+            _require_bindable_operand(column_name, value)
             return column == value
 
 
@@ -593,6 +648,11 @@ def _build_comparison_conditions(
             conditions.append(column.is_not(None))
             continue
 
+        # Every operand bound to a scalar column must itself be a scalar. JSONB
+        # columns are exempt: a dict there is a containment match.
+        if not isinstance(column.type, JSONB):
+            _require_bindable_operand(column_name, op_value, operator)
+
         condition = None
 
         # For datetime columns, cast string values to timestamp
diff --git a/tests/utils/test_filter.py b/tests/utils/test_filter.py
index 140023c1..ac10e53e 100644
--- a/tests/utils/test_filter.py
+++ b/tests/utils/test_filter.py
@@ -1,10 +1,17 @@
 """Unit tests for filter condition building."""
 
+from collections.abc import Sequence
+from datetime import datetime
+from decimal import Decimal
+from typing import Any, cast
+
 import pytest
 from sqlalchemy import select
+from sqlalchemy.dialects import postgresql
+from sqlalchemy.dialects.postgresql import JSONB
 
 from src.exceptions import FilterError
-from src.models import Document, Message
+from src.models import Document, Message, Peer, Session
 from src.utils.filter import apply_filter
 
 
@@ -58,3 +65,115 @@ def test_ne_none_on_numeric_column_is_not_null():
 def test_null_operand_on_non_ne_operator_raises():
     with pytest.raises(FilterError):
         apply_filter(select(Message), Message, {"token_count": {"gt": None}})
+
+
+# --- Invariants over the whole DSL -------------------------------------------
+#
+# The filter body is arbitrary client JSON. Enumerating bad shapes one at a time
+# is endless, so these two tests assert the properties that make any unhandled
+# shape a 422 instead of a 500, and fail on the next shape nobody thought of.
+
+_OPERANDS: list[Any] = [
+    None,
+    True,
+    False,
+    0,
+    -1,
+    1.5,
+    "",
+    "abc",
+    "*",
+    [],
+    [None],
+    [[1]],
+    [{"a": 1}],
+    {},
+    {"operator": "null"},
+    {"ne": None},
+    {"ne": {"a": 1}},
+    {"ne": [1]},
+    {"in": None},
+    {"in": "abc"},
+    {"in": [{"a": 1}]},
+    {"in": [[1]]},
+    {"gt": {}},
+    {"gt": []},
+    {"gt": True},
+    {"contains": None},
+    {"contains": {"a": 1}},
+    {"lt": [1, 2]},
+]
+
+_COLUMNS: dict[Any, list[str]] = {
+    Document: ["session_id", "workspace_id", "metadata", "level", "source_ids", "id"],
+    Message: ["session_id", "peer_id", "token_count", "created_at", "metadata"],
+    Session: ["id", "is_active", "created_at", "configuration"],
+    Peer: ["id", "created_at", "metadata"],
+}
+
+_MALFORMED: list[dict[str, Any]] = [
+    {"AND": "notalist"},
+    {"AND": [None]},
+    {"AND": [[]]},
+    {"AND": [1]},
+    {"OR": [None]},
+    {"OR": [1]},
+    {"NOT": None},
+    {"NOT": [None]},
+    {"unknown_column": 1},
+    {"metadata": None},
+]
+
+
+def _filter_shapes() -> list[tuple[Any, dict[str, Any]]]:
+    shapes: list[tuple[Any, dict[str, Any]]] = []
+    for model, columns in _COLUMNS.items():
+        for column in columns:
+            for operand in _OPERANDS:
+                leaf = {column: operand}
+                shapes.append((model, leaf))
+                shapes.append((model, {"AND": [leaf]}))
+                shapes.append((model, {"NOT": [leaf]}))
+        shapes.extend((model, bad) for bad in _MALFORMED)
+    return shapes
+
+
+def test_every_filter_shape_either_compiles_or_raises_filter_error():
+    """No filter body may escape as anything other than a compiled statement or
+    a FilterError. Anything else reaches the client as an unhandled 500."""
+    escaped: list[tuple[str, dict[str, Any], str]] = []
+    for model, filters in _filter_shapes():
+        try:
+            str(apply_filter(select(model), model, filters))
+        except FilterError:
+            pass
+        except Exception as exc:  # pragma: no cover - failure path
+            escaped.append((model.__name__, filters, type(exc).__name__))
+    assert not escaped, f"non-FilterError escapes: {escaped[:10]}"
+
+
+def test_no_non_scalar_value_is_bound_to_a_scalar_column():
+    """A dict or list bound to a non-JSONB parameter compiles cleanly and then
+    fails in psycopg at execute time — the original 500. Nothing may reach that
+    state, including non-scalars nested inside an `in` list."""
+    offenders: list[tuple[str, dict[str, Any], str]] = []
+    for model, filters in _filter_shapes():
+        try:
+            stmt = apply_filter(select(model), model, filters)
+        except FilterError:
+            continue
+        compiled = stmt.compile(dialect=postgresql.dialect())
+        for bind in compiled.binds.values():
+            if isinstance(bind.type, JSONB):
+                continue
+            value: Any = bind.value
+            # An expanding IN bind holds the list itself; check its elements.
+            elements = cast(
+                "Sequence[Any]", value if isinstance(value, list | tuple) else [value]
+            )
+            for element in elements:
+                if element is not None and not isinstance(
+                    element, str | bool | int | float | Decimal | datetime
+                ):
+                    offenders.append((model.__name__, filters, repr(element)[:40]))
+    assert not offenders, f"non-scalar bound to scalar column: {offenders[:10]}"