feat(assets): require cursor o field, drop legacy permissive path

Cursor pagination hasn't shipped on either runtime yet — this PR is
still draft and cloud's mirror is just behind it — so there are no
legacy no-o cursors in the wild. Make o mandatory from day one
rather than landing permissive and tightening later.

decode_cursor now rejects any payload without o (or with a non-string
o) as malformed. CursorPayload.order becomes a required str. Tests
that constructed CursorPayload directly now pass order="desc";
test_legacy_cursor_without_order_accepted flips to
test_cursor_without_order_rejected.

app/assets/services/cursor.py

@@ -9,11 +9,10 @@ runtimes. Payload JSON uses short keys to keep the encoded length small:
 The `o` key binds the cursor to the sort direction it was minted under,
 so replaying a `desc` cursor against an `asc` request fails with
 ``INVALID_CURSOR`` rather than silently walking the wrong direction.
-Decoders accept payloads without `o` for backward compatibility with
+`o` is mandatory on every payload — a cursor without it is rejected as
-cursors minted before the binding was introduced (these skip the order
+malformed. Cloud will land the same field in its mirror PR; until then,
-check); new cursors always include it. Cloud has a follow-up to mirror
+Python and cloud cursors differ by exactly the `o` key, and a cloud-
-the field — until then, Python and cloud cursors differ by exactly the
+minted cursor cannot be decoded by this endpoint.
-`o` key.
 
 Encoding is base64url with no padding. JSON serialization escapes `<`,
 `>`, `&`, U+2028, and U+2029 to match Go's default `json.Marshal`
@@ -61,9 +60,7 @@ class CursorPayload:
     sort_field: str
     value: str
     id: str
-    # None means "minted by a producer that did not bind order" (e.g. a cloud
+    order: str
-    # cursor from before BE-944's follow-up lands). New cursors always set it.
-    order: str | None = None
 
 
 # Order direction tokens. Mirrored on the cloud follow-up so cursors carrying
@@ -123,9 +120,8 @@ def decode_cursor(
     timestamp string compared against a ``name`` column).
 
     ``expected_order`` (``"asc"``/``"desc"``), when supplied, must match the
-    payload's ``o`` field. Cursors minted without ``o`` (e.g. by an older
+    payload's ``o`` field. ``o`` is required on every payload; a cursor
-    cloud build) pass the check unconditionally — the binding is best-effort
+    missing it is rejected as malformed.
-    until both runtimes ship the field.
 
     Passing no allowed fields rejects every cursor.
     """
@@ -151,7 +147,7 @@ def decode_cursor(
     sort_field = decoded.get("s")
     value = decoded.get("v")
     id = decoded.get("id")
-    order = decoded.get("o")  # may be absent on legacy cursors
+    order = decoded.get("o")
 
     if not isinstance(sort_field, str) or not isinstance(value, str) or not isinstance(id, str):
         raise InvalidCursorError("payload: missing or non-string s/v/id")
@@ -166,11 +162,11 @@ def decode_cursor(
     if sort_field not in allowed_sort_fields:
         raise InvalidCursorError(f"unsupported sort field {sort_field!r}")
 
-    if order is not None and not isinstance(order, str):
+    if not isinstance(order, str):
-        raise InvalidCursorError("payload: non-string o")
+        raise InvalidCursorError("missing or non-string o")
-    if order is not None and order not in _VALID_ORDERS:
+    if order not in _VALID_ORDERS:
         raise InvalidCursorError(f"unsupported order {order!r}")
-    if expected_order is not None and order is not None and order != expected_order:
+    if expected_order is not None and order != expected_order:
         raise InvalidCursorError(
             f"cursor order {order!r} does not match request order {expected_order!r}"
         )

tests-unit/assets_test/services/test_cursor.py

@@ -61,7 +61,7 @@ class TestTimeCursor:
         assert decoded == ts
 
     def test_decode_returns_utc(self):
-        payload = CursorPayload(sort_field="created_at", value="1716200000123456", id="id-1")
+        payload = CursorPayload(sort_field="created_at", value="1716200000123456", id="id-1", order="desc")
         decoded = decode_cursor_time(payload)
         assert decoded.tzinfo == timezone.utc
 
@@ -72,7 +72,7 @@ class TestTimeCursor:
 
     def test_non_integer_value_rejected_on_decode(self):
         with pytest.raises(InvalidCursorError):
-            decode_cursor_time(CursorPayload("created_at", "not-a-number", "id-1"))
+            decode_cursor_time(CursorPayload("created_at", "not-a-number", "id-1", "desc"))
 
     def test_none_payload_rejected(self):
         with pytest.raises(InvalidCursorError):
@@ -89,11 +89,11 @@ class TestTimeCursor:
 
 class TestIntCursor:
     def test_decode_int(self):
-        assert decode_cursor_int(CursorPayload("size", "1024", "id-1")) == 1024
+        assert decode_cursor_int(CursorPayload("size", "1024", "id-1", "desc")) == 1024
 
     def test_decode_int_rejects_non_int(self):
         with pytest.raises(InvalidCursorError):
-            decode_cursor_int(CursorPayload("size", "abc", "id-1"))
+            decode_cursor_int(CursorPayload("size", "abc", "id-1", "desc"))
 
     def test_decode_int_rejects_none(self):
         with pytest.raises(InvalidCursorError):
@@ -223,14 +223,14 @@ class TestOrderBinding:
         with pytest.raises(InvalidCursorError, match="unsupported order"):
             decode_cursor(encoded, ALLOWED)
 
-    def test_legacy_cursor_without_order_accepted(self):
+    def test_cursor_without_order_rejected(self):
-        """Cursors minted by a producer that didn't include `o` (e.g. an older
+        """`o` is mandatory. A cursor minted without it is rejected as
-        cloud build) must still decode — the order binding is best-effort
+        malformed rather than silently walking the keyset in whatever
-        until cloud mirrors the field."""
+        direction the request happens to ask for."""
         raw = b'{"s":"name","v":"x","id":"id-1"}'
         encoded = base64.urlsafe_b64encode(raw).rstrip(b"=").decode("ascii")
-        payload = decode_cursor(encoded, ALLOWED, expected_order="desc")
+        with pytest.raises(InvalidCursorError, match="missing or non-string o"):
-        assert payload.order is None  # binding skipped, decode succeeds
+            decode_cursor(encoded, ALLOWED, expected_order="desc")
 
 
 class TestGoCompatJsonEscaping: