From 3c7d3303d3ef9335b6d91b49acff1a6f6609936c Mon Sep 17 00:00:00 2001
From: wutongyuonce <147830929+wutongyuonce@users.noreply.github.com>
Date: Wed, 9 Sep 2026 14:56:31 +0800
Subject: [PATCH] feat(gateway): paginate thread run history (#5283)

* feat(gateway): paginate thread run history (#5282)

GET /api/threads/{thread_id}/runs stays a bare array of the newest 100
runs so LangGraph SDK clients keep working. Add GET /runs/page with a
(created_at, run_id) keyset cursor so callers can walk older history.

* fix(gateway): reject one-sided run history cursors

RunManager.list_by_thread now raises if only one of before_created_at
or before_run_id is set, matching the HTTP 422. Document the per-page
sort cost on the SQL keyset query, and add the missing CHANGELOG
[#5282] link definition.

* fix(gateway): round-trip run page cursors through query strings

Emit next_before_created_at with a Z suffix so '+' is not decoded as a
space. Accept that space, and Z, when parsing. Treat blank cursor
fields as absent and reject a non-ISO before_created_at in RunManager
so a harness caller cannot silently restart at the newest page.

* style(gateway): ruff-format run page cursor files

Collapse the one-sided cursor ValueError and the two before_created_at
asserts so ruff format --check passes at line-length 240.
---
 CHANGELOG.md                                  |   6 +
 backend/app/gateway/AGENTS.md                 |   2 +-
 backend/app/gateway/routers/thread_runs.py    |  61 +++++++-
 .../harness/deerflow/persistence/run/sql.py   |  23 ++-
 .../harness/deerflow/runtime/runs/manager.py  |  70 ++++++++-
 .../deerflow/runtime/runs/store/base.py       |  62 ++++++++
 .../deerflow/runtime/runs/store/memory.py     |  35 ++++-
 backend/tests/test_persistence_scaffold.py    |  14 ++
 backend/tests/test_run_manager.py             | 106 ++++++++++++-
 backend/tests/test_run_repository.py          |  35 +++++
 .../tests/test_thread_run_query_validation.py |  20 +++
 backend/tests/test_thread_runs_page.py        | 148 ++++++++++++++++++
 12 files changed, 564 insertions(+), 18 deletions(-)
 create mode 100644 backend/tests/test_thread_runs_page.py

diff --git a/CHANGELOG.md b/CHANGELOG.md
index e16606344..63881901b 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -118,6 +118,11 @@ This section accumulates work toward the **2.1.0** milestone
 
 #### Agents & runtime
 
+- **gateway:** `GET /api/threads/{thread_id}/runs/page` walks thread run history
+  with a `(created_at, run_id)` keyset cursor (`{data, has_more,
+  next_before_created_at, next_before_run_id}`). `GET /api/threads/{thread_id}/runs`
+  still returns a bare array of the newest 100 runs so LangGraph SDK clients keep
+  working. ([#5282])
 - **middleware:** New `TokenBudgetMiddleware` enforces a per-run token budget,
   shared additively across the lead agent and subagents. ([#3412])
 - **middleware:** Structured tool-result metadata and a tool-progress state
@@ -2670,5 +2675,6 @@ with **180 merged pull requests** since the first 2.0 milestone tag.
 [#5278]: https://github.com/bytedance/deer-flow/pull/5278
 [#5280]: https://github.com/bytedance/deer-flow/pull/5280
 [#5281]: https://github.com/bytedance/deer-flow/pull/5281
+[#5282]: https://github.com/bytedance/deer-flow/pull/5282
 [#5284]: https://github.com/bytedance/deer-flow/pull/5284
 [#5287]: https://github.com/bytedance/deer-flow/pull/5287
diff --git a/backend/app/gateway/AGENTS.md b/backend/app/gateway/AGENTS.md
index 1eada7a8b..eb624ac15 100644
--- a/backend/app/gateway/AGENTS.md
+++ b/backend/app/gateway/AGENTS.md
@@ -65,7 +65,7 @@ owner-scoped assistant version selection remains enabled.
 | **Artifacts** (`/api/threads/{id}/artifacts`) | `GET /{path}` - stream regular text and binary artifacts with `FileResponse`, including byte-`Range` 206/416 behavior used by bounded text previews and media seeking; active content types (`text/html`, `application/xhtml+xml`, `image/svg+xml`) are always forced as download attachments to reduce XSS risk; `?download=true` still forces download for other file types. `PUT /{path}` atomically replaces an existing UTF-8 text file under `/mnt/user-data/outputs` when its expected SHA-256 still matches; active runs conflict, and non-mounted sandbox providers receive the same update under a request lease. Atomic replacement applies the existing POSIX permission handling when descriptor-based APIs are available and otherwise keeps the platform-native temporary-file permissions (Windows). |
 | **Suggestions** (`/api/suggestions`) | `GET /config` - returns global suggestions config boolean; `POST /threads/{id}/suggestions` - generate follow-up questions; rich list/block model content is normalized and inline reasoning (`<think>...</think>`, including unclosed/truncated blocks from reasoning models like MiniMax-M3) is stripped before JSON parsing |
 | **Input Polish** (`/api/input-polish`) | `POST /` - rewrite a composer draft before it is sent. This is a short authenticated `runs:create` LLM request using `input_polish` config; it does not create a LangGraph run, persist a message, or modify thread state. Shares the non-graph one-shot LLM path (`deerflow.utils.oneshot_llm.run_oneshot_llm`) with the suggestions route so model build + Langfuse metadata + invoke stay in one place; validates the same stripped view of the draft it sends to the model, and preserves literal `<think>` substrings in the rewrite (`strip_think_blocks(truncate_unclosed=False)`) |
-| **Thread Runs** (`/api/threads/{id}/runs`) | `POST /` - create background run; `POST /stream` - create + SSE stream; `POST /wait` - create + block. Before the first journaled run, seed an empty feed from a checkpoint so legacy checkpoint-only history keeps its order and visibility; skip absent checkpoints or populated feeds. `POST /regenerate/prepare` - prepare clean input + checkpoint metadata for regenerating the latest completed or interrupted assistant answer, carrying the latest non-empty thread title in graph input so resuming an older checkpoint cannot roll back a later manual rename (#4457); `POST /edit-regenerate/prepare` - prepare a checkpoint replay from the latest editable human turn with a replacement user message and edit replay metadata; it carries the current thread title the same way, but only when the replay base already has one — an untitled base belongs to a thread the title middleware has not named yet, so pinning the current title there would keep a name generated from the prompt the edit just replaced; `GET /` - list runs; `GET /{rid}` - run details; `POST /{rid}/cancel` - cancel; `GET /{rid}/join` - join SSE; `GET /{rid}/stream` hides action/wait; GET action 405 pre-owner; POST needs `runs:cancel`; `GET /{rid}/messages` - paginated per-run messages `{data, has_more}`; `GET /{rid}/events` - full event stream; `GET /{rid}/workspace-changes` - workspace/output file change summary and optional diffs; `GET/POST /{rid}/artifacts/archive` - receipt manifest / bounded ZIP; `GET /../messages` - legacy thread message array; `GET /../messages/page` - backward thread-global `seq` history page with middleware/subagent-AI/successful-regenerate/edit-replay filtering and page-run-scoped feedback enrichment; subagent AI callbacks remain available through run events while parent `task` ToolMessages stay visible for card restoration; `GET /../token-usage` - aggregate tokens plus an optional `context_usage` percentage. Context usage approximately counts messages from the latest materialized thread state through `build_thread_checkpoint_state_accessor`, so full and delta checkpoint modes expose the same input. The percentage uses the latest run's model and its `context_window`. |
+| **Thread Runs** (`/api/threads/{id}/runs`) | `POST /` - create background run; `POST /stream` - create + SSE stream; `POST /wait` - create + block. Before the first journaled run, seed an empty feed from a checkpoint so legacy checkpoint-only history keeps its order and visibility; skip absent checkpoints or populated feeds. `POST /regenerate/prepare` - prepare clean input + checkpoint metadata for regenerating the latest completed or interrupted assistant answer, carrying the latest non-empty thread title in graph input so resuming an older checkpoint cannot roll back a later manual rename (#4457); `POST /edit-regenerate/prepare` - prepare a checkpoint replay from the latest editable human turn with a replacement user message and edit replay metadata; it carries the current thread title the same way, but only when the replay base already has one — an untitled base belongs to a thread the title middleware has not named yet, so pinning the current title there would keep a name generated from the prompt the edit just replaced; `GET /` - newest 100 runs as an array; `GET /page` - keyset history page `{data, has_more, next_before_created_at, next_before_run_id}`; `GET /{rid}` - run details; `POST /{rid}/cancel` - cancel; `GET /{rid}/join` - join SSE; `GET /{rid}/stream` hides action/wait; GET action 405 pre-owner; POST needs `runs:cancel`; `GET /{rid}/messages` - paginated per-run messages `{data, has_more}`; `GET /{rid}/events` - full event stream; `GET /{rid}/workspace-changes` - workspace/output file change summary and optional diffs; `GET/POST /{rid}/artifacts/archive` - receipt manifest / bounded ZIP; `GET /../messages` - legacy thread message array; `GET /../messages/page` - backward thread-global `seq` history page with middleware/subagent-AI/successful-regenerate/edit-replay filtering and page-run-scoped feedback enrichment; subagent AI callbacks remain available through run events while parent `task` ToolMessages stay visible for card restoration; `GET /../token-usage` - aggregate tokens plus an optional `context_usage` percentage. Context usage approximately counts messages from the latest materialized thread state through `build_thread_checkpoint_state_accessor`, so full and delta checkpoint modes expose the same input. The percentage uses the latest run's model and its `context_window`. |
 | **Feedback** (`/api/threads/{id}/runs/{rid}/feedback`) | `PUT /` - upsert feedback; `DELETE /` - delete user feedback; `POST /` - create feedback; `GET /` - list feedback; `GET /stats` - aggregate stats; `DELETE /{fid}` - delete specific |
 | **Runs** (`/api/runs`) | `POST /stream`, `/wait` - stateless runs requiring `runs:create`; optional body `thread_id` is owner-checked. Scheduled-task create/update/resume/trigger also require `threads:write` plus `runs:create`. `GET /{rid}/messages`, `/feedback` - run messages/feedback |
 | **GitHub Webhooks** (`/api/webhooks/github`) | `POST /` - receive GitHub App / repo webhook deliveries. Verifies `X-Hub-Signature-256` against `GITHUB_WEBHOOK_SECRET`; exempt from auth + CSRF because authenticity is enforced by HMAC. The route is fail-closed: mounted only when `GITHUB_WEBHOOK_SECRET` is set, or when explicit dev opt-in `DEER_FLOW_ALLOW_UNVERIFIED_GITHUB_WEBHOOKS=1` is set. Recognized events include `ping`, `issues`, `issue_comment`, `pull_request`, `pull_request_review`, and `pull_request_review_comment`; unknown events return 200 with `handled=false`. Fan-out runtime failures return 503, keeping the delivery recorded as failed for manual/API/scripted redelivery (GitHub does not automatically retry any failed delivery, 5xx included); permanent/non-retryable conditions such as `channels.github.enabled: false`, unknown events, malformed payloads, or unavailable channel service return 200 with a skipped/handled response. |
diff --git a/backend/app/gateway/routers/thread_runs.py b/backend/app/gateway/routers/thread_runs.py
index bda521fa7..399b9e6fb 100644
--- a/backend/app/gateway/routers/thread_runs.py
+++ b/backend/app/gateway/routers/thread_runs.py
@@ -48,6 +48,7 @@ from deerflow.agents.middlewares.dynamic_context_middleware import strip_injecte
 from deerflow.authz.sandbox_authz import safe_app_config_async
 from deerflow.config.paths import get_paths, make_safe_user_id
 from deerflow.runtime import CancelOutcome, ConflictError, RunRecord, RunStatus, ThreadOperationKind, serialize_channel_values_for_api
+from deerflow.runtime.runs.store.base import format_run_cursor_created_at, normalize_run_created_at_iso
 from deerflow.runtime.secret_context import redact_config_secrets, redact_metadata_secrets
 from deerflow.runtime.user_context import get_effective_user_id
 from deerflow.utils.messages import ORIGINAL_USER_CONTENT_KEY, get_original_user_content_text, message_to_text
@@ -231,6 +232,13 @@ class RunResponse(BaseModel):
     stop_reason: str | None = None
 
 
+class ThreadRunsPageResponse(BaseModel):
+    data: list[RunResponse]
+    has_more: bool
+    next_before_created_at: str | None = None
+    next_before_run_id: str | None = None
+
+
 class ArtifactArchiveManifestResponse(BaseModel):
     file_count: int
 
@@ -1047,16 +1055,67 @@ async def wait_run(
     return {"status": record.status.value, "error": record.error}
 
 
+def _parse_run_page_created_at(value: str) -> str:
+    try:
+        normalized = normalize_run_created_at_iso(value)
+        datetime.fromisoformat(normalized)
+    except ValueError:
+        raise HTTPException(status_code=422, detail="before_created_at must be an ISO-8601 timestamp") from None
+    return normalized
+
+
 @router.get("/{thread_id}/runs", response_model=list[RunResponse])
 @require_permission("runs", "read", owner_check=True)
 async def list_runs(thread_id: ThreadId, request: Request) -> list[RunResponse]:
-    """List all runs for a thread."""
+    """List the newest runs for a thread (default 100, as a bare array)."""
     run_mgr = get_run_manager(request)
     user_id = await get_current_user(request)
     records = await run_mgr.list_by_thread(thread_id, user_id=user_id)
     return [_record_to_response(r) for r in records]
 
 
+@router.get("/{thread_id}/runs/page", response_model=ThreadRunsPageResponse)
+@require_permission("runs", "read", owner_check=True)
+async def list_runs_page(
+    thread_id: ThreadId,
+    request: Request,
+    limit: int = Query(default=50, ge=1, le=200),
+    before_created_at: str | None = Query(default=None),
+    before_run_id: str | None = Query(default=None, min_length=1),
+) -> ThreadRunsPageResponse:
+    """Return a newest-first keyset page of runs for a thread.
+
+    Response: { data: [...], has_more: bool, next_before_created_at, next_before_run_id }
+    Pass both cursor fields from the previous page's last row to continue.
+    """
+    if (before_created_at is None) != (before_run_id is None):
+        raise HTTPException(
+            status_code=422,
+            detail="before_created_at and before_run_id must be provided together",
+        )
+    if before_created_at is not None:
+        before_created_at = _parse_run_page_created_at(before_created_at)
+
+    run_mgr = get_run_manager(request)
+    user_id = await get_current_user(request)
+    records = await run_mgr.list_by_thread(
+        thread_id,
+        user_id=user_id,
+        limit=limit + 1,
+        before_created_at=before_created_at,
+        before_run_id=before_run_id,
+    )
+    has_more = len(records) > limit
+    page = records[:limit]
+    last = page[-1] if page and has_more else None
+    return ThreadRunsPageResponse(
+        data=[_record_to_response(record) for record in page],
+        has_more=has_more,
+        next_before_created_at=format_run_cursor_created_at(last.created_at) if last else None,
+        next_before_run_id=last.run_id if last else None,
+    )
+
+
 @router.get("/{thread_id}/runs/{run_id}", response_model=RunResponse)
 @require_permission("runs", "read", owner_check=True)
 async def get_run(thread_id: ThreadId, run_id: str, request: Request) -> RunResponse:
diff --git a/backend/packages/harness/deerflow/persistence/run/sql.py b/backend/packages/harness/deerflow/persistence/run/sql.py
index 8f8613b0c..4b9df58cd 100644
--- a/backend/packages/harness/deerflow/persistence/run/sql.py
+++ b/backend/packages/harness/deerflow/persistence/run/sql.py
@@ -11,7 +11,7 @@ import json
 from datetime import UTC, datetime, timedelta
 from typing import Any
 
-from sqlalchemy import case, or_, select, update
+from sqlalchemy import and_, case, or_, select, update
 from sqlalchemy.exc import IntegrityError
 from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
 
@@ -21,6 +21,7 @@ from deerflow.runtime.runs.store.base import (
     RunIdempotencyConflict,
     RunStore,
     StatusFinalization,
+    normalize_run_created_at_iso,
 )
 from deerflow.runtime.user_context import AUTO, _AutoSentinel, resolve_user_id
 from deerflow.utils.time import coerce_iso
@@ -168,12 +169,30 @@ class RunRepository(RunStore):
         *,
         user_id: str | None | _AutoSentinel = AUTO,
         limit=100,
+        before_created_at: str | None = None,
+        before_run_id: str | None = None,
     ):
         resolved_user_id = resolve_user_id(user_id, method_name="RunRepository.list_by_thread")
         stmt = select(RunRow).where(RunRow.thread_id == thread_id, RunRow.operation_kind == "run")
         if resolved_user_id is not None:
             stmt = stmt.where(RunRow.user_id == resolved_user_id)
-        stmt = stmt.order_by(RunRow.created_at.desc()).limit(limit)
+        if before_created_at and before_run_id:
+            cursor_dt = datetime.fromisoformat(normalize_run_created_at_iso(before_created_at))
+            if cursor_dt.tzinfo is None:
+                cursor_dt = cursor_dt.replace(tzinfo=UTC)
+            else:
+                cursor_dt = cursor_dt.astimezone(UTC)
+            stmt = stmt.where(
+                or_(
+                    RunRow.created_at < cursor_dt,
+                    and_(RunRow.created_at == cursor_dt, RunRow.run_id < before_run_id),
+                )
+            )
+        # Keyset pages filter on (created_at, run_id) after thread_id. Existing
+        # indexes are (thread_id) and (thread_id, status), so each page still
+        # sorts matching rows. A covering (thread_id, created_at, run_id) index
+        # is a follow-up if deep paging shows up in profiles.
+        stmt = stmt.order_by(RunRow.created_at.desc(), RunRow.run_id.desc()).limit(limit)
         async with self._sf() as session:
             result = await session.execute(stmt)
             return [self._row_to_dict(r) for r in result.scalars()]
diff --git a/backend/packages/harness/deerflow/runtime/runs/manager.py b/backend/packages/harness/deerflow/runtime/runs/manager.py
index 38ab9fd66..3e6bc26ce 100644
--- a/backend/packages/harness/deerflow/runtime/runs/manager.py
+++ b/backend/packages/harness/deerflow/runtime/runs/manager.py
@@ -21,7 +21,13 @@ from deerflow.utils.time import is_lease_expired
 from deerflow.utils.time import now_iso as _now_iso
 
 from .schemas import DisconnectMode, RunStatus, ThreadOperationKind
-from .store.base import EditReplayVisibility, RunIdempotencyConflict
+from .store.base import (
+    EditReplayVisibility,
+    RunIdempotencyConflict,
+    normalize_run_created_at_iso,
+    run_is_before_cursor,
+    run_sort_key,
+)
 
 if TYPE_CHECKING:
     from deerflow.config.run_ownership_config import RunOwnershipConfig
@@ -57,6 +63,14 @@ def _generate_worker_id() -> str:
     return f"{socket.gethostname()}:{uuid.uuid4().hex}"
 
 
+def _cursor_part(value: str | None) -> str | None:
+    """Treat missing/blank cursor fields as absent so a one-sided empty string fails."""
+    if value is None:
+        return None
+    stripped = value.strip()
+    return stripped or None
+
+
 def _is_unique_violation(exc: BaseException) -> bool:
     """Return True when *exc* (or its cause chain) is a unique-constraint violation.
 
@@ -686,22 +700,58 @@ class RunManager:
             raise_on_store_error=raise_on_store_error,
         )
 
-    async def list_by_thread(self, thread_id: str, *, user_id: str | None = None, limit: int = 100) -> list[RunRecord]:
+    async def list_by_thread(
+        self,
+        thread_id: str,
+        *,
+        user_id: str | None = None,
+        limit: int = 100,
+        before_created_at: str | None = None,
+        before_run_id: str | None = None,
+    ) -> list[RunRecord]:
         """Return runs for a given thread, newest first, at most ``limit`` records.
 
         In-memory runs take precedence only when the same ``run_id`` exists in both
         memory and the backing store. The merged result is then sorted newest-first
-        by ``created_at`` and trimmed to ``limit`` (default 100).
+        by ``(created_at, run_id)`` and trimmed to ``limit`` (default 100).
+        Optional ``before_created_at`` + ``before_run_id`` is a keyset cursor for
+        walking older pages; both must be provided together.
 
         Args:
             thread_id: The thread ID to filter by.
             user_id: Optional user ID for permission filtering when hydrating from store.
             limit: Maximum number of runs to return.
+            before_created_at: ISO timestamp of the last run on the previous page.
+            before_run_id: Run id of the last run on the previous page.
         """
+        before_created_at = _cursor_part(before_created_at)
+        before_run_id = _cursor_part(before_run_id)
+        if (before_created_at is None) != (before_run_id is None):
+            raise ValueError("before_created_at and before_run_id must be provided together")
+        if before_created_at is not None:
+            try:
+                before_created_at = normalize_run_created_at_iso(before_created_at)
+                datetime.fromisoformat(before_created_at)
+            except ValueError:
+                raise ValueError("before_created_at must be an ISO-8601 timestamp") from None
+
+        def _page(records: list[RunRecord]) -> list[RunRecord]:
+            return sorted(records, key=lambda record: run_sort_key(record.created_at, record.run_id), reverse=True)[:limit]
+
         async with self._lock:
-            memory_records = [record for record in self._thread_records_locked(thread_id) if record.operation_kind == ThreadOperationKind.run]
+            memory_records = [
+                record
+                for record in self._thread_records_locked(thread_id)
+                if record.operation_kind == ThreadOperationKind.run
+                and run_is_before_cursor(
+                    record.created_at,
+                    record.run_id,
+                    before_created_at=before_created_at,
+                    before_run_id=before_run_id,
+                )
+            ]
         if self._store is None:
-            return sorted(memory_records, key=lambda r: r.created_at, reverse=True)[:limit]
+            return _page(memory_records)
         records_by_id = {record.run_id: record for record in memory_records}
         # Query enough rows to cover both the requested page and every possible
         # in-memory/store duplicate. Local records can be older than persisted
@@ -709,11 +759,15 @@ class RunManager:
         # newest run before the merge; querying only ``limit`` can still lose a
         # distinct row when that page is occupied by duplicate local records.
         store_limit = limit + len(memory_records)
+        store_kwargs: dict[str, Any] = {"user_id": user_id, "limit": store_limit}
+        if before_created_at is not None and before_run_id is not None:
+            store_kwargs["before_created_at"] = before_created_at
+            store_kwargs["before_run_id"] = before_run_id
         try:
-            rows = await self._store.list_by_thread(thread_id, user_id=user_id, limit=store_limit)
+            rows = await self._store.list_by_thread(thread_id, **store_kwargs)
         except Exception:
             logger.warning("Failed to hydrate runs for thread %s from store", thread_id, exc_info=True)
-            return sorted(memory_records, key=lambda r: r.created_at, reverse=True)[:limit]
+            return _page(memory_records)
         for row in rows:
             run_id = row.get("run_id")
             if run_id and run_id not in records_by_id:
@@ -721,7 +775,7 @@ class RunManager:
                     records_by_id[run_id] = self._record_from_store(row)
                 except Exception:
                     logger.warning("Failed to map store row for run %s", run_id, exc_info=True)
-        return sorted(records_by_id.values(), key=lambda record: record.created_at, reverse=True)[:limit]
+        return _page(list(records_by_id.values()))
 
     async def list_successful_regenerate_sources(
         self,
diff --git a/backend/packages/harness/deerflow/runtime/runs/store/base.py b/backend/packages/harness/deerflow/runtime/runs/store/base.py
index 06dd92bf4..9314be75f 100644
--- a/backend/packages/harness/deerflow/runtime/runs/store/base.py
+++ b/backend/packages/harness/deerflow/runtime/runs/store/base.py
@@ -12,8 +12,11 @@ from __future__ import annotations
 
 import abc
 from dataclasses import dataclass, field
+from datetime import UTC, datetime
 from typing import Any
 
+from deerflow.utils.time import coerce_iso
+
 
 @dataclass(frozen=True)
 class EditReplayVisibility:
@@ -49,6 +52,63 @@ class RunIdempotencyConflict(RuntimeError):
         self.existing = existing
 
 
+def normalize_run_created_at_iso(value: str) -> str:
+    """Make a run timestamp parseable as ISO-8601.
+
+    ``Z`` becomes ``+00:00``. An unencoded ``+`` in a query string arrives as a
+    space (``...T00:00:00 00:00``); restore the offset ``+``.
+    """
+    value = value.strip().replace("Z", "+00:00")
+    if "T" in value and " " in value and "+" not in value.split("T", 1)[1]:
+        date, _, rest = value.partition("T")
+        time_part, sep, offset = rest.rpartition(" ")
+        if sep and offset.replace(":", "").isdigit():
+            value = f"{date}T{time_part}+{offset}"
+    return value
+
+
+def format_run_cursor_created_at(value: str) -> str:
+    """UTC keyset cursor using ``Z`` so ``+`` is not decoded as space in query strings."""
+    dt = datetime.fromisoformat(normalize_run_created_at_iso(value))
+    if dt.tzinfo is None:
+        dt = dt.replace(tzinfo=UTC)
+    else:
+        dt = dt.astimezone(UTC)
+    return dt.isoformat().replace("+00:00", "Z")
+
+
+def parse_run_created_at(value: object) -> datetime:
+    """Parse a stored run timestamp into an aware UTC datetime for keyset order."""
+    iso = coerce_iso(value)
+    if not iso:
+        return datetime.min.replace(tzinfo=UTC)
+    try:
+        dt = datetime.fromisoformat(normalize_run_created_at_iso(iso))
+    except ValueError:
+        return datetime.min.replace(tzinfo=UTC)
+    if dt.tzinfo is None:
+        return dt.replace(tzinfo=UTC)
+    return dt.astimezone(UTC)
+
+
+def run_sort_key(created_at: object, run_id: str) -> tuple[datetime, str]:
+    """Total order for newest-first run listings: ``created_at`` then ``run_id``."""
+    return (parse_run_created_at(created_at), run_id)
+
+
+def run_is_before_cursor(
+    created_at: object,
+    run_id: str,
+    *,
+    before_created_at: str | None,
+    before_run_id: str | None,
+) -> bool:
+    """Return True when ``(created_at, run_id)`` is older than the keyset cursor."""
+    if not before_created_at or not before_run_id:
+        return True
+    return run_sort_key(created_at, run_id) < run_sort_key(before_created_at, before_run_id)
+
+
 class RunStore(abc.ABC):
     @abc.abstractmethod
     async def put(
@@ -89,6 +149,8 @@ class RunStore(abc.ABC):
         *,
         user_id: str | None = None,
         limit: int = 100,
+        before_created_at: str | None = None,
+        before_run_id: str | None = None,
     ) -> list[dict[str, Any]]:
         pass
 
diff --git a/backend/packages/harness/deerflow/runtime/runs/store/memory.py b/backend/packages/harness/deerflow/runtime/runs/store/memory.py
index 19b40215f..cf74a21f4 100644
--- a/backend/packages/harness/deerflow/runtime/runs/store/memory.py
+++ b/backend/packages/harness/deerflow/runtime/runs/store/memory.py
@@ -8,7 +8,14 @@ from __future__ import annotations
 from datetime import UTC, datetime, timedelta
 from typing import Any
 
-from deerflow.runtime.runs.store.base import LeaseRenewal, RunIdempotencyConflict, RunStore, StatusFinalization
+from deerflow.runtime.runs.store.base import (
+    LeaseRenewal,
+    RunIdempotencyConflict,
+    RunStore,
+    StatusFinalization,
+    run_is_before_cursor,
+    run_sort_key,
+)
 
 
 class MemoryRunStore(RunStore):
@@ -87,15 +94,35 @@ class MemoryRunStore(RunStore):
             return None
         return run
 
-    async def list_by_thread(self, thread_id, *, user_id=None, limit=100):
+    async def list_by_thread(
+        self,
+        thread_id,
+        *,
+        user_id=None,
+        limit=100,
+        before_created_at=None,
+        before_run_id=None,
+    ):
         # Use the thread index for an O(runs-in-thread) lookup instead of
         # scanning every run. ``self._runs.get`` is defense-in-depth: it drops a
         # stale id still in the index but already gone from ``_runs``.
         run_ids = self._runs_by_thread.get(thread_id)
         if not run_ids:
             return []
-        results = [run for run_id in run_ids if (run := self._runs.get(run_id)) is not None and run.get("operation_kind", "run") == "run" and (user_id is None or run.get("user_id") == user_id)]
-        results.sort(key=lambda r: r["created_at"], reverse=True)
+        results = [
+            run
+            for run_id in run_ids
+            if (run := self._runs.get(run_id)) is not None
+            and run.get("operation_kind", "run") == "run"
+            and (user_id is None or run.get("user_id") == user_id)
+            and run_is_before_cursor(
+                run.get("created_at"),
+                run["run_id"],
+                before_created_at=before_created_at,
+                before_run_id=before_run_id,
+            )
+        ]
+        results.sort(key=lambda r: run_sort_key(r.get("created_at"), r["run_id"]), reverse=True)
         return results[:limit]
 
     async def list_successful_regenerate_sources(self, thread_id, *, user_id=None):
diff --git a/backend/tests/test_persistence_scaffold.py b/backend/tests/test_persistence_scaffold.py
index 3e831e38c..77ebaa466 100644
--- a/backend/tests/test_persistence_scaffold.py
+++ b/backend/tests/test_persistence_scaffold.py
@@ -204,6 +204,20 @@ class TestMemoryRunStore:
         rows = await store.list_by_thread("t1", limit=2)
         assert [r["run_id"] for r in rows] == ["r4", "r3"]
 
+    @pytest.mark.anyio
+    async def test_list_by_thread_keyset_cursor(self, store):
+        for i in range(5):
+            await store.put(f"r{i}", thread_id="t1", created_at=f"2024-01-0{i + 1}T00:00:00+00:00")
+        first = await store.list_by_thread("t1", limit=2)
+        assert [r["run_id"] for r in first] == ["r4", "r3"]
+        second = await store.list_by_thread(
+            "t1",
+            limit=2,
+            before_created_at=first[-1]["created_at"],
+            before_run_id=first[-1]["run_id"],
+        )
+        assert [r["run_id"] for r in second] == ["r2", "r1"]
+
     @pytest.mark.anyio
     async def test_delete_keeps_thread_index_consistent(self, store):
         await store.put("r1", thread_id="t1")
diff --git a/backend/tests/test_run_manager.py b/backend/tests/test_run_manager.py
index 750017cd5..ef042e971 100644
--- a/backend/tests/test_run_manager.py
+++ b/backend/tests/test_run_manager.py
@@ -643,14 +643,14 @@ async def test_list_by_thread(manager: RunManager, monkeypatch: pytest.MonkeyPat
 
 @pytest.mark.anyio
 async def test_list_by_thread_is_stable_when_timestamps_tie(manager: RunManager, monkeypatch: pytest.MonkeyPatch):
-    """Ordering should be stable (insertion order) even when timestamps tie."""
+    """Timestamp ties break on run_id so keyset pagination has a total order."""
     monkeypatch.setattr("deerflow.runtime.runs.manager._now_iso", lambda: "2026-01-01T00:00:00+00:00")
 
     r1 = await manager.create("thread-1")
     r2 = await manager.create("thread-1")
 
     runs = await manager.list_by_thread("thread-1")
-    assert [run.run_id for run in runs] == [r1.run_id, r2.run_id]
+    assert [run.run_id for run in runs] == sorted([r1.run_id, r2.run_id], reverse=True)
 
 
 @pytest.mark.anyio
@@ -901,6 +901,108 @@ async def test_list_by_thread_limit_does_not_let_old_memory_hide_new_store_run()
     assert [run.run_id for run in runs] == ["new-store"]
 
 
+@pytest.mark.anyio
+async def test_list_by_thread_keyset_returns_older_page():
+    """A (created_at, run_id) cursor walks past the newest page."""
+    store = MemoryRunStore()
+    manager = RunManager(store=store)
+    for run_id, created_at in (
+        ("r1", "2026-01-01T00:00:00+00:00"),
+        ("r2", "2026-01-02T00:00:00+00:00"),
+        ("r3", "2026-01-03T00:00:00+00:00"),
+    ):
+        await store.put(run_id, thread_id="thread-1", status="success", created_at=created_at)
+
+    first = await manager.list_by_thread("thread-1", limit=2)
+    assert [run.run_id for run in first] == ["r3", "r2"]
+
+    second = await manager.list_by_thread(
+        "thread-1",
+        limit=2,
+        before_created_at=first[-1].created_at,
+        before_run_id=first[-1].run_id,
+    )
+    assert [run.run_id for run in second] == ["r1"]
+
+
+@pytest.mark.anyio
+async def test_list_by_thread_keyset_is_stable_when_timestamps_tie():
+    """Tied created_at values must not skip or duplicate across pages."""
+    store = MemoryRunStore()
+    manager = RunManager(store=store)
+    tied = "2026-01-01T00:00:00+00:00"
+    for run_id in ("a", "b", "c"):
+        await store.put(run_id, thread_id="thread-1", status="success", created_at=tied)
+
+    first = await manager.list_by_thread("thread-1", limit=2)
+    assert [run.run_id for run in first] == ["c", "b"]
+    second = await manager.list_by_thread(
+        "thread-1",
+        limit=2,
+        before_created_at=first[-1].created_at,
+        before_run_id=first[-1].run_id,
+    )
+    assert [run.run_id for run in second] == ["a"]
+
+
+@pytest.mark.anyio
+async def test_list_by_thread_rejects_one_sided_keyset_cursor():
+    """A one-sided cursor would silently drop the bound; fail instead of paging from the start."""
+    manager = RunManager()
+    with pytest.raises(
+        ValueError,
+        match="before_created_at and before_run_id must be provided together",
+    ):
+        await manager.list_by_thread(
+            "thread-1",
+            before_created_at="2026-01-02T00:00:00+00:00",
+        )
+    with pytest.raises(
+        ValueError,
+        match="before_created_at and before_run_id must be provided together",
+    ):
+        await manager.list_by_thread("thread-1", before_run_id="r2")
+    with pytest.raises(
+        ValueError,
+        match="before_created_at and before_run_id must be provided together",
+    ):
+        await manager.list_by_thread(
+            "thread-1",
+            before_created_at="2026-01-02T00:00:00+00:00",
+            before_run_id="",
+        )
+    with pytest.raises(
+        ValueError,
+        match="before_created_at must be an ISO-8601 timestamp",
+    ):
+        await manager.list_by_thread(
+            "thread-1",
+            before_created_at="not-a-timestamp",
+            before_run_id="r2",
+        )
+
+
+@pytest.mark.anyio
+async def test_list_by_thread_keyset_accepts_space_decoded_offset():
+    """Query-decoded '+00:00' (a space) must still walk to the older page."""
+    store = MemoryRunStore()
+    manager = RunManager(store=store)
+    for run_id, created_at in (
+        ("r1", "2026-01-01T00:00:00+00:00"),
+        ("r2", "2026-01-02T00:00:00+00:00"),
+        ("r3", "2026-01-03T00:00:00+00:00"),
+    ):
+        await store.put(run_id, thread_id="thread-1", status="success", created_at=created_at)
+
+    older = await manager.list_by_thread(
+        "thread-1",
+        limit=2,
+        before_created_at="2026-01-02T00:00:00 00:00",
+        before_run_id="r2",
+    )
+    assert [run.run_id for run in older] == ["r1"]
+
+
 @pytest.mark.anyio
 async def test_create_defaults(manager: RunManager):
     """Create with no optional args should use defaults."""
diff --git a/backend/tests/test_run_repository.py b/backend/tests/test_run_repository.py
index 52cf6b854..8c08c4a84 100644
--- a/backend/tests/test_run_repository.py
+++ b/backend/tests/test_run_repository.py
@@ -507,6 +507,41 @@ class TestRunRepository:
         assert len(rows) == 2
         await _cleanup()
 
+    @pytest.mark.anyio
+    async def test_list_by_thread_keyset_cursor(self, tmp_path):
+        repo = await _make_repo(tmp_path)
+        await repo.put("r1", thread_id="t1", status="success", created_at="2024-01-01T00:00:00+00:00")
+        await repo.put("r2", thread_id="t1", status="success", created_at="2024-01-02T00:00:00+00:00")
+        await repo.put("r3", thread_id="t1", status="success", created_at="2024-01-03T00:00:00+00:00")
+        first = await repo.list_by_thread("t1", limit=2)
+        assert [row["run_id"] for row in first] == ["r3", "r2"]
+        second = await repo.list_by_thread(
+            "t1",
+            limit=2,
+            before_created_at=first[-1]["created_at"],
+            before_run_id=first[-1]["run_id"],
+        )
+        assert [row["run_id"] for row in second] == ["r1"]
+        await _cleanup()
+
+    @pytest.mark.anyio
+    async def test_list_by_thread_keyset_breaks_timestamp_ties(self, tmp_path):
+        repo = await _make_repo(tmp_path)
+        tied = "2024-01-01T00:00:00+00:00"
+        await repo.put("a", thread_id="t1", status="success", created_at=tied)
+        await repo.put("b", thread_id="t1", status="success", created_at=tied)
+        await repo.put("c", thread_id="t1", status="success", created_at=tied)
+        first = await repo.list_by_thread("t1", limit=2)
+        assert [row["run_id"] for row in first] == ["c", "b"]
+        second = await repo.list_by_thread(
+            "t1",
+            limit=2,
+            before_created_at=first[-1]["created_at"],
+            before_run_id=first[-1]["run_id"],
+        )
+        assert [row["run_id"] for row in second] == ["a"]
+        await _cleanup()
+
     @pytest.mark.anyio
     async def test_owner_none_returns_all(self, tmp_path):
         repo = await _make_repo(tmp_path)
diff --git a/backend/tests/test_thread_run_query_validation.py b/backend/tests/test_thread_run_query_validation.py
index be3490ffd..9ff0e1b9f 100644
--- a/backend/tests/test_thread_run_query_validation.py
+++ b/backend/tests/test_thread_run_query_validation.py
@@ -35,6 +35,8 @@ def _make_app():
         ("/api/threads/thread-1/messages", -1),
         ("/api/threads/thread-1/runs/run-1/events", 0),
         ("/api/threads/thread-1/runs/run-1/events", -1),
+        ("/api/threads/thread-1/runs/page", 0),
+        ("/api/threads/thread-1/runs/page", -1),
     ],
 )
 def test_read_endpoints_reject_non_positive_limits(path: str, limit: int):
@@ -93,3 +95,21 @@ def test_read_endpoints_accept_positive_limits_and_hit_store():
         limit=1,
         after_seq=None,
     )
+
+
+@pytest.mark.parametrize("params", [{"before_created_at": "2026-01-01T00:00:00+00:00"}, {"before_run_id": "run-1"}])
+def test_runs_page_rejects_split_cursor(params: dict):
+    with TestClient(_make_app()) as client:
+        response = client.get("/api/threads/thread-1/runs/page", params=params)
+
+    assert response.status_code == 422
+
+
+def test_runs_page_rejects_invalid_created_at_cursor():
+    with TestClient(_make_app()) as client:
+        response = client.get(
+            "/api/threads/thread-1/runs/page",
+            params={"before_created_at": "not-a-timestamp", "before_run_id": "run-1"},
+        )
+
+    assert response.status_code == 422
diff --git a/backend/tests/test_thread_runs_page.py b/backend/tests/test_thread_runs_page.py
new file mode 100644
index 000000000..c92cfcb36
--- /dev/null
+++ b/backend/tests/test_thread_runs_page.py
@@ -0,0 +1,148 @@
+"""Tests for GET /api/threads/{thread_id}/runs and /runs/page."""
+
+from __future__ import annotations
+
+from unittest.mock import AsyncMock, MagicMock
+
+from _router_auth_helpers import make_authed_test_app
+from fastapi.testclient import TestClient
+
+from app.gateway.routers import thread_runs
+from deerflow.runtime import DisconnectMode, RunRecord, RunStatus
+
+
+def _record(run_id: str, created_at: str) -> RunRecord:
+    return RunRecord(
+        run_id=run_id,
+        thread_id="thread-1",
+        assistant_id=None,
+        status=RunStatus.success,
+        on_disconnect=DisconnectMode.cancel,
+        created_at=created_at,
+        updated_at=created_at,
+    )
+
+
+def _make_app(records: list[RunRecord]) -> tuple:
+    app = make_authed_test_app()
+    app.include_router(thread_runs.router)
+    run_manager = MagicMock()
+    run_manager.list_by_thread = AsyncMock(return_value=records)
+    app.state.run_manager = run_manager
+    return app, run_manager
+
+
+def test_list_runs_returns_bare_array():
+    records = [
+        _record("r3", "2026-01-03T00:00:00+00:00"),
+        _record("r2", "2026-01-02T00:00:00+00:00"),
+    ]
+    app, run_manager = _make_app(records)
+    with TestClient(app) as client:
+        response = client.get("/api/threads/thread-1/runs")
+
+    assert response.status_code == 200
+    body = response.json()
+    assert isinstance(body, list)
+    assert [row["run_id"] for row in body] == ["r3", "r2"]
+    run_manager.list_by_thread.assert_awaited_once()
+    kwargs = run_manager.list_by_thread.await_args.kwargs
+    assert kwargs.get("limit", 100) == 100
+    assert "before_created_at" not in kwargs
+    assert "before_run_id" not in kwargs
+
+
+def test_list_runs_ignores_langgraph_sdk_limit_query():
+    """SDK always sends limit=10; honoring it would shrink the existing array."""
+    records = [_record(f"r{i}", f"2026-01-{i:02d}T00:00:00+00:00") for i in range(1, 4)]
+    app, run_manager = _make_app(records)
+    with TestClient(app) as client:
+        response = client.get("/api/threads/thread-1/runs", params={"limit": 10, "offset": 0})
+
+    assert response.status_code == 200
+    assert isinstance(response.json(), list)
+    assert len(response.json()) == 3
+    assert "limit" not in run_manager.list_by_thread.await_args.kwargs
+
+
+def test_runs_page_returns_envelope_and_has_more():
+    records = [
+        _record("r3", "2026-01-03T00:00:00+00:00"),
+        _record("r2", "2026-01-02T00:00:00+00:00"),
+        _record("r1", "2026-01-01T00:00:00+00:00"),
+    ]
+    app, run_manager = _make_app(records)
+    with TestClient(app) as client:
+        response = client.get("/api/threads/thread-1/runs/page", params={"limit": 2})
+
+    assert response.status_code == 200
+    body = response.json()
+    assert [row["run_id"] for row in body["data"]] == ["r3", "r2"]
+    assert body["has_more"] is True
+    assert body["next_before_created_at"] == "2026-01-02T00:00:00Z"
+    assert body["next_before_run_id"] == "r2"
+    kwargs = run_manager.list_by_thread.await_args.kwargs
+    assert kwargs["limit"] == 3
+    assert kwargs["before_created_at"] is None
+    assert kwargs["before_run_id"] is None
+
+
+def test_runs_page_passes_cursor_through():
+    app, run_manager = _make_app([_record("r1", "2026-01-01T00:00:00+00:00")])
+    with TestClient(app) as client:
+        response = client.get(
+            "/api/threads/thread-1/runs/page",
+            params={
+                "limit": 2,
+                "before_created_at": "2026-01-02T00:00:00+00:00",
+                "before_run_id": "r2",
+            },
+        )
+
+    assert response.status_code == 200
+    body = response.json()
+    assert body["has_more"] is False
+    assert body["next_before_created_at"] is None
+    assert body["next_before_run_id"] is None
+    kwargs = run_manager.list_by_thread.await_args.kwargs
+    assert kwargs["before_created_at"] == "2026-01-02T00:00:00+00:00"
+    assert kwargs["before_run_id"] == "r2"
+
+
+def test_runs_page_accepts_z_and_space_decoded_offset():
+    """Unencoded +00:00 arrives as a space; Z cursors must round-trip too."""
+    app, run_manager = _make_app([_record("r1", "2026-01-01T00:00:00+00:00")])
+    with TestClient(app) as client:
+        spaced = client.get(
+            "/api/threads/thread-1/runs/page",
+            params={
+                "before_created_at": "2026-01-02T00:00:00 00:00",
+                "before_run_id": "r2",
+            },
+        )
+        zoned = client.get(
+            "/api/threads/thread-1/runs/page",
+            params={
+                "before_created_at": "2026-01-02T00:00:00Z",
+                "before_run_id": "r2",
+            },
+        )
+
+    assert spaced.status_code == 200
+    assert zoned.status_code == 200
+    assert run_manager.list_by_thread.await_args_list[0].kwargs["before_created_at"] == ("2026-01-02T00:00:00+00:00")
+    assert run_manager.list_by_thread.await_args_list[1].kwargs["before_created_at"] == ("2026-01-02T00:00:00+00:00")
+
+
+def test_runs_page_is_not_captured_as_run_id():
+    app, _run_manager = _make_app([])
+    with TestClient(app) as client:
+        response = client.get("/api/threads/thread-1/runs/page")
+
+    assert response.status_code == 200
+    assert response.json() == {
+        "data": [],
+        "has_more": False,
+        "next_before_created_at": None,
+        "next_before_run_id": None,
+    }