diff --git a/comfy_cli/command/assets_library.py b/comfy_cli/command/assets_library.py
index 07ca2cfd9..68e79fef0 100644
--- a/comfy_cli/command/assets_library.py
+++ b/comfy_cli/command/assets_library.py
@@ -15,6 +15,7 @@
 
 from comfy_cli import tracking
 from comfy_cli.command.cloud_http import (
+    ResponseUnparseable,
     cloud_target_or_local_error,
     handle_cloud_http_error,
     http_request,
@@ -56,29 +57,98 @@ def ls_cmd(
     url = target.url("assets") + "?" + urllib.parse.urlencode(params)
 
     try:
-        _, body = http_request(url, target)
+        # `strict_json` so a 200 carrying a proxy/captive-portal error page is
+        # reported as malformed rather than collapsed to `None`, which is
+        # indistinguishable here from a genuinely empty body — and would render
+        # a broken response as a successful EMPTY library, the same masquerade
+        # the shape guards below refuse to make.
+        _, body = http_request(url, target, strict_json=True)
     except (urllib.error.HTTPError, urllib.error.URLError, OSError) as e:
         raise handle_cloud_http_error(renderer, e, operation="list") from e
+    except ResponseUnparseable as e:
+        renderer.error(
+            code="cloud_http_error",
+            message="unexpected response from /api/assets (body was not valid JSON)",
+            hint="check whether a proxy or captive portal is intercepting the request",
+            details={"operation": "list"},
+        )
+        raise typer.Exit(code=1) from e
+
+    # A non-empty body decodes here only if it was valid JSON, but valid JSON is
+    # not necessarily the shape we asked for: a proxy or error page can answer 200
+    # with an array or a scalar, and `b.get(...)` would then raise a raw
+    # `AttributeError` past the `except` above — a traceback instead of an error
+    # envelope. Guard it the way `workflow list` guards the same failure on the
+    # sibling endpoint. An empty body stays a legitimate `None` (→ no rows).
+    if body is not None and not isinstance(body, dict):
+        renderer.error(
+            code="cloud_http_error",
+            message="unexpected response shape from /api/assets (expected a JSON object)",
+            details={"got_type": type(body).__name__},
+        )
+        raise typer.Exit(code=1)
 
-    rows = (body or {}).get("assets") or []
+    b = body or {}
+    # A missing/empty `assets` is a legitimately-empty listing; a present non-list
+    # `assets` is malformed the same way, and `len(rows)` would raise `TypeError`.
+    rows = b.get("assets")
+    if rows is None:
+        rows = []
+    elif not isinstance(rows, list):
+        renderer.error(
+            code="cloud_http_error",
+            message="unexpected response shape from /api/assets (assets must be a JSON array)",
+            details={"got_type": type(rows).__name__},
+        )
+        raise typer.Exit(code=1)
+    assets = [
+        {
+            "id": r.get("id"),
+            "name": r.get("name"),
+            "hash": r.get("hash"),
+            "mime_type": r.get("mime_type"),
+            "size": r.get("size"),
+            "tags": r.get("tags"),
+            "preview_url": r.get("preview_url"),
+            "job_id": r.get("job_id"),
+            "created_at": r.get("created_at"),
+        }
+        for r in rows
+        if isinstance(r, dict)
+    ]
     payload = {
-        "count": len(rows),
-        "assets": [
-            {
-                "id": r.get("id"),
-                "name": r.get("name"),
-                "hash": r.get("hash"),
-                "mime_type": r.get("mime_type"),
-                "size": r.get("size"),
-                "tags": r.get("tags"),
-                "preview_url": r.get("preview_url"),
-                "job_id": r.get("job_id"),
-                "created_at": r.get("created_at"),
-            }
-            for r in rows
-            if isinstance(r, dict)
-        ],
+        # `count` describes the array actually emitted, not the raw server rows.
+        # A non-dict row is dropped from `assets` above, so counting raw rows
+        # reported more items than the payload carries — and a knowingly-skewed
+        # `count` sitting next to a forwarded `total` undercuts the point of
+        # forwarding an authoritative count at all. Identical to the old value
+        # for every well-formed response; it differs only where the old value
+        # was simply wrong. Matches the repo convention in
+        # `comfy_cli/command/nodes.py`, where `count` is over the emitted rows.
+        "count": len(assets),
+        "assets": assets,
     }
+    # Forward the server's own truncation signal instead of making callers infer
+    # it from an exactly-full page: `has_more`/`total` are both `required` on the
+    # cloud API's `ListAssetsResponse`, and `has_more` comes off a limit+1
+    # sentinel row rather than off the returned row count, so it stays correct
+    # on a page that comes back short. Only forward what the server actually
+    # sent — an older or local server may omit them, and a JSON `null` in the
+    # envelope would poison a consumer's type assertion, so omit the key rather
+    # than emitting None.
+    #
+    # Each field is checked against its OWN type, not a shared bool-or-int test:
+    # `bool` is a subclass of `int` in Python, so one shared check would forward
+    # `has_more: 0` and `total: false` and emit an envelope that violates
+    # schemas/assets_library.json — the very contract this command publishes.
+    if isinstance(b.get("has_more"), bool):
+        payload["has_more"] = b["has_more"]
+    total = b.get("total")
+    # `>= 0` because the schema publishes `total` as a non-negative integer, and a
+    # guard looser than the declared contract is how the envelope ends up violating
+    # its own schema.
+    if isinstance(total, int) and not isinstance(total, bool) and total >= 0:
+        payload["total"] = total
     renderer.emit(payload, command="assets library ls", where="cloud")
 
 
diff --git a/comfy_cli/command/cloud_http.py b/comfy_cli/command/cloud_http.py
index a49b5ab72..ce903886b 100644
--- a/comfy_cli/command/cloud_http.py
+++ b/comfy_cli/command/cloud_http.py
@@ -46,11 +46,32 @@ def _authed_request(
     return req
 
 
+class ResponseUnparseable(Exception):
+    """A non-empty response body that was not valid JSON.
+
+    Only raised when a caller opts in via ``http_request(..., strict_json=True)``;
+    otherwise a decode failure keeps collapsing to ``None`` as it always has.
+    """
+
+
 def http_request(
-    url: str, target, *, method: str = "GET", body: dict | None = None, timeout: float = 30.0
+    url: str,
+    target,
+    *,
+    method: str = "GET",
+    body: dict | None = None,
+    timeout: float = 30.0,
+    strict_json: bool = False,
 ) -> tuple[int, dict | None]:
     """Authed HTTP call returning (status, parsed_json_or_none). Raises
-    urllib errors verbatim so callers can surface the right error code."""
+    urllib errors verbatim so callers can surface the right error code.
+
+    An *empty* body is reported as ``None``. A *non-empty* body that fails to
+    decode is reported as ``None`` too, unless ``strict_json`` is set — then it
+    raises ``ResponseUnparseable``. The two are different answers ("nothing to
+    say" vs. "a malformed answer"), and a listing caller that conflates them
+    reports a proxy or captive-portal error page as a successful empty listing.
+    """
     import urllib.request
 
     data = json.dumps(body).encode("utf-8") if body is not None else None
@@ -59,11 +80,28 @@ def http_request(
     with urllib.request.urlopen(req, timeout=timeout) as resp:
         status = resp.status
         raw = resp.read(64 * 1024 * 1024)  # 64 MiB cap
-    if not raw:
+    # `.strip()` so a whitespace-only body counts as empty rather than as a
+    # decode failure: identical to the old behaviour on the default path (it
+    # collapsed to `None` either way), but it keeps `strict_json` from calling a
+    # server that answers `b"\n"` malformed.
+    if not raw.strip():
         return status, None
     try:
         return status, json.loads(raw)
-    except json.JSONDecodeError:
+    except (ValueError, RecursionError) as e:
+        # Catch the ``ValueError`` BASE, not ``JSONDecodeError``: handed raw bytes,
+        # ``json.loads`` rejects a malformed body with three different exceptions and
+        # only one of them is a ``JSONDecodeError``. Non-UTF-8 bytes (``b"\xff"``, a
+        # binary error page) raise ``UnicodeDecodeError``; a JSON integer past
+        # CPython's 4300-digit int/str limit raises a bare ``ValueError``; both
+        # subclass ``ValueError``. Pathologically nested input raises
+        # ``RecursionError``, which does not — the 64 MiB read permits deep nesting.
+        # Narrowing to ``JSONDecodeError`` let those escape past ``strict_json`` and
+        # past every call site's ``except``, crashing the CLI with a raw traceback
+        # for exactly the malformed response this helper exists to classify. Same
+        # reasoning, and the same catch, as the sibling helper in ``workflow.py``.
+        if strict_json:
+            raise ResponseUnparseable(f"non-JSON response body from {url}") from e
         return status, None
 
 
diff --git a/comfy_cli/schemas/assets_library.json b/comfy_cli/schemas/assets_library.json
index 4c3aa529d..7494ba9ce 100644
--- a/comfy_cli/schemas/assets_library.json
+++ b/comfy_cli/schemas/assets_library.json
@@ -25,6 +25,8 @@
     }
    }
   },
+  "has_more": { "type": "boolean" },
+  "total": { "type": "integer", "minimum": 0 },
   "id": { "type": ["string", "null"] },
   "hash": { "type": ["string", "null"] },
   "created_new": { "type": ["boolean", "null"] }
diff --git a/tests/comfy_cli/command/test_assets_library.py b/tests/comfy_cli/command/test_assets_library.py
index d8477caa1..3faf5c500 100644
--- a/tests/comfy_cli/command/test_assets_library.py
+++ b/tests/comfy_cli/command/test_assets_library.py
@@ -1,6 +1,17 @@
-"""``comfy assets library`` error envelopes.
+"""``comfy assets library`` envelopes.
 
-Pins the 404 mapping for ``assets library ensure``. Found in prod (Langfuse
+Pins two things: the ``assets library ls`` pagination passthrough, and the 404
+mapping for ``assets library ensure``.
+
+``ls`` forwards the server's ``has_more``/``total`` so a downstream consumer
+(the cloud agent's asset gate) reads an authoritative truncation signal instead
+of inferring one from a page that came back exactly full — which both misses a
+short truncated page and misreads a library of exactly ``--limit`` assets. Both
+fields are ``required`` on the cloud API's ``ListAssetsResponse``, but a server
+that omits them must leave the keys ABSENT rather than emit ``null``, because
+the consumer type-asserts them out of the decoded envelope.
+
+The 404 mapping was found in prod (Langfuse
 2026-08-25, ``use_asset_as_input``): an agent passed a FILE NAME
 (``comfyorg_logo.png``) where the content hash belongs, the API answered 404,
 and the CLI reported ``workflow_not_found`` / "workflow not found (ensure)"
@@ -89,6 +100,211 @@ def _fake(req, timeout=None):
     return calls
 
 
+def _patch_urlopen_raw(monkeypatch: pytest.MonkeyPatch, raw: bytes):
+    """Like ``_patch_urlopen`` but serves ``raw`` verbatim, so a test can send a
+    body that is not valid JSON at all (or is genuinely empty) rather than one
+    that round-trips through ``json.dumps``."""
+
+    class _Resp:
+        status = 200
+
+        def __enter__(self):
+            return self
+
+        def __exit__(self, *a):
+            return False
+
+        def read(self, n=None):
+            return raw
+
+    monkeypatch.setattr("urllib.request.urlopen", lambda req, timeout=None: _Resp())
+
+
+class TestLsPagination:
+    """`ls` forwards the server's truncation signal, and only when it sent one."""
+
+    def _ls(self, monkeypatch, capsys, body: dict) -> dict[str, Any]:
+        _patch_urlopen(monkeypatch, body)
+        env = _run(["ls", "--where", "cloud"], capsys)
+        assert env["ok"] is True, env
+        return env["data"]
+
+    def test_forwards_has_more_and_total(self, cloud_target, monkeypatch, capsys):
+        data = self._ls(
+            monkeypatch,
+            capsys,
+            {"assets": [{"id": "a1", "name": "cat.png", "hash": "h1"}], "has_more": True, "total": 1234},
+        )
+        assert data["has_more"] is True
+        assert data["total"] == 1234
+        # Alongside, not instead of, the existing shape.
+        assert data["count"] == 1
+        assert data["assets"][0]["id"] == "a1"
+
+    def test_forwards_has_more_false(self, cloud_target, monkeypatch, capsys):
+        # `false` is a real answer, not a missing one — the falsy value must
+        # survive, otherwise the consumer cannot distinguish "not truncated"
+        # from "server did not say".
+        data = self._ls(monkeypatch, capsys, {"assets": [], "has_more": False, "total": 0})
+        assert data["has_more"] is False
+        assert data["total"] == 0
+
+    def test_omits_keys_when_server_does_not_send_them(self, cloud_target, monkeypatch, capsys):
+        data = self._ls(monkeypatch, capsys, {"assets": [{"id": "a1"}]})
+        assert "has_more" not in data
+        assert "total" not in data
+        assert data["count"] == 1
+
+    def test_omits_keys_when_server_sends_nulls(self, cloud_target, monkeypatch, capsys):
+        # A forwarded `null` would poison the consumer's type assertion, so a
+        # null is treated exactly like an absent key.
+        data = self._ls(monkeypatch, capsys, {"assets": [], "has_more": None, "total": None})
+        assert "has_more" not in data
+        assert "total" not in data
+
+    def test_omits_cross_typed_values(self, cloud_target, monkeypatch, capsys):
+        # `bool` is a subclass of `int` in Python, so a shared bool-or-int check
+        # would let `has_more: 0` / `total: false` through and emit an envelope
+        # that violates this command's own published schema. Each field is
+        # validated against its own type instead, and a mistyped value is
+        # dropped exactly like an absent one.
+        data = self._ls(monkeypatch, capsys, {"assets": [], "has_more": 0, "total": False})
+        assert "has_more" not in data
+        assert "total" not in data
+
+    def test_omits_a_negative_total(self, cloud_target, monkeypatch, capsys):
+        # The schema publishes `total` as `minimum: 0`, so forwarding a negative
+        # server value would emit an envelope violating this command's own
+        # contract — the same class of bug as the cross-typed case above.
+        data = self._ls(monkeypatch, capsys, {"assets": [], "has_more": False, "total": -1})
+        assert "total" not in data
+        assert data["has_more"] is False
+
+    def test_omits_values_of_the_wrong_json_type(self, cloud_target, monkeypatch, capsys):
+        data = self._ls(monkeypatch, capsys, {"assets": [], "has_more": "true", "total": "1234"})
+        assert "has_more" not in data
+        assert "total" not in data
+
+    def test_non_object_body_is_an_error_envelope_not_a_traceback(self, cloud_target, monkeypatch, capsys):
+        # A proxy or error page can answer 200 with valid JSON that is not an
+        # object. `b.get(...)` would raise a bare AttributeError past the
+        # HTTPError/URLError/OSError handler, so the user would see a traceback
+        # rather than an envelope.
+        _patch_urlopen(monkeypatch, [1, 2, 3])
+        env = _run(["ls", "--where", "cloud"], capsys)
+        assert env["ok"] is False
+        assert env["error"]["code"] == "cloud_http_error"
+        assert error_codes.is_registered(env["error"]["code"])
+        assert env["error"]["details"]["got_type"] == "list"
+
+    def test_non_list_assets_is_an_error_envelope_not_a_traceback(self, cloud_target, monkeypatch, capsys):
+        # Same failure one level down: `len(rows)` on a scalar raises TypeError.
+        _patch_urlopen(monkeypatch, {"assets": 42})
+        env = _run(["ls", "--where", "cloud"], capsys)
+        assert env["ok"] is False
+        assert env["error"]["code"] == "cloud_http_error"
+        assert env["error"]["details"]["got_type"] == "int"
+
+    def test_json_null_body_is_an_empty_listing_not_an_error(self, cloud_target, monkeypatch, capsys):
+        data = self._ls(monkeypatch, capsys, None)
+        assert data["count"] == 0
+        assert data["assets"] == []
+
+    def test_truly_empty_body_is_an_empty_listing_not_an_error(self, cloud_target, monkeypatch, capsys):
+        # Zero bytes: the server had nothing to say, which is a legitimate
+        # empty library and must stay a success envelope.
+        _patch_urlopen_raw(monkeypatch, b"")
+        env = _run(["ls", "--where", "cloud"], capsys)
+        assert env["ok"] is True, env
+        assert env["data"]["count"] == 0
+        assert env["data"]["assets"] == []
+
+    def test_whitespace_only_body_is_an_empty_listing_not_an_error(self, cloud_target, monkeypatch, capsys):
+        # A body of just a newline is "nothing to say", not a malformed answer;
+        # `strict_json` must not turn it into an error envelope.
+        _patch_urlopen_raw(monkeypatch, b"\n")
+        env = _run(["ls", "--where", "cloud"], capsys)
+        assert env["ok"] is True, env
+        assert env["data"]["count"] == 0
+
+    def test_invalid_json_body_is_an_error_envelope_not_an_empty_listing(self, cloud_target, monkeypatch, capsys):
+        # `http_request` collapses a JSONDecodeError to `None` by default, which
+        # is indistinguishable from the empty body above — so a proxy or
+        # captive-portal error page answering 200 rendered as a successful EMPTY
+        # library. `ls` opts into `strict_json` so the two stay distinct.
+        _patch_urlopen_raw(monkeypatch, b"<html><body>502 Bad Gateway</body></html>")
+        env = _run(["ls", "--where", "cloud"], capsys)
+        assert env["ok"] is False
+        assert env["error"]["code"] == "cloud_http_error"
+        assert error_codes.is_registered(env["error"]["code"])
+
+    def test_invalid_utf8_body_is_an_error_envelope_not_a_traceback(self, cloud_target, monkeypatch, capsys):
+        # Handed raw bytes, `json.loads` rejects a NON-UTF-8 body with
+        # `UnicodeDecodeError`, not `JSONDecodeError` — so a binary error page (or
+        # a gzip/TLS fragment from a misbehaving proxy) escaped the `strict_json`
+        # catch entirely and crashed the CLI with a traceback, past the very
+        # handler added to stop invalid JSON from masquerading as an empty
+        # library. Same malformed answer as the test above, different bytes.
+        #
+        # These bytes are chosen to be invalid UTF-8 that is ALSO not a BOM:
+        # `json.loads` sniffs raw bytes for UTF-16/32 (RFC 4627), so a body
+        # starting `\xff\x00` is guessed as UTF-16-LE, decodes to garbage text and
+        # fails as an ordinary `JSONDecodeError` — which the narrow catch already
+        # handled, making it a test that passes with or without the fix.
+        _patch_urlopen_raw(monkeypatch, b"\x80\x81\x82 binary garbage")
+        env = _run(["ls", "--where", "cloud"], capsys)
+        assert env["ok"] is False
+        assert env["error"]["code"] == "cloud_http_error"
+        assert error_codes.is_registered(env["error"]["code"])
+
+    def test_count_matches_the_emitted_assets_and_projection_is_unchanged(self, cloud_target, monkeypatch, capsys):
+        rows = [
+            {
+                "id": "a1",
+                "name": "cat.png",
+                "hash": "h1",
+                "mime_type": "image/png",
+                "size": 12,
+                "tags": ["input"],
+                "preview_url": "https://example.com/p.png",
+                "job_id": "j1",
+                "created_at": "2026-01-01T00:00:00Z",
+                "extra_server_field": "dropped",
+            },
+            "not-a-dict",
+        ]
+        data = self._ls(monkeypatch, capsys, {"assets": rows, "has_more": False, "total": 1})
+        # `count` describes the array actually emitted: the non-dict row is
+        # dropped from `assets`, so counting it too would report more items than
+        # the payload carries — misleading in general, and self-defeating beside
+        # a forwarded `total` whose whole purpose is an authoritative count.
+        assert data["count"] == 1
+        assert data["total"] == 1
+        assert data["assets"] == [
+            {
+                "id": "a1",
+                "name": "cat.png",
+                "hash": "h1",
+                "mime_type": "image/png",
+                "size": 12,
+                "tags": ["input"],
+                "preview_url": "https://example.com/p.png",
+                "job_id": "j1",
+                "created_at": "2026-01-01T00:00:00Z",
+            }
+        ]
+
+    def test_envelope_validates_against_the_published_schema(self, cloud_target, monkeypatch, capsys):
+        import json as _json
+        from pathlib import Path
+
+        import jsonschema
+
+        data = self._ls(monkeypatch, capsys, {"assets": [{"id": "a1"}], "has_more": True, "total": 7})
+        schema_path = Path(assets_library.__file__).resolve().parents[1] / "schemas" / "assets_library.json"
+        jsonschema.Draft202012Validator(_json.loads(schema_path.read_text())).validate(data)
+
+
 class TestEnsure:
     def test_404_is_asset_not_found_not_workflow_not_found(self, cloud_target, monkeypatch, capsys):
         _patch_urlopen(monkeypatch, _http_error(404))
@@ -116,3 +332,69 @@ def test_success_reports_id_hash_and_created_new(self, cloud_target, monkeypatch
         assert env["ok"] is True
         assert env["data"] == {"id": "asset-1", "hash": "a" * 64, "created_new": True}
         assert calls[0]["url"].endswith("/api/assets/from-hash") and calls[0]["method"] == "POST"
+
+
+class TestHttpRequestRejectsEveryUnparseableBody:
+    """`http_request`'s decode guard is keyed on the malformed body, not on which
+    exception the parser happened to pick.
+
+    `json.loads` rejects a malformed body with three different exceptions and only
+    one is a `JSONDecodeError`; the other two are exercised here through a patched
+    `json.loads` rather than through pathological input, because the thresholds
+    that produce them (CPython's 4300-digit int limit, the recursion limit) move
+    between interpreter versions and would make the test a platform coin-flip.
+    The invalid-UTF-8 case is deterministic everywhere and is pinned end-to-end
+    in `TestLsPagination` above.
+    """
+
+    @staticmethod
+    def _request(monkeypatch, target, exc: BaseException, *, strict_json: bool):
+        from comfy_cli.command import cloud_http
+
+        class _Resp:
+            status = 200
+
+            def __enter__(self):
+                return self
+
+            def __exit__(self, *a):
+                return False
+
+            def read(self, n=None):
+                return b'{"assets": []}'
+
+        monkeypatch.setattr("urllib.request.urlopen", lambda req, timeout=None: _Resp())
+
+        def _boom(_raw):
+            raise exc
+
+        monkeypatch.setattr(cloud_http.json, "loads", _boom)
+        return cloud_http.http_request(target.url("assets"), target, strict_json=strict_json)
+
+    @pytest.mark.parametrize(
+        "exc",
+        [
+            pytest.param(ValueError("Exceeds the limit for integer string conversion"), id="bare-ValueError"),
+            pytest.param(RecursionError("maximum recursion depth exceeded"), id="RecursionError"),
+        ],
+    )
+    def test_strict_json_raises_response_unparseable(self, cloud_target, monkeypatch, exc):
+        from comfy_cli.command.cloud_http import ResponseUnparseable
+
+        # Not `pytest.raises(type(exc))`: the point is that the raw exception is
+        # translated, so a call site's `except ResponseUnparseable` sees it.
+        with pytest.raises(ResponseUnparseable):
+            self._request(monkeypatch, cloud_target, exc, strict_json=True)
+
+    @pytest.mark.parametrize(
+        "exc",
+        [
+            pytest.param(ValueError("Exceeds the limit for integer string conversion"), id="bare-ValueError"),
+            pytest.param(RecursionError("maximum recursion depth exceeded"), id="RecursionError"),
+        ],
+    )
+    def test_default_path_still_collapses_to_none(self, cloud_target, monkeypatch, exc):
+        # The default path's documented contract is that an undecodable body
+        # collapses to `None`; broadening the catch makes that true for these two
+        # as well, instead of letting them escape as a traceback.
+        assert self._request(monkeypatch, cloud_target, exc, strict_json=False) == (200, None)