diff --git a/src/dualentry_cli/client.py b/src/dualentry_cli/client.py
index 280f9c9..da16788 100644
--- a/src/dualentry_cli/client.py
+++ b/src/dualentry_cli/client.py
@@ -33,6 +33,10 @@
 
 _MAX_RETRY_AFTER = 60
 
+# Hard ceiling for --all crawls: page_size 100 x 1000 pages = 100_000 items.
+# Truncation must warn; never report the truncated length as the API total.
+_MAX_PAGES = 1000
+
 
 def _retry_after_seconds(response: httpx.Response) -> int | None:
     """Seconds from the Retry-After header, or None if absent or unusable."""
@@ -206,27 +210,44 @@ def _request(self, method: str, path: str, **kwargs) -> dict:
     def get(self, path: str, params: dict[str, Any] | None = None) -> dict:
         return self._request("GET", path, params=params)
 
-    def paginate(self, path: str, params: dict[str, Any] | None = None, page_size: int = 100, max_items: int | None = None) -> dict:
-        """Fetch all pages and return combined {items: [...], count: N}."""
+    def paginate(
+        self,
+        path: str,
+        params: dict[str, Any] | None = None,
+        page_size: int = 100,
+        max_items: int | None = None,
+        *,
+        start_offset: int = 0,
+    ) -> dict:
+        """
+        Fetch pages and return {items, count, next_offset?}.
+
+        ``count`` is always the API total, regardless of truncation.
+        When the page ceiling (or ``max_items``) stops the crawl early,
+        ``next_offset`` is set so callers can resume.
+        """
         params = dict(params or {})
         params["limit"] = page_size
-        params["offset"] = 0
-        all_items = []
-        max_pages = 1000
+        params["offset"] = start_offset
+        all_items: list = []
+        total = 0
 
-        for _ in range(max_pages):
+        for _ in range(_MAX_PAGES):
             data = self.get(path, params=params)
             items = data.get("items", [])
             all_items.extend(items)
-            total = data.get("count", len(items))
+            total = data.get("count", start_offset + len(all_items))
             if max_items and len(all_items) >= max_items:
                 all_items = all_items[:max_items]
                 break
-            if len(all_items) >= total or not items:
+            if start_offset + len(all_items) >= total or not items:
                 break
             params["offset"] += page_size
 
-        return {"items": all_items, "count": len(all_items)}
+        result: dict[str, Any] = {"items": all_items, "count": total}
+        if start_offset + len(all_items) < total:
+            result["next_offset"] = start_offset + len(all_items)
+        return result
 
     def post(self, path: str, json: dict[str, Any] | None = None) -> dict:
         return self._request("POST", path, json=json)
diff --git a/src/dualentry_cli/commands/__init__.py b/src/dualentry_cli/commands/__init__.py
index 41f6db1..cd4c215 100644
--- a/src/dualentry_cli/commands/__init__.py
+++ b/src/dualentry_cli/commands/__init__.py
@@ -9,6 +9,7 @@
 import typer
 
 from dualentry_cli.cli import HelpfulGroup
+from dualentry_cli.client import _MAX_PAGES
 from dualentry_cli.output import _RECORD_PREFIX, format_output
 
 # ── Shared option defaults ──────────────────────────────────────────
@@ -72,15 +73,58 @@ def _build_filter_params(
     return params
 
 
+# Map _do_list filter kwargs to CLI flags for the --all resume hint.
+_FILTER_CLI_FLAGS = {
+    "search": "--search",
+    "status": "--status",
+    "start_date": "--start-date",
+    "end_date": "--end-date",
+    "company_id": "--company",
+    "customer_id": "--customer",
+    "vendor_id": "--vendor",
+}
+
+
+def _resume_all_command(path: str, next_offset: int, filters: dict) -> str:
+    """Build a copy-paste dualentry list --all command that continues from next_offset."""
+    parts = ["dualentry", *path.split("/"), "list", "--all", "--offset", str(next_offset)]
+    for key, flag in _FILTER_CLI_FLAGS.items():
+        value = filters.get(key)
+        if value is not None:
+            parts.extend([flag, str(value)])
+    return " ".join(parts)
+
+
+def _warn_all_truncated(path: str, *, fetched_through: int, total: int, next_offset: int, filters: dict) -> None:
+    """Tell the user --all stopped early and how to continue."""
+    cmd = _resume_all_command(path, next_offset, filters)
+    typer.secho(
+        f"Warning: reached {fetched_through} of {total} items; stopped at the {_MAX_PAGES}-page limit.\nTo continue, re-run with the same filters:\n  {cmd}",
+        fg=typer.colors.YELLOW,
+        err=True,
+    )
+
+
 def _do_list(client, path: str, resource: str, *, limit: int, offset: int, all_pages: bool, output: str, **filters):
     """Shared list logic for all resources."""
     params = _build_filter_params(**filters)
+    next_offset = None
     if all_pages:
-        data = client.paginate(f"/{path}/", params=params)
+        data = client.paginate(f"/{path}/", params=params, start_offset=offset)
+        next_offset = data.pop("next_offset", None)
     else:
         params.update({"limit": limit, "offset": offset})
         data = client.get(f"/{path}/", params=params)
     format_output(data, resource=resource, fmt=output)
+    # After the table so the resume hint is visible without scrolling up.
+    if next_offset is not None:
+        _warn_all_truncated(
+            path,
+            fetched_through=next_offset,
+            total=data.get("count", next_offset),
+            next_offset=next_offset,
+            filters=filters,
+        )
 
 
 def _load_json_file(file: Path) -> dict:
diff --git a/tests/test_client.py b/tests/test_client.py
index 18974a7..aa552fa 100644
--- a/tests/test_client.py
+++ b/tests/test_client.py
@@ -501,3 +501,65 @@ def test_backoff_table_and_retry_count_cannot_drift(self):
         from dualentry_cli.client import _MAX_RETRIES, _RETRY_DELAYS
 
         assert len(_RETRY_DELAYS) == _MAX_RETRIES
+
+
+class TestPaginate:
+    BASE = "https://api.dualentry.com/public/v2"
+
+    @staticmethod
+    def _client():
+        from dualentry_cli.client import DualEntryClient
+
+        return DualEntryClient(api_url="https://api.dualentry.com", api_key="test_key")
+
+    @respx.mock
+    def test_complete_crawl_has_no_next_offset(self):
+        respx.get(f"{self.BASE}/invoices/").mock(
+            side_effect=[
+                httpx.Response(200, json={"items": [{"id": 1}, {"id": 2}], "count": 3}),
+                httpx.Response(200, json={"items": [{"id": 3}], "count": 3}),
+            ]
+        )
+
+        data = self._client().paginate("/invoices/", page_size=2)
+
+        assert data["items"] == [{"id": 1}, {"id": 2}, {"id": 3}]
+        assert data["count"] == 3
+        assert "next_offset" not in data
+
+    @respx.mock
+    def test_page_cap_sets_next_offset_and_keeps_api_count(self, monkeypatch):
+        monkeypatch.setattr("dualentry_cli.client._MAX_PAGES", 2)
+
+        def _page(request: httpx.Request) -> httpx.Response:
+            offset = int(request.url.params.get("offset", "0"))
+            return httpx.Response(200, json={"items": [{"id": offset}, {"id": offset + 1}], "count": 10})
+
+        respx.get(f"{self.BASE}/invoices/").mock(side_effect=_page)
+
+        data = self._client().paginate("/invoices/", page_size=2)
+
+        assert len(data["items"]) == 4
+        assert data["count"] == 10
+        assert data["next_offset"] == 4
+
+    @respx.mock
+    def test_start_offset_is_sent_on_first_request(self):
+        route = respx.get(f"{self.BASE}/invoices/").mock(return_value=httpx.Response(200, json={"items": [{"id": 5}], "count": 5}))
+
+        data = self._client().paginate("/invoices/", page_size=2, start_offset=4)
+
+        assert data["items"] == [{"id": 5}]
+        assert data["count"] == 5
+        assert "next_offset" not in data
+        assert route.calls[0].request.url.params["offset"] == "4"
+
+    @respx.mock
+    def test_max_items_truncation_sets_next_offset(self):
+        respx.get(f"{self.BASE}/invoices/").mock(return_value=httpx.Response(200, json={"items": [{"id": 1}, {"id": 2}, {"id": 3}], "count": 9}))
+
+        data = self._client().paginate("/invoices/", page_size=3, max_items=2)
+
+        assert data["items"] == [{"id": 1}, {"id": 2}]
+        assert data["count"] == 9
+        assert data["next_offset"] == 2
diff --git a/tests/test_commands.py b/tests/test_commands.py
index 40447ba..e7abf22 100644
--- a/tests/test_commands.py
+++ b/tests/test_commands.py
@@ -207,3 +207,47 @@ def test_unmatchable_command_still_shows_help(self):
         assert result.exit_code == 2
         assert "Unknown command 'zzzzzz'" in result.output
         assert "Did you mean" not in result.output
+
+
+class TestListAllTruncation:
+    def test_all_passes_offset_to_paginate(self, mock_get_client):
+        mock_get_client.paginate.return_value = {"items": [], "count": 0}
+        result = runner.invoke(app, ["invoices", "list", "--all", "--offset", "100"])
+        assert result.exit_code == 0
+        mock_get_client.paginate.assert_called_once_with("/invoices/", params={}, start_offset=100)
+
+    def test_all_truncation_prints_resume_command(self, mock_get_client):
+        mock_get_client.paginate.return_value = {
+            "items": [{"internal_id": 1, "number": 1}],
+            "count": 250000,
+            "next_offset": 100000,
+        }
+        result = runner.invoke(app, ["invoices", "list", "--all", "--search", "acme", "--company", "9"])
+        assert result.exit_code == 0
+        assert "reached 100000 of 250000" in result.output
+        assert "dualentry invoices list --all --offset 100000" in result.output
+        assert "--search acme" in result.output
+        assert "--company 9" in result.output
+        # Warning must come after the list so users see it without scrolling up.
+        assert result.output.index("Showing") < result.output.index("reached 100000 of 250000")
+        mock_get_client.paginate.assert_called_once_with(
+            "/invoices/",
+            params={"search": "acme", "company_id": "9"},
+            start_offset=0,
+        )
+
+    def test_resume_command_helper_for_nested_path(self):
+        from dualentry_cli.commands import _resume_all_command
+
+        cmd = _resume_all_command("recurring/invoices", 200, {"status": "posted"})
+        assert cmd == "dualentry recurring invoices list --all --offset 200 --status posted"
+
+    def test_complete_all_does_not_warn(self, mock_get_client):
+        mock_get_client.paginate.return_value = {
+            "items": [{"internal_id": 1, "number": 1}],
+            "count": 1,
+        }
+        result = runner.invoke(app, ["invoices", "list", "--all"])
+        assert result.exit_code == 0
+        assert "Warning" not in result.output
+        assert "stopped at the" not in result.output