mirror of
https://github.com/Sea-Haven-Industries/open-swe.git
synced 2026-09-30 15:03:16 +00:00
fix: optimize agent thread lists (#1570)
* fix: optimize agent thread lists Co-authored-by: open-swe[bot] <open-swe@users.noreply.github.com> * fix: refresh missing thread run status Co-authored-by: open-swe[bot] <open-swe@users.noreply.github.com> --------- Co-authored-by: open-swe[bot] <open-swe@users.noreply.github.com>
This commit is contained in:
Johannes du Plessis
GitHub
parent
4030001ebf
commit
39a26e16b5
9 changed files with 542 additions and 206 deletions
|
|
@ -133,6 +133,7 @@ from .thread_api import (
|
|||
get_dashboard_thread_state,
|
||||
list_dashboard_threads,
|
||||
list_dashboard_threads_page,
|
||||
list_dashboard_threads_sidebar,
|
||||
proxy_dashboard_thread_commands,
|
||||
proxy_dashboard_thread_history,
|
||||
proxy_dashboard_thread_run_cancel,
|
||||
|
|
@ -1183,6 +1184,24 @@ async def api_list_threads(
|
|||
return await list_dashboard_threads(session["sub"], email=session.get("email"), include_all=all)
|
||||
|
||||
|
||||
@router.get("/threads/sidebar")
|
||||
async def api_list_threads_sidebar(
|
||||
active_limit: int = 50,
|
||||
resolved_limit: int = 20,
|
||||
all: bool = False,
|
||||
session: dict[str, Any] = _SESSION_DEP,
|
||||
) -> dict[str, Any]:
|
||||
if all and not is_admin(session.get("email")):
|
||||
raise HTTPException(403, "admin only")
|
||||
return await list_dashboard_threads_sidebar(
|
||||
session["sub"],
|
||||
email=session.get("email"),
|
||||
active_limit=active_limit,
|
||||
resolved_limit=resolved_limit,
|
||||
include_all=all,
|
||||
)
|
||||
|
||||
|
||||
@router.get("/threads/page")
|
||||
async def api_list_threads_page(
|
||||
limit: int = 25,
|
||||
|
|
|
|||
|
|
@ -2,6 +2,7 @@
|
|||
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
import base64
|
||||
import binascii
|
||||
import json
|
||||
|
|
@ -434,58 +435,71 @@ async def _refresh_latest_run_metadata(
|
|||
return thread, latest_run_status, latest_run_id
|
||||
|
||||
|
||||
async def list_dashboard_threads(
|
||||
login: str, *, email: str | None = None, limit: int = 50, include_all: bool = False
|
||||
_THREADS_SEARCH_PAGE = 500
|
||||
_THREADS_PAGE_SCAN_CAP = 5000
|
||||
_THREAD_LIST_SELECT = ["thread_id", "status", "metadata", "updated_at"]
|
||||
_RUN_REFRESH_CONCURRENCY = 8
|
||||
_RUNNING_METADATA_STATUSES = {"pending", "running"}
|
||||
|
||||
|
||||
def _thread_id(thread: dict[str, Any]) -> str | None:
|
||||
thread_id = thread.get("thread_id") or thread.get("id")
|
||||
return thread_id if isinstance(thread_id, str) and thread_id else None
|
||||
|
||||
|
||||
def _thread_metadata(thread: dict[str, Any]) -> dict[str, Any]:
|
||||
return thread.get("metadata") if isinstance(thread.get("metadata"), dict) else {}
|
||||
|
||||
|
||||
def _owner_search_filters(
|
||||
login: str, *, email: str | None = None, include_all: bool = False
|
||||
) -> list[dict[str, Any]]:
|
||||
client = langgraph_client()
|
||||
searches: list[dict[str, Any]] = [{}] if include_all else [{"github_login": login}]
|
||||
if not include_all and email and email.strip():
|
||||
if include_all:
|
||||
return [{}]
|
||||
searches = [{"github_login": login}]
|
||||
if email and email.strip():
|
||||
searches.append({"triggering_user_email": email.strip().lower()})
|
||||
|
||||
seen: dict[str, dict[str, Any]] = {}
|
||||
for metadata_filter in searches:
|
||||
threads = await client.threads.search(
|
||||
metadata=metadata_filter,
|
||||
limit=limit,
|
||||
sort_by="updated_at",
|
||||
sort_order="desc",
|
||||
)
|
||||
for thread in threads or []:
|
||||
if not isinstance(thread, dict):
|
||||
continue
|
||||
meta = thread.get("metadata") if isinstance(thread.get("metadata"), dict) else {}
|
||||
if not include_all and not _user_owns_thread(meta, login, email):
|
||||
continue
|
||||
thread_id = thread.get("thread_id") or thread.get("id")
|
||||
if isinstance(thread_id, str) and thread_id not in seen:
|
||||
seen[thread_id] = thread
|
||||
|
||||
summaries: list[dict[str, Any]] = []
|
||||
for thread in seen.values():
|
||||
refreshed, latest_run_status, latest_run_id = await _refresh_latest_run_metadata(
|
||||
client, thread
|
||||
)
|
||||
summaries.append(
|
||||
_thread_summary(
|
||||
refreshed,
|
||||
latest_run_status=latest_run_status,
|
||||
latest_run_id=latest_run_id,
|
||||
)
|
||||
)
|
||||
summaries.sort(key=lambda item: item.get("updatedAt", 0), reverse=True)
|
||||
return summaries[:limit]
|
||||
return searches
|
||||
|
||||
|
||||
# Threads are paged out of `client.threads.search` in batches of this size,
|
||||
# scanning up to `_THREADS_PAGE_SCAN_CAP` so matches older than a single batch
|
||||
# are still found (the page is the "show all"/search surface).
|
||||
_THREADS_SEARCH_PAGE = 100
|
||||
_THREADS_PAGE_SCAN_CAP = 2000
|
||||
def _search_metadata_filter(
|
||||
owner_filter: dict[str, Any], *, resolved: bool | None = None, source: str | None = None
|
||||
) -> dict[str, Any]:
|
||||
metadata = dict(owner_filter)
|
||||
if resolved is True:
|
||||
metadata["resolved"] = True
|
||||
if source and source != _DASHBOARD_SOURCE:
|
||||
metadata["source"] = source
|
||||
return metadata
|
||||
|
||||
|
||||
def _thread_updated_ms(metadata: dict[str, Any]) -> int:
|
||||
async def _search_threads_batch(
|
||||
client: Any, metadata: dict[str, Any], *, limit: int, offset: int
|
||||
) -> list[dict[str, Any]]:
|
||||
batch = await client.threads.search(
|
||||
metadata=metadata,
|
||||
limit=limit,
|
||||
offset=offset,
|
||||
sort_by="updated_at",
|
||||
sort_order="desc",
|
||||
select=_THREAD_LIST_SELECT,
|
||||
)
|
||||
return [thread for thread in batch or [] if isinstance(thread, dict)]
|
||||
|
||||
|
||||
def _thread_updated_ms(thread: dict[str, Any]) -> int:
|
||||
metadata = _thread_metadata(thread)
|
||||
value = metadata.get("updated_at_ms")
|
||||
return int(value) if isinstance(value, (int, float)) else 0
|
||||
if isinstance(value, (int, float)):
|
||||
return int(value)
|
||||
updated_at = thread.get("updated_at")
|
||||
if isinstance(updated_at, str) and updated_at:
|
||||
try:
|
||||
parsed = datetime.fromisoformat(updated_at.replace("Z", "+00:00"))
|
||||
except ValueError:
|
||||
return 0
|
||||
return int(parsed.timestamp() * 1000)
|
||||
return 0
|
||||
|
||||
|
||||
def _metadata_matches_filters(
|
||||
|
|
@ -532,50 +546,210 @@ def _summary_matches_filters(
|
|||
return True
|
||||
|
||||
|
||||
async def _gather_candidate_threads(
|
||||
def _should_refresh_latest_run(thread: dict[str, Any]) -> bool:
|
||||
metadata = _thread_metadata(thread)
|
||||
metadata_status = metadata.get("latest_run_status")
|
||||
thread_status = thread.get("status")
|
||||
return (
|
||||
thread_status == "busy"
|
||||
or metadata_status in _RUNNING_METADATA_STATUSES
|
||||
or not isinstance(metadata_status, str)
|
||||
)
|
||||
|
||||
|
||||
async def _summarize_thread(
|
||||
client: Any,
|
||||
thread: dict[str, Any],
|
||||
*,
|
||||
owner_login: str | None = None,
|
||||
owner_email: str | None = None,
|
||||
refresh_active_run: bool = True,
|
||||
) -> dict[str, Any]:
|
||||
latest_run_status = latest_run_id = None
|
||||
if refresh_active_run and _should_refresh_latest_run(thread):
|
||||
thread, latest_run_status, latest_run_id = await _refresh_latest_run_metadata(
|
||||
client, thread
|
||||
)
|
||||
return _thread_summary(
|
||||
thread,
|
||||
latest_run_status=latest_run_status,
|
||||
latest_run_id=latest_run_id,
|
||||
owner_login=owner_login,
|
||||
owner_email=owner_email,
|
||||
)
|
||||
|
||||
|
||||
async def _summarize_threads(
|
||||
client: Any,
|
||||
threads: list[dict[str, Any]],
|
||||
*,
|
||||
owner_login: str | None = None,
|
||||
owner_email: str | None = None,
|
||||
) -> list[dict[str, Any]]:
|
||||
semaphore = asyncio.Semaphore(_RUN_REFRESH_CONCURRENCY)
|
||||
|
||||
async def summarize(thread: dict[str, Any]) -> dict[str, Any]:
|
||||
if not _should_refresh_latest_run(thread):
|
||||
return await _summarize_thread(
|
||||
client,
|
||||
thread,
|
||||
owner_login=owner_login,
|
||||
owner_email=owner_email,
|
||||
refresh_active_run=False,
|
||||
)
|
||||
async with semaphore:
|
||||
return await _summarize_thread(
|
||||
client,
|
||||
thread,
|
||||
owner_login=owner_login,
|
||||
owner_email=owner_email,
|
||||
)
|
||||
|
||||
return list(await asyncio.gather(*(summarize(thread) for thread in threads)))
|
||||
|
||||
|
||||
async def _collect_thread_candidates(
|
||||
client: Any,
|
||||
searches: list[dict[str, Any]],
|
||||
*,
|
||||
include_all: bool,
|
||||
login: str,
|
||||
email: str | None,
|
||||
resolved: bool | None = None,
|
||||
source: str | None = None,
|
||||
query: str | None = None,
|
||||
target_per_search: int | None = None,
|
||||
) -> list[dict[str, Any]]:
|
||||
"""Page through `threads.search` (up to the scan cap) and dedupe by id."""
|
||||
seen: dict[str, dict[str, Any]] = {}
|
||||
for metadata_filter in searches:
|
||||
for owner_filter in searches:
|
||||
matched_for_search = 0
|
||||
offset = 0
|
||||
metadata_filter = _search_metadata_filter(owner_filter, resolved=resolved, source=source)
|
||||
while offset < _THREADS_PAGE_SCAN_CAP:
|
||||
batch = await client.threads.search(
|
||||
metadata=metadata_filter,
|
||||
batch = await _search_threads_batch(
|
||||
client,
|
||||
metadata_filter,
|
||||
limit=_THREADS_SEARCH_PAGE,
|
||||
offset=offset,
|
||||
sort_by="updated_at",
|
||||
sort_order="desc",
|
||||
)
|
||||
if not batch:
|
||||
break
|
||||
for thread in batch:
|
||||
if not isinstance(thread, dict):
|
||||
metadata = _thread_metadata(thread)
|
||||
if not include_all and not _user_owns_thread(metadata, login, email):
|
||||
continue
|
||||
meta = thread.get("metadata") if isinstance(thread.get("metadata"), dict) else {}
|
||||
if not include_all and not _user_owns_thread(meta, login, email):
|
||||
if not _metadata_matches_filters(
|
||||
metadata,
|
||||
resolved=resolved,
|
||||
source=source,
|
||||
query=query,
|
||||
):
|
||||
continue
|
||||
thread_id = thread.get("thread_id") or thread.get("id")
|
||||
if isinstance(thread_id, str) and thread_id not in seen:
|
||||
seen[thread_id] = thread
|
||||
thread_id = _thread_id(thread)
|
||||
if not thread_id:
|
||||
continue
|
||||
matched_for_search += 1
|
||||
seen.setdefault(thread_id, thread)
|
||||
if len(batch) < _THREADS_SEARCH_PAGE:
|
||||
break
|
||||
if target_per_search is not None and matched_for_search >= target_per_search:
|
||||
break
|
||||
offset += _THREADS_SEARCH_PAGE
|
||||
return sorted(seen.values(), key=_thread_updated_ms, reverse=True)
|
||||
|
||||
|
||||
async def list_dashboard_threads(
|
||||
login: str, *, email: str | None = None, limit: int = 50, include_all: bool = False
|
||||
) -> list[dict[str, Any]]:
|
||||
page = await list_dashboard_threads_page(
|
||||
login,
|
||||
email=email,
|
||||
limit=limit,
|
||||
offset=0,
|
||||
include_all=include_all,
|
||||
)
|
||||
return page["items"]
|
||||
|
||||
|
||||
async def list_dashboard_threads_sidebar(
|
||||
login: str,
|
||||
*,
|
||||
email: str | None = None,
|
||||
active_limit: int = 50,
|
||||
resolved_limit: int = 20,
|
||||
include_all: bool = False,
|
||||
) -> dict[str, Any]:
|
||||
client = langgraph_client()
|
||||
searches = _owner_search_filters(login, email=email, include_all=include_all)
|
||||
safe_active_limit = min(max(active_limit, 1), 100)
|
||||
safe_resolved_limit = min(max(resolved_limit, 1), 100)
|
||||
active_target = safe_active_limit + 1
|
||||
resolved_target = safe_resolved_limit + 1
|
||||
active: dict[str, dict[str, Any]] = {}
|
||||
resolved_threads: dict[str, dict[str, Any]] = {}
|
||||
|
||||
for owner_filter in searches:
|
||||
local_active = 0
|
||||
local_resolved = 0
|
||||
offset = 0
|
||||
while offset < _THREADS_PAGE_SCAN_CAP and (
|
||||
local_active < active_target or local_resolved < resolved_target
|
||||
):
|
||||
batch = await _search_threads_batch(
|
||||
client,
|
||||
owner_filter,
|
||||
limit=_THREADS_SEARCH_PAGE,
|
||||
offset=offset,
|
||||
)
|
||||
if not batch:
|
||||
break
|
||||
for thread in batch:
|
||||
metadata = _thread_metadata(thread)
|
||||
if not include_all and not _user_owns_thread(metadata, login, email):
|
||||
continue
|
||||
thread_id = _thread_id(thread)
|
||||
if not thread_id or thread_id in active or thread_id in resolved_threads:
|
||||
continue
|
||||
if _is_thread_resolved(metadata):
|
||||
local_resolved += 1
|
||||
resolved_threads[thread_id] = thread
|
||||
else:
|
||||
local_active += 1
|
||||
active[thread_id] = thread
|
||||
if len(batch) < _THREADS_SEARCH_PAGE:
|
||||
break
|
||||
offset += _THREADS_SEARCH_PAGE
|
||||
return list(seen.values())
|
||||
|
||||
|
||||
async def _summarize_thread(client: Any, thread: dict[str, Any]) -> dict[str, Any]:
|
||||
refreshed, latest_run_status, latest_run_id = await _refresh_latest_run_metadata(client, thread)
|
||||
return _thread_summary(
|
||||
refreshed,
|
||||
latest_run_status=latest_run_status,
|
||||
latest_run_id=latest_run_id,
|
||||
active_candidates = sorted(active.values(), key=_thread_updated_ms, reverse=True)
|
||||
resolved_candidates = sorted(resolved_threads.values(), key=_thread_updated_ms, reverse=True)
|
||||
active_window = active_candidates[:safe_active_limit]
|
||||
resolved_window = resolved_candidates[:safe_resolved_limit]
|
||||
active_items, resolved_items = await asyncio.gather(
|
||||
_summarize_threads(
|
||||
client,
|
||||
active_window,
|
||||
owner_login=None if include_all else login,
|
||||
owner_email=None if include_all else email,
|
||||
),
|
||||
_summarize_threads(
|
||||
client,
|
||||
resolved_window,
|
||||
owner_login=None if include_all else login,
|
||||
owner_email=None if include_all else email,
|
||||
),
|
||||
)
|
||||
return {
|
||||
"active": {
|
||||
"items": active_items,
|
||||
"limit": safe_active_limit,
|
||||
"hasMore": len(active_candidates) > safe_active_limit,
|
||||
},
|
||||
"resolved": {
|
||||
"items": resolved_items,
|
||||
"limit": safe_resolved_limit,
|
||||
"hasMore": len(resolved_candidates) > safe_resolved_limit,
|
||||
},
|
||||
}
|
||||
|
||||
|
||||
async def list_dashboard_threads_page(
|
||||
|
|
@ -591,63 +765,58 @@ async def list_dashboard_threads_page(
|
|||
status: str | None = None,
|
||||
query: str | None = None,
|
||||
) -> dict[str, Any]:
|
||||
"""Paginated + filterable thread list for the full threads page."""
|
||||
client = langgraph_client()
|
||||
searches: list[dict[str, Any]] = [{}] if include_all else [{"github_login": login}]
|
||||
if not include_all and email and email.strip():
|
||||
searches.append({"triggering_user_email": email.strip().lower()})
|
||||
|
||||
candidates = await _gather_candidate_threads(
|
||||
client, searches, include_all=include_all, login=login, email=email
|
||||
)
|
||||
|
||||
# Apply metadata-only filters first so the frequent sidebar polls only fetch
|
||||
# the latest run for threads that can actually match.
|
||||
matched = [
|
||||
thread
|
||||
for thread in candidates
|
||||
if _metadata_matches_filters(
|
||||
thread.get("metadata") if isinstance(thread.get("metadata"), dict) else {},
|
||||
resolved=resolved,
|
||||
source=source,
|
||||
query=query,
|
||||
)
|
||||
]
|
||||
matched.sort(
|
||||
key=lambda thread: _thread_updated_ms(
|
||||
thread.get("metadata") if isinstance(thread.get("metadata"), dict) else {}
|
||||
),
|
||||
reverse=True,
|
||||
)
|
||||
|
||||
searches = _owner_search_filters(login, email=email, include_all=include_all)
|
||||
safe_offset = max(offset, 0)
|
||||
safe_limit = max(limit, 1)
|
||||
safe_limit = min(max(limit, 1), 100)
|
||||
summary_filters = viewed is not None or status is not None
|
||||
target = None if summary_filters else safe_offset + safe_limit + 1
|
||||
|
||||
# `viewed`/`status` derive from the latest run, so they can only be applied
|
||||
# after enrichment. When neither is requested we enrich just the page window.
|
||||
if viewed is None and status is None:
|
||||
total = len(matched)
|
||||
window = matched[safe_offset : safe_offset + safe_limit]
|
||||
items = [await _summarize_thread(client, thread) for thread in window]
|
||||
return {"items": items, "total": total, "limit": safe_limit, "offset": safe_offset}
|
||||
candidates = await _collect_thread_candidates(
|
||||
client,
|
||||
searches,
|
||||
include_all=include_all,
|
||||
login=login,
|
||||
email=email,
|
||||
resolved=resolved,
|
||||
source=source,
|
||||
query=query,
|
||||
target_per_search=target,
|
||||
)
|
||||
|
||||
summaries: list[dict[str, Any]] = []
|
||||
for thread in matched:
|
||||
summary = await _summarize_thread(client, thread)
|
||||
if _summary_matches_filters(
|
||||
summary,
|
||||
resolved=resolved,
|
||||
viewed=viewed,
|
||||
source=source,
|
||||
status=status,
|
||||
query=query,
|
||||
):
|
||||
summaries.append(summary)
|
||||
if summary_filters:
|
||||
summaries = await _summarize_threads(
|
||||
client,
|
||||
candidates,
|
||||
owner_login=None if include_all else login,
|
||||
owner_email=None if include_all else email,
|
||||
)
|
||||
filtered = [
|
||||
summary
|
||||
for summary in summaries
|
||||
if _summary_matches_filters(
|
||||
summary,
|
||||
resolved=resolved,
|
||||
viewed=viewed,
|
||||
source=source,
|
||||
status=status,
|
||||
query=query,
|
||||
)
|
||||
]
|
||||
filtered.sort(key=lambda item: item.get("updatedAt", 0), reverse=True)
|
||||
items = filtered[safe_offset : safe_offset + safe_limit]
|
||||
has_more = len(filtered) > safe_offset + safe_limit
|
||||
else:
|
||||
window = candidates[safe_offset : safe_offset + safe_limit]
|
||||
items = await _summarize_threads(
|
||||
client,
|
||||
window,
|
||||
owner_login=None if include_all else login,
|
||||
owner_email=None if include_all else email,
|
||||
)
|
||||
has_more = len(candidates) > safe_offset + safe_limit
|
||||
|
||||
summaries.sort(key=lambda item: item.get("updatedAt", 0), reverse=True)
|
||||
total = len(summaries)
|
||||
items = summaries[safe_offset : safe_offset + safe_limit]
|
||||
return {"items": items, "total": total, "limit": safe_limit, "offset": safe_offset}
|
||||
return {"items": items, "limit": safe_limit, "offset": safe_offset, "hasMore": has_more}
|
||||
|
||||
|
||||
async def _mark_thread_viewed(
|
||||
|
|
|
|||
|
|
@ -801,14 +801,54 @@ def _make_threads(count: int, *, resolved_before: int) -> list[dict[str, object]
|
|||
|
||||
|
||||
async def test_list_dashboard_threads_page_pages_beyond_first_search_batch(monkeypatch) -> None:
|
||||
# The 100 most-recent threads are resolved; the unresolved ones only appear
|
||||
# in the second search batch (offset >= 100).
|
||||
threads = _make_threads(150, resolved_before=100)
|
||||
page_size = thread_api._THREADS_SEARCH_PAGE
|
||||
threads = _make_threads(page_size + 50, resolved_before=page_size)
|
||||
for thread in threads:
|
||||
thread["metadata"]["latest_run_status"] = "success"
|
||||
offsets: list[int] = []
|
||||
run_list_calls = 0
|
||||
|
||||
class FakeThreads:
|
||||
async def search(self, *, metadata, limit, offset, sort_by, sort_order):
|
||||
async def search(self, *, metadata, limit, offset, sort_by, sort_order, select):
|
||||
offsets.append(offset)
|
||||
assert select == thread_api._THREAD_LIST_SELECT
|
||||
return threads[offset : offset + limit]
|
||||
|
||||
async def update(self, *, thread_id, metadata):
|
||||
return None
|
||||
|
||||
class FakeRuns:
|
||||
async def list(self, thread_id, limit=1):
|
||||
nonlocal run_list_calls
|
||||
run_list_calls += 1
|
||||
return []
|
||||
|
||||
class FakeClient:
|
||||
threads = FakeThreads()
|
||||
runs = FakeRuns()
|
||||
|
||||
monkeypatch.setattr(thread_api, "langgraph_client", lambda: FakeClient())
|
||||
|
||||
result = await thread_api.list_dashboard_threads_page(
|
||||
"octocat", email=None, limit=25, offset=0, resolved=False
|
||||
)
|
||||
|
||||
assert result["hasMore"] is True
|
||||
assert len(result["items"]) == 25
|
||||
assert all(item["resolved"] is False for item in result["items"])
|
||||
assert page_size in offsets
|
||||
assert run_list_calls == 0
|
||||
|
||||
|
||||
async def test_list_dashboard_threads_sidebar_fills_buckets_with_one_endpoint(monkeypatch) -> None:
|
||||
page_size = thread_api._THREADS_SEARCH_PAGE
|
||||
threads = _make_threads(page_size + 10, resolved_before=page_size)
|
||||
searches: list[dict[str, object]] = []
|
||||
|
||||
class FakeThreads:
|
||||
async def search(self, *, metadata, limit, offset, sort_by, sort_order, select):
|
||||
searches.append({"metadata": metadata, "offset": offset})
|
||||
assert select == thread_api._THREAD_LIST_SELECT
|
||||
return threads[offset : offset + limit]
|
||||
|
||||
async def update(self, *, thread_id, metadata):
|
||||
|
|
@ -824,11 +864,84 @@ async def test_list_dashboard_threads_page_pages_beyond_first_search_batch(monke
|
|||
|
||||
monkeypatch.setattr(thread_api, "langgraph_client", lambda: FakeClient())
|
||||
|
||||
result = await thread_api.list_dashboard_threads_page(
|
||||
"octocat", email=None, limit=25, offset=0, resolved=False
|
||||
result = await thread_api.list_dashboard_threads_sidebar(
|
||||
"octocat", email=None, active_limit=5, resolved_limit=5
|
||||
)
|
||||
|
||||
assert result["total"] == 50
|
||||
assert len(result["items"]) == 25
|
||||
assert all(item["resolved"] is False for item in result["items"])
|
||||
assert 100 in offsets
|
||||
assert len(result["active"]["items"]) == 5
|
||||
assert len(result["resolved"]["items"]) == 5
|
||||
assert result["active"]["hasMore"] is True
|
||||
assert result["resolved"]["hasMore"] is True
|
||||
assert {call["offset"] for call in searches} == {0, page_size}
|
||||
|
||||
|
||||
async def test_list_dashboard_threads_page_refreshes_only_unsettled_threads(monkeypatch) -> None:
|
||||
threads = _make_threads(3, resolved_before=0)
|
||||
threads[0]["metadata"]["latest_run_status"] = "success"
|
||||
threads[1]["metadata"]["latest_run_status"] = "pending"
|
||||
threads[2]["metadata"]["latest_run_status"] = "error"
|
||||
run_list_thread_ids: list[str] = []
|
||||
updates: list[dict[str, object]] = []
|
||||
|
||||
class FakeThreads:
|
||||
async def search(self, *, metadata, limit, offset, sort_by, sort_order, select):
|
||||
return threads[offset : offset + limit]
|
||||
|
||||
async def update(self, *, thread_id, metadata):
|
||||
updates.append({"thread_id": thread_id, "metadata": metadata})
|
||||
|
||||
class FakeRuns:
|
||||
async def list(self, thread_id, limit=1):
|
||||
run_list_thread_ids.append(thread_id)
|
||||
return [{"id": "run-1", "status": "success"}]
|
||||
|
||||
class FakeClient:
|
||||
threads = FakeThreads()
|
||||
runs = FakeRuns()
|
||||
|
||||
monkeypatch.setattr(thread_api, "langgraph_client", lambda: FakeClient())
|
||||
|
||||
result = await thread_api.list_dashboard_threads_page("octocat", email=None, limit=3, offset=0)
|
||||
|
||||
assert run_list_thread_ids == ["t1"]
|
||||
assert updates == [
|
||||
{
|
||||
"thread_id": "t1",
|
||||
"metadata": {"latest_run_status": "success", "latest_run_id": "run-1"},
|
||||
}
|
||||
]
|
||||
assert [item["status"] for item in result["items"]] == ["finished", "finished", "error"]
|
||||
|
||||
|
||||
async def test_status_filter_refreshes_threads_missing_run_status(monkeypatch) -> None:
|
||||
threads = _make_threads(2, resolved_before=0)
|
||||
for thread in threads:
|
||||
thread["metadata"]["source"] = "slack"
|
||||
run_statuses = {"t0": "success", "t1": "error"}
|
||||
run_list_thread_ids: list[str] = []
|
||||
|
||||
class FakeThreads:
|
||||
async def search(self, *, metadata, limit, offset, sort_by, sort_order, select):
|
||||
return threads[offset : offset + limit]
|
||||
|
||||
async def update(self, *, thread_id, metadata):
|
||||
return None
|
||||
|
||||
class FakeRuns:
|
||||
async def list(self, thread_id, limit=1):
|
||||
run_list_thread_ids.append(thread_id)
|
||||
return [{"id": f"run-{thread_id}", "status": run_statuses[thread_id]}]
|
||||
|
||||
class FakeClient:
|
||||
threads = FakeThreads()
|
||||
runs = FakeRuns()
|
||||
|
||||
monkeypatch.setattr(thread_api, "langgraph_client", lambda: FakeClient())
|
||||
|
||||
result = await thread_api.list_dashboard_threads_page(
|
||||
"octocat", email=None, limit=25, offset=0, status="finished"
|
||||
)
|
||||
|
||||
assert {item["id"] for item in result["items"]} == {"t0"}
|
||||
assert result["items"][0]["status"] == "finished"
|
||||
assert set(run_list_thread_ids) == {"t0", "t1"}
|
||||
|
|
|
|||
|
|
@ -3,7 +3,7 @@ import { useStreamContext as useAgentThreadStream } from "@langchain/react"
|
|||
import { useQueryClient } from "@tanstack/react-query"
|
||||
import { useNavigate } from "@tanstack/react-router"
|
||||
|
||||
import type { AgentThread, ImageChunk } from "@/lib/agents/types"
|
||||
import type { ImageChunk } from "@/lib/agents/types"
|
||||
import type { CreateAgentThreadVariables } from "@/lib/agents/queries"
|
||||
import type { ModelSelection } from "@/lib/agents/provider/useModelOptions"
|
||||
import { AgentPromptBar } from "@/components/agents/AgentPromptBar"
|
||||
|
|
@ -13,6 +13,7 @@ import {
|
|||
agentThreadKeys,
|
||||
invalidateAgentThreadLists,
|
||||
optimisticThread,
|
||||
seedAgentThreadLists,
|
||||
} from "@/lib/agents/queries"
|
||||
import { useModelOptions } from "@/lib/agents/provider/useModelOptions"
|
||||
import { useProfile, useRepos } from "@/lib/profile"
|
||||
|
|
@ -67,12 +68,7 @@ export function AgentsHome() {
|
|||
draftRef.current = null
|
||||
const thread = optimisticThread(id, draft)
|
||||
queryClient.setQueryData(agentThreadKeys.detail(id), thread)
|
||||
// Surface the thread in the sidebar immediately; the list's running
|
||||
// refetch reconciles to server truth once the run.start stamps it.
|
||||
queryClient.setQueryData<Array<AgentThread>>(agentThreadKeys.all, (prev) => [
|
||||
thread,
|
||||
...(prev?.filter((existing) => existing.id !== id) ?? []),
|
||||
])
|
||||
seedAgentThreadLists(queryClient, thread)
|
||||
invalidateAgentThreadLists(queryClient)
|
||||
void navigate({ to: "/agents/$threadId", params: { threadId: id } })
|
||||
}, [stream.threadId, queryClient, navigate])
|
||||
|
|
@ -101,7 +97,9 @@ export function AgentsHome() {
|
|||
|
||||
stream
|
||||
.submit(
|
||||
{ messages: [{ type: "human", content: promptContent(prompt, images) }] },
|
||||
{
|
||||
messages: [{ type: "human", content: promptContent(prompt, images) }],
|
||||
},
|
||||
{ config: { configurable } }
|
||||
)
|
||||
.catch(() => {
|
||||
|
|
|
|||
|
|
@ -44,6 +44,7 @@ import {
|
|||
useSeedAgentThreadDetails,
|
||||
useSidebarThreads,
|
||||
} from "@/lib/agents/queries"
|
||||
import { useRunCompletionNotifier } from "@/lib/agents/useRunCompletionNotifier"
|
||||
import { cn } from "@/lib/utils"
|
||||
|
||||
const RESOLVED_SIDEBAR_LIMIT = 20
|
||||
|
|
@ -98,14 +99,13 @@ const NAV = [
|
|||
] as const
|
||||
|
||||
export function AgentsSidebar({ user, activeThreadId }: AgentsSidebarProps) {
|
||||
const { active, resolved } = useSidebarThreads(RESOLVED_SIDEBAR_LIMIT)
|
||||
const activeThreads = active.data?.items ?? []
|
||||
const resolvedThreads = resolved.data?.items ?? []
|
||||
const resolvedTotal = resolved.data?.total ?? resolvedThreads.length
|
||||
useSeedAgentThreadDetails(
|
||||
[...activeThreads, ...resolvedThreads],
|
||||
activeThreadId
|
||||
)
|
||||
const sidebar = useSidebarThreads(RESOLVED_SIDEBAR_LIMIT)
|
||||
const activeThreads = sidebar.data?.active.items ?? []
|
||||
const resolvedThreads = sidebar.data?.resolved.items ?? []
|
||||
const resolvedHasMore = sidebar.data?.resolved.hasMore ?? false
|
||||
const visibleThreads = [...activeThreads, ...resolvedThreads]
|
||||
useSeedAgentThreadDetails(visibleThreads, activeThreadId)
|
||||
useRunCompletionNotifier(visibleThreads, activeThreadId)
|
||||
const groups = groupThreads(activeThreads)
|
||||
const layout = useSidebarLayout()
|
||||
const reviewSidebar = useReviewSidebarData()
|
||||
|
|
@ -190,7 +190,7 @@ export function AgentsSidebar({ user, activeThreadId }: AgentsSidebarProps) {
|
|||
/>
|
||||
<ResolvedThreadGroup
|
||||
threads={resolvedThreads}
|
||||
total={resolvedTotal}
|
||||
hasMore={resolvedHasMore}
|
||||
activeThreadId={activeThreadId}
|
||||
onNavigate={layout.closeOnMobile}
|
||||
/>
|
||||
|
|
@ -249,12 +249,12 @@ function ThreadGroup({
|
|||
|
||||
function ResolvedThreadGroup({
|
||||
threads,
|
||||
total,
|
||||
hasMore,
|
||||
activeThreadId,
|
||||
onNavigate,
|
||||
}: {
|
||||
threads: Array<AgentThread>
|
||||
total: number
|
||||
hasMore: boolean
|
||||
activeThreadId?: string
|
||||
onNavigate?: () => void
|
||||
}) {
|
||||
|
|
@ -263,7 +263,6 @@ function ResolvedThreadGroup({
|
|||
|
||||
const ToggleIcon = collapsed ? CaretRightIcon : CaretDownIcon
|
||||
const visible = threads.slice(0, RESOLVED_SIDEBAR_LIMIT)
|
||||
const hasMore = total > visible.length
|
||||
|
||||
return (
|
||||
<div className="mb-3">
|
||||
|
|
@ -275,7 +274,10 @@ function ResolvedThreadGroup({
|
|||
>
|
||||
<ToggleIcon className="size-3" />
|
||||
<span className="min-w-0 flex-1 truncate">Resolved</span>
|
||||
<span>{total}</span>
|
||||
<span>
|
||||
{threads.length}
|
||||
{hasMore ? "+" : ""}
|
||||
</span>
|
||||
</button>
|
||||
{!collapsed && (
|
||||
<>
|
||||
|
|
|
|||
|
|
@ -82,8 +82,9 @@ export function AgentsThreadsPage({
|
|||
|
||||
const data = query.data
|
||||
const items = data?.items ?? []
|
||||
const total = data?.total ?? 0
|
||||
const totalPages = Math.max(1, Math.ceil(total / PAGE_SIZE))
|
||||
const hasMore = data?.hasMore ?? false
|
||||
const exactTotal = data?.total
|
||||
const end = offset + items.length
|
||||
|
||||
const update = (patch: Partial<ThreadsPageFilters>) => {
|
||||
onFiltersChange({ ...filters, ...patch, page: patch.page ?? 1 })
|
||||
|
|
@ -167,10 +168,11 @@ export function AgentsThreadsPage({
|
|||
)}
|
||||
</div>
|
||||
|
||||
{total > 0 && (
|
||||
{(items.length > 0 || filters.page > 1) && (
|
||||
<div className="mt-auto flex items-center justify-between pt-2 text-xs text-[var(--ui-text-muted)]">
|
||||
<span>
|
||||
{offset + 1}–{Math.min(offset + PAGE_SIZE, total)} of {total}
|
||||
{items.length > 0 ? `${offset + 1}–${end}` : "No results"}
|
||||
{exactTotal != null ? ` of ${exactTotal}` : hasMore ? "+" : ""}
|
||||
</span>
|
||||
<div className="flex items-center gap-2">
|
||||
<Button
|
||||
|
|
@ -182,13 +184,11 @@ export function AgentsThreadsPage({
|
|||
<CaretLeftIcon className="size-3" />
|
||||
Prev
|
||||
</Button>
|
||||
<span>
|
||||
Page {filters.page} / {totalPages}
|
||||
</span>
|
||||
<span>Page {filters.page}</span>
|
||||
<Button
|
||||
size="sm"
|
||||
variant="outline"
|
||||
disabled={filters.page >= totalPages}
|
||||
disabled={!hasMore}
|
||||
onClick={() => update({ page: filters.page + 1 })}
|
||||
>
|
||||
Next
|
||||
|
|
|
|||
|
|
@ -69,9 +69,21 @@ export interface ThreadsPageParams {
|
|||
|
||||
export interface ThreadsPage {
|
||||
items: Array<AgentThread>
|
||||
total: number
|
||||
total?: number
|
||||
limit: number
|
||||
offset: number
|
||||
hasMore?: boolean
|
||||
}
|
||||
|
||||
export interface SidebarThreadsGroup {
|
||||
items: Array<AgentThread>
|
||||
limit: number
|
||||
hasMore: boolean
|
||||
}
|
||||
|
||||
export interface SidebarThreads {
|
||||
active: SidebarThreadsGroup
|
||||
resolved: SidebarThreadsGroup
|
||||
}
|
||||
|
||||
const API_BASE = (import.meta.env.VITE_DASHBOARD_API_BASE_URL ?? "").replace(
|
||||
|
|
@ -125,9 +137,30 @@ function buildThreadsPageQuery(params: ThreadsPageParams): string {
|
|||
return query ? `?${query}` : ""
|
||||
}
|
||||
|
||||
function buildSidebarThreadsQuery(params: {
|
||||
activeLimit?: number
|
||||
resolvedLimit?: number
|
||||
}): string {
|
||||
const search = new URLSearchParams()
|
||||
if (params.activeLimit != null) {
|
||||
search.set("active_limit", String(params.activeLimit))
|
||||
}
|
||||
if (params.resolvedLimit != null) {
|
||||
search.set("resolved_limit", String(params.resolvedLimit))
|
||||
}
|
||||
const query = search.toString()
|
||||
return query ? `?${query}` : ""
|
||||
}
|
||||
|
||||
export const agentsApi = {
|
||||
langGraphApiUrl: agentsLangGraphApiUrl,
|
||||
listThreads: () => agentsRequest<Array<AgentThread>>("/threads"),
|
||||
listSidebarThreads: (params: {
|
||||
activeLimit?: number
|
||||
resolvedLimit?: number
|
||||
}) =>
|
||||
agentsRequest<SidebarThreads>(
|
||||
`/threads/sidebar${buildSidebarThreadsQuery(params)}`
|
||||
),
|
||||
listThreadsPage: (params: ThreadsPageParams = {}) =>
|
||||
agentsRequest<ThreadsPage>(`/threads/page${buildThreadsPageQuery(params)}`),
|
||||
resolveThread: (threadId: string, resolved: boolean) =>
|
||||
|
|
|
|||
|
|
@ -4,12 +4,17 @@ import { useEffect } from "react"
|
|||
|
||||
import { agentsApi } from "./api"
|
||||
import type { QueryClient } from "@tanstack/react-query"
|
||||
import type { ScheduleUpdateRequest, ThreadsPageParams } from "./api"
|
||||
import type {
|
||||
ScheduleUpdateRequest,
|
||||
SidebarThreads,
|
||||
ThreadsPageParams,
|
||||
} from "./api"
|
||||
import type { AgentThread, Chunk, ImageChunk, Message } from "./types"
|
||||
|
||||
export const agentThreadKeys = {
|
||||
lists: ["agent-threads", "lists"] as const,
|
||||
all: ["agent-threads", "lists", "all"] as const,
|
||||
sidebar: (params: { activeLimit: number; resolvedLimit: number }) =>
|
||||
["agent-threads", "lists", "sidebar", params] as const,
|
||||
detail: (threadId: string) => ["agent-threads", threadId] as const,
|
||||
prDiff: (threadId: string) => ["agent-threads", threadId, "pr-diff"] as const,
|
||||
page: (params: ThreadsPageParams) =>
|
||||
|
|
@ -20,14 +25,37 @@ export function invalidateAgentThreadLists(queryClient: QueryClient): void {
|
|||
void queryClient.invalidateQueries({ queryKey: agentThreadKeys.lists })
|
||||
}
|
||||
|
||||
export function seedAgentThreadLists(
|
||||
queryClient: QueryClient,
|
||||
thread: AgentThread
|
||||
): void {
|
||||
queryClient.setQueriesData<SidebarThreads>(
|
||||
{ queryKey: ["agent-threads", "lists", "sidebar"] },
|
||||
(prev) => {
|
||||
if (!prev) return prev
|
||||
const activeItems = [
|
||||
thread,
|
||||
...prev.active.items.filter((item) => item.id !== thread.id),
|
||||
].slice(0, prev.active.limit)
|
||||
const resolvedItems = prev.resolved.items.filter(
|
||||
(item) => item.id !== thread.id
|
||||
)
|
||||
return {
|
||||
...prev,
|
||||
active: { ...prev.active, items: activeItems },
|
||||
resolved: { ...prev.resolved, items: resolvedItems },
|
||||
}
|
||||
}
|
||||
)
|
||||
}
|
||||
|
||||
export const agentScheduleKeys = {
|
||||
all: ["agent-schedules"] as const,
|
||||
}
|
||||
|
||||
// The list endpoint (`GET /threads`) and the detail endpoint
|
||||
// (`GET /threads/{id}`) return the same per-thread summary, so warming the
|
||||
// detail cache from the already-fetched list avoids a fan-out of one request
|
||||
// per sidebar thread. Navigation stays instant; the real (mark-viewed) fetch
|
||||
// Sidebar lists and detail reads return the same per-thread summary, so warming
|
||||
// the detail cache from the already-fetched sidebar avoids a fan-out of one
|
||||
// request per thread. Navigation stays instant; the real (mark-viewed) fetch
|
||||
// fires only when a thread is actually opened. The active thread is skipped so
|
||||
// its live detail query stays the source of truth.
|
||||
export function useSeedAgentThreadDetails(
|
||||
|
|
@ -49,52 +77,31 @@ export function useSeedAgentThreadDetails(
|
|||
}, [activeThreadId, queryClient, threads])
|
||||
}
|
||||
|
||||
export function useAgentThreads() {
|
||||
return useQuery({
|
||||
queryKey: agentThreadKeys.all,
|
||||
queryFn: () => agentsApi.listThreads(),
|
||||
refetchInterval: (query) =>
|
||||
query.state.data?.some((thread) => thread.status === "running")
|
||||
? 2000
|
||||
: false,
|
||||
})
|
||||
}
|
||||
|
||||
// The sidebar fetches active (unresolved) and resolved threads separately so
|
||||
// resolving the most-recent threads can't hide older active ones behind a
|
||||
// shared cap — each list is filled server-side from its own filtered query.
|
||||
const SIDEBAR_ACTIVE_LIMIT = 50
|
||||
|
||||
function sidebarRefetchInterval(query: {
|
||||
state: { data?: { items: Array<AgentThread> } }
|
||||
}) {
|
||||
return query.state.data?.items.some((thread) => thread.status === "running")
|
||||
function sidebarThreads(data?: SidebarThreads): Array<AgentThread> {
|
||||
return [...(data?.active.items ?? []), ...(data?.resolved.items ?? [])]
|
||||
}
|
||||
|
||||
function sidebarRefetchInterval(query: { state: { data?: SidebarThreads } }) {
|
||||
return sidebarThreads(query.state.data).some(
|
||||
(thread) => thread.status === "running"
|
||||
)
|
||||
? 2000
|
||||
: false
|
||||
}
|
||||
|
||||
export function useSidebarThreads(resolvedLimit: number) {
|
||||
const active = useQuery({
|
||||
queryKey: agentThreadKeys.page({
|
||||
resolved: false,
|
||||
limit: SIDEBAR_ACTIVE_LIMIT,
|
||||
}),
|
||||
queryFn: () =>
|
||||
agentsApi.listThreadsPage({
|
||||
resolved: false,
|
||||
limit: SIDEBAR_ACTIVE_LIMIT,
|
||||
}),
|
||||
const params = {
|
||||
activeLimit: SIDEBAR_ACTIVE_LIMIT,
|
||||
resolvedLimit,
|
||||
}
|
||||
return useQuery({
|
||||
queryKey: agentThreadKeys.sidebar(params),
|
||||
queryFn: () => agentsApi.listSidebarThreads(params),
|
||||
refetchInterval: sidebarRefetchInterval,
|
||||
placeholderData: (prev) => prev,
|
||||
})
|
||||
const resolved = useQuery({
|
||||
queryKey: agentThreadKeys.page({ resolved: true, limit: resolvedLimit }),
|
||||
queryFn: () =>
|
||||
agentsApi.listThreadsPage({ resolved: true, limit: resolvedLimit }),
|
||||
refetchInterval: sidebarRefetchInterval,
|
||||
placeholderData: (prev) => prev,
|
||||
})
|
||||
return { active, resolved }
|
||||
}
|
||||
|
||||
export function useAgentThread(threadId: string) {
|
||||
|
|
|
|||
|
|
@ -9,8 +9,6 @@ import { AgentsShell } from "@/components/agents/AgentsSidebar"
|
|||
import { Skeleton } from "@/components/ui/skeleton"
|
||||
import agentsCss from "@/styles/agents.css?url"
|
||||
import { AgentThreadStreamProvider } from "@/lib/agents/AgentThreadStreamProvider"
|
||||
import { useAgentThreads } from "@/lib/agents/queries"
|
||||
import { useRunCompletionNotifier } from "@/lib/agents/useRunCompletionNotifier"
|
||||
import { useSession } from "@/lib/session"
|
||||
|
||||
export const Route = createFileRoute("/agents")({
|
||||
|
|
@ -35,9 +33,6 @@ function AgentsLayout() {
|
|||
? threadId
|
||||
: undefined
|
||||
|
||||
const threadsQuery = useAgentThreads()
|
||||
useRunCompletionNotifier(threadsQuery.data, activeThreadId)
|
||||
|
||||
if (session.isLoading) {
|
||||
return (
|
||||
<main className="agents-ui flex h-svh items-center justify-center bg-[var(--ui-bg)] p-6">
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue