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.
This commit is contained in:
wutongyuonce 2026-09-09 14:56:31 +08:00 committed by GitHub
parent 05432f4b43
commit 3c7d3303d3
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
12 changed files with 564 additions and 18 deletions

View File

@ -118,6 +118,11 @@ This section accumulates work toward the **2.1.0** milestone
#### Agents & runtime #### 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, - **middleware:** New `TokenBudgetMiddleware` enforces a per-run token budget,
shared additively across the lead agent and subagents. ([#3412]) shared additively across the lead agent and subagents. ([#3412])
- **middleware:** Structured tool-result metadata and a tool-progress state - **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 [#5278]: https://github.com/bytedance/deer-flow/pull/5278
[#5280]: https://github.com/bytedance/deer-flow/pull/5280 [#5280]: https://github.com/bytedance/deer-flow/pull/5280
[#5281]: https://github.com/bytedance/deer-flow/pull/5281 [#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 [#5284]: https://github.com/bytedance/deer-flow/pull/5284
[#5287]: https://github.com/bytedance/deer-flow/pull/5287 [#5287]: https://github.com/bytedance/deer-flow/pull/5287

View File

@ -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). | | **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 | | **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)`) | | **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 | | **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 | | **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. | | **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. |

View File

@ -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.authz.sandbox_authz import safe_app_config_async
from deerflow.config.paths import get_paths, make_safe_user_id 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 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.secret_context import redact_config_secrets, redact_metadata_secrets
from deerflow.runtime.user_context import get_effective_user_id 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 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 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): class ArtifactArchiveManifestResponse(BaseModel):
file_count: int file_count: int
@ -1047,16 +1055,67 @@ async def wait_run(
return {"status": record.status.value, "error": record.error} 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]) @router.get("/{thread_id}/runs", response_model=list[RunResponse])
@require_permission("runs", "read", owner_check=True) @require_permission("runs", "read", owner_check=True)
async def list_runs(thread_id: ThreadId, request: Request) -> list[RunResponse]: 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) run_mgr = get_run_manager(request)
user_id = await get_current_user(request) user_id = await get_current_user(request)
records = await run_mgr.list_by_thread(thread_id, user_id=user_id) records = await run_mgr.list_by_thread(thread_id, user_id=user_id)
return [_record_to_response(r) for r in records] 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) @router.get("/{thread_id}/runs/{run_id}", response_model=RunResponse)
@require_permission("runs", "read", owner_check=True) @require_permission("runs", "read", owner_check=True)
async def get_run(thread_id: ThreadId, run_id: str, request: Request) -> RunResponse: async def get_run(thread_id: ThreadId, run_id: str, request: Request) -> RunResponse:

View File

@ -11,7 +11,7 @@ import json
from datetime import UTC, datetime, timedelta from datetime import UTC, datetime, timedelta
from typing import Any 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.exc import IntegrityError
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
@ -21,6 +21,7 @@ from deerflow.runtime.runs.store.base import (
RunIdempotencyConflict, RunIdempotencyConflict,
RunStore, RunStore,
StatusFinalization, StatusFinalization,
normalize_run_created_at_iso,
) )
from deerflow.runtime.user_context import AUTO, _AutoSentinel, resolve_user_id from deerflow.runtime.user_context import AUTO, _AutoSentinel, resolve_user_id
from deerflow.utils.time import coerce_iso from deerflow.utils.time import coerce_iso
@ -168,12 +169,30 @@ class RunRepository(RunStore):
*, *,
user_id: str | None | _AutoSentinel = AUTO, user_id: str | None | _AutoSentinel = AUTO,
limit=100, 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") 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") stmt = select(RunRow).where(RunRow.thread_id == thread_id, RunRow.operation_kind == "run")
if resolved_user_id is not None: if resolved_user_id is not None:
stmt = stmt.where(RunRow.user_id == resolved_user_id) 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: async with self._sf() as session:
result = await session.execute(stmt) result = await session.execute(stmt)
return [self._row_to_dict(r) for r in result.scalars()] return [self._row_to_dict(r) for r in result.scalars()]

View File

@ -21,7 +21,13 @@ from deerflow.utils.time import is_lease_expired
from deerflow.utils.time import now_iso as _now_iso from deerflow.utils.time import now_iso as _now_iso
from .schemas import DisconnectMode, RunStatus, ThreadOperationKind 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: if TYPE_CHECKING:
from deerflow.config.run_ownership_config import RunOwnershipConfig from deerflow.config.run_ownership_config import RunOwnershipConfig
@ -57,6 +63,14 @@ def _generate_worker_id() -> str:
return f"{socket.gethostname()}:{uuid.uuid4().hex}" 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: def _is_unique_violation(exc: BaseException) -> bool:
"""Return True when *exc* (or its cause chain) is a unique-constraint violation. """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, 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. """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 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 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: Args:
thread_id: The thread ID to filter by. thread_id: The thread ID to filter by.
user_id: Optional user ID for permission filtering when hydrating from store. user_id: Optional user ID for permission filtering when hydrating from store.
limit: Maximum number of runs to return. 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: 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: 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} records_by_id = {record.run_id: record for record in memory_records}
# Query enough rows to cover both the requested page and every possible # Query enough rows to cover both the requested page and every possible
# in-memory/store duplicate. Local records can be older than persisted # 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 # newest run before the merge; querying only ``limit`` can still lose a
# distinct row when that page is occupied by duplicate local records. # distinct row when that page is occupied by duplicate local records.
store_limit = limit + len(memory_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: 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: except Exception:
logger.warning("Failed to hydrate runs for thread %s from store", thread_id, exc_info=True) 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: for row in rows:
run_id = row.get("run_id") run_id = row.get("run_id")
if run_id and run_id not in records_by_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) records_by_id[run_id] = self._record_from_store(row)
except Exception: except Exception:
logger.warning("Failed to map store row for run %s", run_id, exc_info=True) 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( async def list_successful_regenerate_sources(
self, self,

View File

@ -12,8 +12,11 @@ from __future__ import annotations
import abc import abc
from dataclasses import dataclass, field from dataclasses import dataclass, field
from datetime import UTC, datetime
from typing import Any from typing import Any
from deerflow.utils.time import coerce_iso
@dataclass(frozen=True) @dataclass(frozen=True)
class EditReplayVisibility: class EditReplayVisibility:
@ -49,6 +52,63 @@ class RunIdempotencyConflict(RuntimeError):
self.existing = existing 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): class RunStore(abc.ABC):
@abc.abstractmethod @abc.abstractmethod
async def put( async def put(
@ -89,6 +149,8 @@ class RunStore(abc.ABC):
*, *,
user_id: str | None = None, user_id: str | None = None,
limit: int = 100, limit: int = 100,
before_created_at: str | None = None,
before_run_id: str | None = None,
) -> list[dict[str, Any]]: ) -> list[dict[str, Any]]:
pass pass

View File

@ -8,7 +8,14 @@ from __future__ import annotations
from datetime import UTC, datetime, timedelta from datetime import UTC, datetime, timedelta
from typing import Any 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): class MemoryRunStore(RunStore):
@ -87,15 +94,35 @@ class MemoryRunStore(RunStore):
return None return None
return run 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 # 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 # scanning every run. ``self._runs.get`` is defense-in-depth: it drops a
# stale id still in the index but already gone from ``_runs``. # stale id still in the index but already gone from ``_runs``.
run_ids = self._runs_by_thread.get(thread_id) run_ids = self._runs_by_thread.get(thread_id)
if not run_ids: if not run_ids:
return [] 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 = [
results.sort(key=lambda r: r["created_at"], reverse=True) 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] return results[:limit]
async def list_successful_regenerate_sources(self, thread_id, *, user_id=None): async def list_successful_regenerate_sources(self, thread_id, *, user_id=None):

View File

@ -204,6 +204,20 @@ class TestMemoryRunStore:
rows = await store.list_by_thread("t1", limit=2) rows = await store.list_by_thread("t1", limit=2)
assert [r["run_id"] for r in rows] == ["r4", "r3"] 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 @pytest.mark.anyio
async def test_delete_keeps_thread_index_consistent(self, store): async def test_delete_keeps_thread_index_consistent(self, store):
await store.put("r1", thread_id="t1") await store.put("r1", thread_id="t1")

View File

@ -643,14 +643,14 @@ async def test_list_by_thread(manager: RunManager, monkeypatch: pytest.MonkeyPat
@pytest.mark.anyio @pytest.mark.anyio
async def test_list_by_thread_is_stable_when_timestamps_tie(manager: RunManager, monkeypatch: pytest.MonkeyPatch): 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") monkeypatch.setattr("deerflow.runtime.runs.manager._now_iso", lambda: "2026-01-01T00:00:00+00:00")
r1 = await manager.create("thread-1") r1 = await manager.create("thread-1")
r2 = await manager.create("thread-1") r2 = await manager.create("thread-1")
runs = await manager.list_by_thread("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 @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"] 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 @pytest.mark.anyio
async def test_create_defaults(manager: RunManager): async def test_create_defaults(manager: RunManager):
"""Create with no optional args should use defaults.""" """Create with no optional args should use defaults."""

View File

@ -507,6 +507,41 @@ class TestRunRepository:
assert len(rows) == 2 assert len(rows) == 2
await _cleanup() 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 @pytest.mark.anyio
async def test_owner_none_returns_all(self, tmp_path): async def test_owner_none_returns_all(self, tmp_path):
repo = await _make_repo(tmp_path) repo = await _make_repo(tmp_path)

View File

@ -35,6 +35,8 @@ def _make_app():
("/api/threads/thread-1/messages", -1), ("/api/threads/thread-1/messages", -1),
("/api/threads/thread-1/runs/run-1/events", 0), ("/api/threads/thread-1/runs/run-1/events", 0),
("/api/threads/thread-1/runs/run-1/events", -1), ("/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): 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, limit=1,
after_seq=None, 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

View File

@ -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,
}