Compare commits

...
Author SHA1 Message Date
kristinheandCursor 946024bf5c style(memhub): format lite pagination changes
Co-authored-by: Cursor <cursoragent@cursor.com>
2026-11-05 00:48:46 +08:00
kristinheandCursor 675f7ac9b0 fix(memhub): page Lite raw turns without full scans
Co-authored-by: Cursor <cursoragent@cursor.com>
2026-11-04 23:53:40 +08:00
15 changed files with 713 additions and 103 deletions

No files matched your search

+195 -1
View File
@@ -644,6 +644,9 @@ class SdkRawPage(BaseModel):
offset: int = Field(ge=0)
limit: int = Field(ge=1, le=500)
unit: Literal["raw", "turn"] = "raw"
next_cursor: str | None = None
has_more: bool = False
total_exact: bool = True
#: Set when a time filter was applied in Core. For storage `raw`, `total`
#: remains the SDK count; for product `turn`, it is the filtered turn count.
time_filtered: bool = False
@@ -4491,6 +4494,8 @@ def create_app(
offset: int = Query(default=0, ge=0),
limit: int = Query(default=100, ge=1, le=500),
unit: Literal["raw", "turn"] = Query(default="raw"),
paging: Literal["offset", "cursor"] = Query(default="offset"),
cursor: str | None = Query(default=None, max_length=160),
user_id: str = Query(default=DEFAULT_USER_ID, pattern=USER_ID_PATTERN),
exclude_sources: list[str] | None = Query(default=None), # noqa: B008
session_id: list[str] | None = Query(default=None), # noqa: B008
@@ -4516,9 +4521,52 @@ def create_app(
_library_values(statuses or [], MEMORY_STATUSES, "memory_status_unknown")
or raw_list_filters()["statuses"]
)
if paging == "cursor" and unit != "turn":
raise HTTPException(status_code=422, detail="raw_cursor_requires_turn_unit")
if cursor is not None and paging != "cursor":
raise HTTPException(status_code=422, detail="raw_cursor_requires_cursor_paging")
if paging == "cursor" and (offset or after is not None or before is not None):
raise HTTPException(status_code=422, detail="raw_cursor_filter_unsupported")
raw_cursor = (0, 0, 0, 0, "", "")
if cursor:
parts = cursor.split(":")
if (
len(parts) != 7
or parts[0] != "turn"
or any(re.fullmatch(r"[0-9]+", part) is None for part in parts[1:5])
or re.fullmatch(r"(?:p?[0-9a-f]{16})?", parts[5]) is None
or re.fullmatch(r"[0-9a-f]{16}", parts[6]) is None
):
raise HTTPException(status_code=422, detail="raw_cursor_invalid")
raw_cursor = (
int(parts[1]),
int(parts[2]),
int(parts[3]),
int(parts[4]),
parts[5],
parts[6],
)
if any(value > 100_000_000 for value in raw_cursor[:4]):
raise HTTPException(status_code=422, detail="raw_cursor_invalid")
sources = split_csv(source)
excluded = split_csv(exclude_sources)
sessions = split_csv(session_id)
cursor_scope = hashlib.sha256(
json.dumps(
{
"user_id": user_id,
"statuses": sorted(chosen_statuses),
"sources": sorted(sources),
"excluded": sorted(excluded),
"sessions": sorted(sessions),
"order": order,
},
sort_keys=True,
separators=(",", ":"),
).encode()
).hexdigest()[:16]
if cursor and raw_cursor[5] != cursor_scope:
raise HTTPException(status_code=422, detail="raw_cursor_scope_mismatch")
try:
client = await _sdk_client_handle(app).client()
query = {
@@ -4527,8 +4575,154 @@ def create_app(
"include_sources": sources or None,
"exclude_sources": excluded or None,
"session_ids": sessions or None,
"order": order if unit == "raw" else "desc",
"order": order if unit == "raw" or paging == "cursor" else "desc",
}
if unit == "turn" and paging == "cursor":
(
raw_offset,
turn_offset,
emitted,
previous_total,
expected_raw,
_scope,
) = raw_cursor
value = await asyncio.to_thread(
client.list_memories,
**query,
offset=raw_offset,
limit=min(limit + 1, 500),
)
raw_page = HySdkRawPage.from_memory_list(
value, limit=min(limit + 1, 500), offset=raw_offset
)
def raw_anchor(record: HySdkRawRecord) -> str:
return hashlib.sha256(
f"{record.raw_id}\0{record.content}".encode()
).hexdigest()[:16]
if cursor:
if raw_page.total < previous_total:
raise HTTPException(status_code=409, detail="raw_cursor_stale")
observed_total = raw_page.total
added = raw_page.total - previous_total
if added and order == "desc":
raw_offset += added
value = await asyncio.to_thread(
client.list_memories,
**query,
offset=raw_offset,
limit=min(limit + 1, 500),
)
raw_page = HySdkRawPage.from_memory_list(
value,
limit=min(limit + 1, 500),
offset=raw_offset,
)
if raw_page.total != observed_total:
raise HTTPException(
status_code=409, detail="raw_cursor_stale"
)
if expected_raw.startswith("p"):
if raw_offset == 0:
raise HTTPException(
status_code=409, detail="raw_cursor_stale"
)
previous_value = await asyncio.to_thread(
client.list_memories,
**query,
offset=raw_offset - 1,
limit=1,
)
previous_page = HySdkRawPage.from_memory_list(
previous_value, limit=1, offset=raw_offset - 1
)
valid_anchor = (
previous_page.total == raw_page.total
and bool(previous_page.raw)
and raw_anchor(previous_page.raw[0])
== expected_raw[1:]
)
else:
valid_anchor = (
bool(raw_page.raw)
and raw_anchor(raw_page.raw[0]) == expected_raw
)
if expected_raw and not valid_anchor:
raise HTTPException(status_code=409, detail="raw_cursor_stale")
selected_turns: list[tuple[HySdkRawRecord, RawDisplayTurn]] = []
next_cursor: str | None = None
for record_index, record in enumerate(raw_page.raw):
turns = raw_display_turns(record)
ordered_turns = (
tuple(reversed(turns)) if order == "desc" else turns
)
start = turn_offset if record_index == 0 else 0
for index in range(start, len(ordered_turns)):
selected_turns.append((record, ordered_turns[index]))
if len(selected_turns) < limit:
continue
absolute_raw = raw_offset + record_index
next_raw = absolute_raw
next_turn = index + 1
if next_turn >= len(ordered_turns):
next_raw += 1
next_turn = 0
if next_raw < raw_page.total:
if next_turn:
anchor_record = record
else:
next_index = record_index + 1
anchor_record = (
raw_page.raw[next_index]
if next_index < len(raw_page.raw)
else None
)
anchor = (
raw_anchor(anchor_record)
if anchor_record is not None
else f"p{raw_anchor(record)}"
)
next_cursor = (
f"turn:{next_raw}:{next_turn}:"
f"{emitted + len(selected_turns)}:{raw_page.total}:"
f"{anchor}:{cursor_scope}"
)
break
if len(selected_turns) >= limit:
break
if (
next_cursor is None
and raw_offset + len(raw_page.raw) < raw_page.total
):
if not raw_page.raw:
raise HTTPException(
status_code=409, detail="raw_cursor_stale"
)
next_raw = raw_offset + len(raw_page.raw)
previous_anchor = raw_anchor(raw_page.raw[-1])
next_cursor = (
f"turn:{next_raw}:0:{emitted + len(selected_turns)}:"
f"{raw_page.total}:p{previous_anchor}:{cursor_scope}"
)
return SdkRawPage(
user_id=user_id,
items=tuple(
SdkRawItem.of(record, turn)
for record, turn in selected_turns
),
# Cursor consumers do not use a precise total. This lower
# bound preserves the legacy numeric field without forcing
# a full scan merely to count variable-size turn batches.
total=emitted + len(selected_turns) + (1 if next_cursor else 0),
offset=emitted,
limit=limit,
unit="turn",
next_cursor=next_cursor,
has_more=next_cursor is not None,
total_exact=False,
)
if unit == "raw":
value = await asyncio.to_thread(
client.list_memories,
@@ -1,6 +1,6 @@
import hashlib
import os
from collections.abc import AsyncIterator, Iterator
from collections.abc import AsyncIterator, Callable, Iterator
from dataclasses import dataclass
from pathlib import Path
@@ -28,6 +28,7 @@ def discovered_off_loop(
suffixes: frozenset[str],
max_targets: int,
excluded_directory_names: frozenset[str] = frozenset(),
directory_predicate: Callable[[Path], bool] | None = None,
) -> AsyncIterator[SourceFileDiscoveryEvent]:
"""The same walk, for callers that have an event loop to keep free.
@@ -43,6 +44,7 @@ def discovered_off_loop(
suffixes=suffixes,
max_targets=max_targets,
excluded_directory_names=excluded_directory_names,
directory_predicate=directory_predicate,
),
scope_key=grant.source_id,
)
@@ -54,6 +56,7 @@ def discover_source_files(
suffixes: frozenset[str],
max_targets: int,
excluded_directory_names: frozenset[str] = frozenset(),
directory_predicate: Callable[[Path], bool] | None = None,
) -> Iterator[SourceFileDiscoveryEvent]:
"""Metadata only: one lstat per candidate, no open and no per-file grant lookup.
@@ -71,10 +74,16 @@ def discover_source_files(
grant.root,
followlinks=False,
):
current = Path(directory)
child_directories[:] = sorted(
child
for child in child_directories
if not (Path(directory) / child).is_symlink() and child not in excluded_directory_names
if not (current / child).is_symlink()
and child not in excluded_directory_names
and (
directory_predicate is None
or directory_predicate((current / child).relative_to(grant.root))
)
)
for filename in sorted(filenames):
if Path(filename).suffix.lower() not in suffixes:
@@ -591,7 +591,11 @@ class PollingLiveSyncSupervisor:
async def _poll_source(self, source_id: str) -> None:
while True:
await asyncio.sleep(self.interval_seconds)
# A watched source is normally triggered by its filesystem event.
# Its timer is only a missed-event backstop; recursively walking
# every multi-gigabyte host root once a minute starved unrelated
# Core reads even when no source had changed.
await asyncio.sleep(self._poll_interval(source_id))
try:
await self.trigger(source_id)
except asyncio.CancelledError:
@@ -608,6 +612,11 @@ class PollingLiveSyncSupervisor:
except Exception as again:
self._note_internal_error(source_id, again)
def _poll_interval(self, source_id: str) -> float:
if source_id in (self._watched_sources or ()):
return max(self.interval_seconds, 15 * 60)
return self.interval_seconds
def _note_worker_still_running(self, source_id: str) -> None:
current = self.health(source_id)
self._set_health(
@@ -226,6 +226,9 @@ class JsonlSourceAdapter:
self.header_gate = header_gate
self._header_cache: dict[str, tuple[str, dict[str, JsonValue] | None]] = {}
self._history_listing_cache: SourceListing | None = None
self._probe_listing_cache: tuple[
dict[str, SourceCursor], int | None, SourceListing
] | None = None
@property
def descriptor(self) -> SourceDescriptor:
@@ -281,6 +284,11 @@ class JsonlSourceAdapter:
predicate=self.file_predicate,
cursors=options.cursors,
)
self._probe_listing_cache = (
dict(options.cursors),
options.max_scan_targets,
listing,
)
return bool(listing.changed or listing.failures)
async def live_baseline(self) -> dict[str, SourceCursor]:
@@ -315,9 +323,11 @@ class JsonlSourceAdapter:
async for event in self._scan_frozen(options):
yield event
return
listing = await asyncio.to_thread(
self._history_listing if options.reuse_listing else self._fresh_listing,
options,
listing = (
await asyncio.to_thread(self._history_listing, options)
if options.reuse_listing
else self._take_probe_listing(options)
or await asyncio.to_thread(self._fresh_listing, options)
)
for failure in listing.failures:
yield ScanFailure(reason_code=discovery_failure_code(failure))
@@ -349,6 +359,16 @@ class JsonlSourceAdapter:
self._history_listing_cache = listing
return listing
def _take_probe_listing(self, options: ScanOptions) -> "SourceListing | None":
cached = self._probe_listing_cache
self._probe_listing_cache = None
if cached is None:
return None
cursors, max_scan_targets, listing = cached
if cursors != options.cursors or max_scan_targets != options.max_scan_targets:
return None
return listing
def _fresh_listing(self, options: ScanOptions) -> "SourceListing":
return list_changed_files(
self.grant,
@@ -9,6 +9,7 @@ from collections.abc import AsyncIterator, Iterator
from dataclasses import dataclass, replace
from datetime import UTC, datetime, timedelta, timezone
from functools import partial
from pathlib import Path
from pydantic import JsonValue
@@ -175,6 +176,9 @@ class CursorUnifiedSourceAdapter:
self.grant = grant
self.source_version = source_version
self._history_listing_cache: _CursorListing | None = None
self._probe_listing_cache: tuple[
dict[str, SourceCursor], int | None, _CursorListing
] | None = None
@property
def descriptor(self) -> SourceDescriptor:
@@ -207,6 +211,20 @@ class CursorUnifiedSourceAdapter:
state = await self.install_state()
return min(state.session_count, limit) if limit is not None else state.session_count
async def has_changes(self, options: ScanOptions) -> bool:
listing = await asyncio.to_thread(self._fresh_listing, options)
self._probe_listing_cache = (
dict(options.cursors),
options.max_scan_targets,
listing,
)
return bool(
listing.transcripts
or listing.stores
or listing.legacy
or listing.failures
)
async def live_baseline(self) -> dict[str, SourceCursor]:
"""Capture Cursor's current tail so the realtime lane never backfills history."""
listing = await asyncio.to_thread(
@@ -222,9 +240,11 @@ class CursorUnifiedSourceAdapter:
async for event in self._scan_frozen(options):
yield event
return
listing = await asyncio.to_thread(
self._history_listing if options.reuse_listing else self._fresh_listing,
options,
listing = (
await asyncio.to_thread(self._history_listing, options)
if options.reuse_listing
else self._take_probe_listing(options)
or await asyncio.to_thread(self._fresh_listing, options)
)
for failure in listing.failures:
yield ScanFailure(reason_code=discovery_failure_code(failure))
@@ -329,6 +349,16 @@ class CursorUnifiedSourceAdapter:
self._history_listing_cache = listing
return listing
def _take_probe_listing(self, options: ScanOptions) -> "_CursorListing | None":
cached = self._probe_listing_cache
self._probe_listing_cache = None
if cached is None:
return None
cursors, max_scan_targets, listing = cached
if cursors != options.cursors or max_scan_targets != options.max_scan_targets:
return None
return listing
def _fresh_listing(self, options: ScanOptions) -> "_CursorListing":
return _list_cursor_files(
self.grant,
@@ -376,6 +406,7 @@ class CursorUnifiedSourceAdapter:
self.grant,
suffixes=frozenset({".db"}),
max_targets=options.max_scan_targets or 100_000,
directory_predicate=_cursor_data_directory,
):
if isinstance(discovered, SourceFileDiscoveryFailure):
yield ScanFailure(reason_code=discovery_failure_code(discovered))
@@ -477,7 +508,10 @@ def _list_cursor_files(
legacy: list[AuthorizedSourceFile] = []
failures: list[SourceFileDiscoveryFailure] = []
for item in discover_source_files(
grant, suffixes=frozenset({".jsonl", ".db", ".vscdb"}), max_targets=max_targets
grant,
suffixes=frozenset({".jsonl", ".db", ".vscdb"}),
max_targets=max_targets,
directory_predicate=_cursor_data_directory,
):
if isinstance(item, SourceFileDiscoveryFailure):
failures.append(item)
@@ -586,6 +620,25 @@ def _is_cursor_data_file(source_file: AuthorizedSourceFile) -> bool:
)
def _cursor_data_directory(relative_path: Path) -> bool:
"""Enter only Cursor directories that can contain conversation authority.
The granted root also contains extensions, skills, caches, logs and agent
artifacts. Recursively walking all of them every live-sync poll made the
reader monopolize Core even though none can pass the file predicates.
"""
parts = relative_path.parts
if not parts:
return False
if parts[0] in {"chats", "acp-sessions"}:
return True
if parts[0] == "projects":
return len(parts) <= 2 or parts[2] == "agent-transcripts"
if parts[0] == "User":
return len(parts) == 1 or parts[1] in {"globalStorage", "workspaceStorage"}
return False
def _is_transcript_file(source_file: AuthorizedSourceFile) -> bool:
parts = source_file.relative_path.parts
return source_file.relative_path.suffix == ".jsonl" and "agent-transcripts" in parts
@@ -634,6 +687,7 @@ def _select_transcript_files(
grant,
suffixes=frozenset({".jsonl"}),
max_targets=100_000,
directory_predicate=_cursor_data_directory,
):
if not isinstance(item, AuthorizedSourceFile) or not _is_transcript_file(item):
continue
@@ -654,6 +708,7 @@ def _subagent_conversation_ids(grant: AuthorizedSourceRoot) -> set[str]:
grant,
suffixes=frozenset({".jsonl"}),
max_targets=100_000,
directory_predicate=_cursor_data_directory,
)
if isinstance(item, AuthorizedSourceFile)
and _is_transcript_file(item)
@@ -293,6 +293,11 @@ def test_health_states_that_watching_is_layered_on_the_timer(tmp_path: Path) ->
assert live.health(PI_DESCRIPTOR.source_id).mode == "watch_and_polling"
assert live.health("unknown-source").mode == "watch_and_polling"
live.set_watched_sources((PI_DESCRIPTOR.source_id,))
assert live._poll_interval(PI_DESCRIPTOR.source_id) == max(
live.interval_seconds, 15 * 60
)
assert live._poll_interval("unknown-source") == live.interval_seconds
await imports.shutdown()
asyncio.run(scenario())
@@ -77,6 +77,36 @@ def test_workbuddy_record_parsing_never_blocks_the_event_loop(tmp_path) -> None:
asyncio.run(scenario())
def test_jsonl_live_probe_listing_is_reused_by_the_import(tmp_path, monkeypatch) -> None:
root = tmp_path / ".workbuddy" / "projects"
write_jsonl(
root / "project" / "session.jsonl",
[{"role": "user", "sessionId": "session", "content": "new conversation"}],
)
adapter = create_workbuddy_adapter(grant("workbuddy", root))
from one_memory.imports.sources import common
calls = 0
original = common.list_changed_files
def counted(*args, **kwargs):
nonlocal calls
calls += 1
return original(*args, **kwargs)
monkeypatch.setattr(common, "list_changed_files", counted)
async def probe_then_scan():
options = ScanOptions()
assert await adapter.has_changes(options)
return [event async for event in adapter.scan(options)]
records = evidence(asyncio.run(probe_then_scan()))
assert [item.evidence.verbatim for item in records] == ["new conversation"]
assert calls == 1
def test_cursor_store_user_info_envelope_is_not_attributed_to_the_person(tmp_path) -> None:
"""A Cursor store blob can be the host's session context, not the user's words.
@@ -1352,6 +1382,73 @@ def test_cursor_transcripts_with_the_same_name_in_two_projects_are_both_kept(
assert len({item.evidence.source_ordinal for item in records}) == 2
def test_cursor_discovery_does_not_enter_non_conversation_trees(tmp_path) -> None:
root = tmp_path / ".cursor"
write_jsonl(
root / "projects" / "project" / "agent-transcripts" / "kept" / "kept.jsonl",
[
{
"role": "user",
"message": {"content": [{"type": "text", "text": "kept conversation"}]},
}
],
)
write_jsonl(
root
/ "extensions"
/ "large-cache"
/ "agent-transcripts"
/ "ignored"
/ "ignored.jsonl",
[
{
"role": "user",
"message": {"content": [{"type": "text", "text": "must not be scanned"}]},
}
],
)
records = evidence(asyncio.run(scan(CursorUnifiedSourceAdapter(grant=grant("cursor", root)))))
assert [item.evidence.verbatim for item in records] == ["kept conversation"]
def test_cursor_live_probe_listing_is_reused_by_the_import(tmp_path, monkeypatch) -> None:
root = tmp_path / ".cursor"
write_jsonl(
root / "projects" / "project" / "agent-transcripts" / "new" / "new.jsonl",
[
{
"role": "user",
"message": {"content": [{"type": "text", "text": "new conversation"}]},
}
],
)
adapter = CursorUnifiedSourceAdapter(grant=grant("cursor", root))
from one_memory.imports.adapters import ScanOptions
from one_memory.imports.sources import cursor_unified
calls = 0
original = cursor_unified._list_cursor_files
def counted(*args, **kwargs):
nonlocal calls
calls += 1
return original(*args, **kwargs)
monkeypatch.setattr(cursor_unified, "_list_cursor_files", counted)
async def probe_then_scan():
options = ScanOptions()
assert await adapter.has_changes(options)
return [event async for event in adapter.scan(options)]
records = evidence(asyncio.run(probe_then_scan()))
assert [item.evidence.verbatim for item in records] == ["new conversation"]
assert calls == 1
def test_cursor_bounded_scans_reuse_the_authorized_file_listing(tmp_path, monkeypatch) -> None:
root = tmp_path / ".cursor"
transcript = root / "projects" / "project" / "agent-transcripts" / "session" / "session.jsonl"
+78 -1
View File
@@ -214,7 +214,15 @@ class RecordingSdk:
)
and (not kwargs.get("session_ids") or row["session_id"] in kwargs["session_ids"])
]
return {"vdb": {"memories": rows, "total": len(rows)}}
total = len(rows)
offset = kwargs.get("offset", 0)
limit = kwargs.get("limit", 100)
return {
"vdb": {
"memories": rows[offset : offset + limit],
"total": total,
}
}
rows = [_chat(user_id, *row[:2], *row[3:]) for row in CHAT_ROWS.get(user_id, [])]
if layers:
rows = [row for row in rows if row["layer"] in layers]
@@ -712,6 +720,75 @@ def test_raw_lists_active_and_shadow_turns_with_roles_and_hashes(tmp_path, monke
["我喜欢绿茶", "也记住了。"],
["我叫费天翔", "好的,记住了。"],
]
cursor = None
first_cursor = None
cursor_turns = []
cursor_calls = len(sdk.all("list_memories"))
while True:
params = {"unit": "turn", "paging": "cursor", "limit": 1}
if cursor is not None:
params["cursor"] = cursor
response = await client.get("/api/v2/raw", params=params)
assert response.status_code == 200, response.text
cursor_page = response.json()
assert cursor_page["total_exact"] is False
assert len(cursor_page["items"]) == 1
cursor_turns.append(cursor_page["items"][0]["messages"][0]["content"])
cursor = cursor_page["next_cursor"]
if first_cursor is None:
first_cursor = cursor
if cursor is None:
assert cursor_page["has_more"] is False
break
assert cursor_page["has_more"] is True
assert cursor_turns == ["我喜欢绿茶", "我叫费天翔", "你好"]
assert len(sdk.all("list_memories")) == cursor_calls + 3
inserted = {
"memory_id": "raw-new",
"content": "[user]: 新消息",
"status": "active",
"session_id": "s-new",
"source": "cursor",
"memory_at": 1_790_200_000,
"gmt_created": 1_790_200_100,
}
RAW_ROWS["local"].insert(0, inserted)
try:
continued = await client.get(
"/api/v2/raw",
params={
"unit": "turn",
"paging": "cursor",
"limit": 1,
"cursor": first_cursor,
},
)
assert continued.status_code == 200, continued.text
assert continued.json()["items"][0]["messages"][0]["content"] == "我叫费天翔"
finally:
RAW_ROWS["local"].remove(inserted)
assert (
await client.get(
"/api/v2/raw",
params={
"unit": "turn",
"paging": "cursor",
"cursor": first_cursor,
"source": "hermes",
},
)
).status_code == 422
assert (
await client.get(
"/api/v2/raw",
params={
"unit": "turn",
"paging": "cursor",
"cursor": "turn:²:0:0:2::0000000000000000",
},
)
).status_code == 422
calls_before_refresh = len(sdk.all("list_memories"))
await client.get("/api/v2/raw", params={"unit": "turn"})
assert len(sdk.all("list_memories")) == calls_before_refresh + 1
+78 -56
View File
@@ -110,12 +110,15 @@ type Props = {
settledSource?: string;
settledQuery?: string;
requestedOffset?: number;
cursorPaging?: boolean;
hasMore?: boolean;
/** The source id the SDK uses for "no source given" (shown as 未识别来源). */
unattributedSource?: string;
onQuery(value: string): void;
onSource(value: string): void;
resetPage(): void;
onPage(offset: number): void;
onMore?(): void;
onRetry(): void;
};
};
@@ -1131,62 +1134,81 @@ export function MemoryMap(props: Props) {
aria-busy={pageTransitioning || undefined}
className="map-pagination"
>
<button
type="button"
className="quiet map-tool"
aria-label={text("上一页记忆")}
title={text("上一页记忆")}
disabled={pageTransitioning || library.page.offset === 0}
onClick={() => {
if (axis === "topic") setGroupKey("");
if (library.page) {
const offset = library.page.offset - library.page.limit;
setPageRetrying(false);
pageRetrySawLoading.current = false;
setPendingPageOffset(offset);
library.onPage(offset);
}
}}
>
<ChevronLeft size={16} aria-hidden="true" />
</button>
<span>
{text(
"共 {count} 条,第 {page} / {pages} 页",
{
count: count(library.page.total),
page:
library.page.total === 0
? 0
: Math.floor(library.page.offset / library.page.limit) +
1,
pages: Math.ceil(library.page.total / library.page.limit),
},
library.page.total,
)}
</span>
<button
type="button"
className="quiet map-tool"
aria-label={text("下一页记忆")}
title={text("下一页记忆")}
disabled={
pageTransitioning ||
library.page.offset + library.page.limit >= library.page.total
}
onClick={() => {
if (axis === "topic") setGroupKey("");
if (library.page) {
const offset = library.page.offset + library.page.limit;
setPageRetrying(false);
pageRetrySawLoading.current = false;
setPendingPageOffset(offset);
library.onPage(offset);
}
}}
>
<ChevronRight size={16} aria-hidden="true" />
</button>
{library.cursorPaging ? (
library.hasMore ? (
<R6Button
disabled={pageTransitioning}
onClick={() => library.onMore?.()}
>
{pageTransitioning ? text("正在读取…") : text("加载更多")}
</R6Button>
) : (
<span>{text("已加载全部")}</span>
)
) : (
<>
<button
type="button"
className="quiet map-tool"
aria-label={text("上一页记忆")}
title={text("上一页记忆")}
disabled={pageTransitioning || library.page.offset === 0}
onClick={() => {
if (axis === "topic") setGroupKey("");
if (library.page) {
const offset = library.page.offset - library.page.limit;
setPageRetrying(false);
pageRetrySawLoading.current = false;
setPendingPageOffset(offset);
library.onPage(offset);
}
}}
>
<ChevronLeft size={16} aria-hidden="true" />
</button>
<span>
{text(
"共 {count} 条,第 {page} / {pages} 页",
{
count: count(library.page.total),
page:
library.page.total === 0
? 0
: Math.floor(
library.page.offset / library.page.limit,
) + 1,
pages: Math.ceil(
library.page.total / library.page.limit,
),
},
library.page.total,
)}
</span>
<button
type="button"
className="quiet map-tool"
aria-label={text("下一页记忆")}
title={text("下一页记忆")}
disabled={
pageTransitioning ||
library.page.offset + library.page.limit >=
library.page.total
}
onClick={() => {
if (axis === "topic") setGroupKey("");
if (library.page) {
const offset = library.page.offset + library.page.limit;
setPageRetrying(false);
pageRetrySawLoading.current = false;
setPendingPageOffset(offset);
library.onPage(offset);
}
}}
>
<ChevronRight size={16} aria-hidden="true" />
</button>
</>
)}
</nav>
)}
{!library && loaded && !filtered.length && (search || groupKey) && (
+28 -12
View File
@@ -320,9 +320,9 @@ it("pages and filters the raw conversation data plane in lite", async () => {
setup();
reads.processing.mockResolvedValue(modeProcessing("lite"));
reads.raw.mockImplementation(async (options = {}) => {
const offset = "offset" in options ? (options.offset ?? 0) : 0;
const cursor = "cursor" in options ? options.cursor : null;
const source = "source" in options ? options.source : undefined;
const rawId = `${source ?? "all"}-${offset}`;
const rawId = `${source ?? "all"}-${cursor ? "more" : "first"}`;
return {
items: [
{
@@ -333,14 +333,17 @@ it("pages and filters the raw conversation data plane in lite", async () => {
source: source ?? "cursor",
session_id: `session-${rawId}`,
memory_at: null,
gmt_created: 1_761_000_000 - offset,
gmt_created: 1_761_000_000,
custom: {},
messages: [{ role: "user", content: `原始会话 ${rawId}` }],
},
],
total: source ? 1 : 21,
total: cursor ? 2 : 1,
limit: 20,
offset,
offset: cursor ? 1 : 0,
next_cursor: !source && !cursor ? "turn:1:0:1" : null,
has_more: !source && !cursor,
total_exact: false,
};
});
const user = userEvent.setup();
@@ -359,30 +362,43 @@ it("pages and filters the raw conversation data plane in lite", async () => {
/>,
);
expect(await screen.findAllByText("原始会话 all-0")).toHaveLength(2);
await user.click(screen.getByRole("button", { name: "下一页记忆" }));
expect(await screen.findAllByText("原始会话 all-20")).toHaveLength(2);
expect(await screen.findAllByText("原始会话 all-first")).toHaveLength(2);
await user.click(screen.getByRole("button", { name: "加载更多" }));
expect(await screen.findAllByText("原始会话 all-more")).toHaveLength(1);
expect(screen.getAllByText("原始会话 all-first")).toHaveLength(2);
expect(reads.raw).toHaveBeenLastCalledWith(
expect.objectContaining({ source: undefined, offset: 20, limit: 20 }),
expect.objectContaining({
source: undefined,
cursor: "turn:1:0:1",
paging: "cursor",
limit: 20,
}),
);
await user.click(screen.getByRole("combobox", { name: "来源" }));
await user.click(screen.getByRole("option", { name: "Cursor" }));
expect(await screen.findAllByText("原始会话 cursor-0")).toHaveLength(2);
expect(await screen.findAllByText("原始会话 cursor-first")).toHaveLength(2);
expect(reads.raw).toHaveBeenLastCalledWith(
expect.objectContaining({ source: "cursor", offset: 0, limit: 20 }),
expect.objectContaining({
source: "cursor",
cursor: null,
paging: "cursor",
limit: 20,
}),
);
await user.click(screen.getByRole("combobox", { name: "来源" }));
await user.click(screen.getByRole("option", { name: "手动添加" }));
expect(
await screen.findAllByText("原始会话 one-memory-manual-0"),
await screen.findAllByText("原始会话 one-memory-manual-first"),
).toHaveLength(2);
expect(reads.raw).toHaveBeenLastCalledWith(
expect.objectContaining({
source: "one-memory-manual",
offset: 0,
limit: 20,
paging: "cursor",
cursor: null,
}),
);
});
+96 -21
View File
@@ -123,6 +123,7 @@ export function MemorySpace({
const selectedRawRef = useRef<RawMapRecord | null>(null);
selectedRawRef.current = selectedRaw;
const rawRecords = useRef(new Map<string, RawMapRecord>());
const rawRecordsScope = useRef("");
const [detailLoad, setDetailLoad] = useState<{
memoryId: string;
status: "loading" | "failed";
@@ -209,6 +210,7 @@ export function MemorySpace({
const [sdkLibraryQuery, setSdkLibraryQuery] = useState("");
const [sdkLibrarySource, setSdkLibrarySource] = useState("");
const [sdkLibraryOffset, setSdkLibraryOffset] = useState(0);
const [sdkRawCursor, setSdkRawCursor] = useState<string | null>(null);
const [processing, setProcessing] = useState<MemoryProcessingSummary | null>(
null,
);
@@ -226,6 +228,10 @@ export function MemorySpace({
evolve_available: pipelineMode === "ultra",
}
: null;
const refreshSdkMap = useCallback(() => {
if (lite) setSdkRawCursor(null);
bumpSdkMap();
}, [bumpSdkMap, lite]);
const requestedQuery = sdkLibraryQuery.trim();
const sdkPageSettled =
sdkLibrary.settledMode === pipelineMode &&
@@ -276,7 +282,10 @@ export function MemorySpace({
if (first) setSdkLibrary((current) => ({ ...current, loading: true }));
try {
const rememberRaw = (items: RawMapRecord[]) => {
const next = new Map<string, RawMapRecord>();
const next =
sdkRawCursor && rawRecordsScope.current === windowKey
? new Map(rawRecords.current)
: new Map<string, RawMapRecord>();
if (selectedRawRef.current)
next.set(
rawRecordId(selectedRawRef.current),
@@ -284,6 +293,7 @@ export function MemorySpace({
);
for (const item of items) next.set(rawRecordId(item), item);
rawRecords.current = next;
rawRecordsScope.current = windowKey;
};
const [page, graph] = lite
? await Promise.all([
@@ -297,8 +307,10 @@ export function MemorySpace({
})
: getSdkRaw({
source,
offset: sdkLibraryOffset,
offset: 0,
limit: SDK_MEMORY_PAGE_SIZE,
paging: "cursor",
cursor: sdkRawCursor,
signal: request.signal,
}).then((rawPage) => {
rememberRaw(rawPage.items);
@@ -346,17 +358,64 @@ export function MemorySpace({
};
// The window expired: the server searched again and answered page 1.
if (page.window_reset && sdkLibraryOffset !== 0) setSdkLibraryOffset(0);
setSdkLibrary({
page,
loading: false,
failed: false,
settledSource: sdkLibrarySource,
settledQuery: sdkLibraryQuery,
settledMode: pipelineMode,
setSdkLibrary((current) => {
const currentPage = current.page;
const append =
lite &&
sdkRawCursor !== null &&
currentPage !== null &&
current.settledSource === sdkLibrarySource &&
current.settledQuery === sdkLibraryQuery &&
current.settledMode === pipelineMode;
if (!append)
return {
page,
loading: false,
failed: false,
settledSource: sdkLibrarySource,
settledQuery: sdkLibraryQuery,
settledMode: pipelineMode,
};
const items = [...currentPage.items];
const known = new Set(items.map((item) => item.memory_id));
for (const item of page.items)
if (!known.has(item.memory_id)) {
known.add(item.memory_id);
items.push(item);
}
return {
page: {
...page,
items,
total: items.length + (page.has_more ? 1 : 0),
offset: 0,
source_ids_by_memory: {
...currentPage.source_ids_by_memory,
...page.source_ids_by_memory,
},
source_time_by_memory: {
...currentPage.source_time_by_memory,
...page.source_time_by_memory,
},
},
loading: false,
failed: false,
settledSource: sdkLibrarySource,
settledQuery: sdkLibraryQuery,
settledMode: pipelineMode,
};
});
setSdkGraph({ scope: windowKey, ...graph });
} catch {
} catch (reason) {
if (cancelled || request.signal.aborted) return;
if (
lite &&
sdkRawCursor !== null &&
apiErrorCode(reason) === "raw_cursor_stale"
) {
setSdkRawCursor(null);
return;
}
setSdkLibrary((current) => ({
...current,
loading: false,
@@ -365,10 +424,13 @@ export function MemorySpace({
}
};
void read(true);
const timer = setInterval(() => void read(false), 30_000);
const timer =
lite && sdkRawCursor !== null
? undefined
: setInterval(() => void read(false), 30_000);
return () => {
cancelled = true;
clearInterval(timer);
if (timer !== undefined) clearInterval(timer);
controller?.abort();
};
}, [
@@ -377,6 +439,7 @@ export function MemorySpace({
sdkLibraryOffset,
sdkLibraryQuery,
sdkLibrarySource,
sdkRawCursor,
lite,
pipelineMode,
]);
@@ -525,6 +588,7 @@ export function MemorySpace({
rawRecords.current.clear();
sdkSearchWindow.current = { key: "", id: null };
setSdkLibraryOffset(0);
setSdkRawCursor(null);
setSdkLibraryQuery("");
setSdkGraph(null);
setSdkLibrary((current) => ({
@@ -804,7 +868,7 @@ export function MemorySpace({
);
trace.done("succeeded");
closeDetail();
bumpSdkMap();
refreshSdkMap();
} catch (reason) {
trace.done(writeTraceStatus(reason), {
reason: "memory_action_failed",
@@ -833,7 +897,7 @@ export function MemorySpace({
trace.done("succeeded");
setDetail(sdkDetail(saved));
setDraft(sdkEditableText(saved));
bumpSdkMap();
refreshSdkMap();
} catch (reason) {
trace.done(writeTraceStatus(reason), {
reason: isDefiniteRefusal(reason)
@@ -898,10 +962,12 @@ export function MemorySpace({
? t("memory:记忆暂时无法读取")
: lite
? sdkPage
? `${t("memory:原始对话")} · ${tn(
"sdk.raw.found",
sdkPage.total,
)}`
? sdkPage.total_exact === false
? t("memory:原始对话")
: `${t("memory:原始对话")} · ${tn(
"sdk.raw.found",
sdkPage.total,
)}`
: sdkLibrary.failed
? null
: t("memory:正在读取…")
@@ -934,7 +1000,7 @@ export function MemorySpace({
onSaved={() => {
// A finished addition is new page content: re-read now rather than
// leave it for the next poll.
bumpSdkMap();
refreshSdkMap();
}}
onOpenMemory={(id) => {
if (!simplified) setMode("library");
@@ -1201,12 +1267,21 @@ export function MemorySpace({
settledSource: sdkLibrary.settledSource,
settledQuery: sdkLibrary.settledQuery,
requestedOffset: sdkLibraryOffset,
cursorPaging: lite && !sdkLibraryQuery.trim(),
hasMore: !!sdkPage?.has_more && !!sdkPage?.next_cursor,
unattributedSource: UNATTRIBUTED_SOURCE_ID,
onQuery: setSdkLibraryQuery,
onSource: setSdkLibrarySource,
resetPage: () => setSdkLibraryOffset(0),
resetPage: () => {
setSdkLibraryOffset(0);
setSdkRawCursor(null);
},
onPage: (offset) => setSdkLibraryOffset(offset),
onRetry: bumpSdkMap,
onMore: () => {
if (sdkPage?.next_cursor)
setSdkRawCursor(sdkPage.next_cursor);
},
onRetry: refreshSdkMap,
}
: undefined
}
+2
View File
@@ -33,6 +33,8 @@ it("lists raw conversations for the selected application", async () => {
offset: 20,
limit: 20,
unit: "turn",
paging: undefined,
cursor: undefined,
},
signal,
});
+13 -1
View File
@@ -987,6 +987,8 @@ export async function getSdkRaw({
offset = 0,
limit = 100,
unit = "turn",
paging,
cursor,
signal,
}: {
source?: string;
@@ -994,10 +996,20 @@ export async function getSdkRaw({
offset?: number;
limit?: number;
unit?: "raw" | "turn";
paging?: "offset" | "cursor";
cursor?: string | null;
signal?: AbortSignal;
} = {}): Promise<SdkRawPage> {
const response = await client.get<SdkRawPage>("/api/v2/raw", {
params: { source, session_id: sessionId, offset, limit, unit },
params: {
source,
session_id: sessionId,
offset,
limit,
unit,
paging,
cursor: cursor || undefined,
},
signal,
});
return response.data;
+11 -1
View File
@@ -72,6 +72,11 @@ function pageFor(
total: number,
offset: number,
limit: number,
cursor?: {
next_cursor?: string | null;
has_more?: boolean;
total_exact?: boolean;
},
): MemoryLibraryPage {
const sourceIds: Record<string, string[]> = {};
const sourceTime: Record<
@@ -97,6 +102,7 @@ function pageFor(
total,
offset,
limit,
...cursor,
source_ids_by_memory: sourceIds,
source_time_by_memory: sourceTime,
search_window: null,
@@ -107,7 +113,11 @@ function pageFor(
}
export function rawLibraryPage(page: SdkRawPage): MemoryLibraryPage {
return pageFor(page.items, page.total, page.offset, page.limit);
return pageFor(page.items, page.total, page.offset, page.limit, {
next_cursor: page.next_cursor,
has_more: page.has_more,
total_exact: page.total_exact,
});
}
export function rawSearchPage(items: SdkRawContextItem[]): MemoryLibraryPage {
+7
View File
@@ -510,6 +510,10 @@ export interface MemoryLibraryPage {
total: number;
offset: number;
limit: number;
/** Cursor paging: opaque continuation for a page whose exact total is not read. */
next_cursor?: string | null;
has_more?: boolean;
total_exact?: boolean;
/** Active evidence sources for each returned memory, from the same authorized page snapshot. */
source_ids_by_memory?: Record<string, string[]>;
/** When each returned memory's cited conversation happened (D96). `import`
@@ -709,6 +713,9 @@ export interface SdkRawPage {
limit: number;
offset: number;
unit?: "raw" | "turn";
next_cursor?: string | null;
has_more?: boolean;
total_exact?: boolean;
/** A time range was applied to this page in Core (the SDK list has none). */
time_filtered?: boolean;
}