tui_gateway: session.list reports scan-cap truncation honestly

This commit is contained in:
alt-glitch 2026-06-10 21:44:17 +05:30
parent 529d8084be
commit f86bc5170a
2 changed files with 53 additions and 2 deletions

View file

@ -6587,3 +6587,41 @@ def test_tool_complete_derives_error_from_result_convention(monkeypatch):
json.dumps({"results": [1, 2], "error": None}),
)
assert "error" not in events[-1][2]
def test_session_list_reports_scan_cap_truncation(monkeypatch):
# The bounded multi-source scan must say so when the 10k safety cap stops
# it with the requested window unfilled — an empty page past the cap is
# "truncated", not "no more sessions".
class _DenyAllDB:
def __init__(self):
self.offsets = []
def list_sessions_rich(self, source=None, limit=20, offset=0, **kw):
self.offsets.append(offset)
return [
{"id": f"s{offset}-{i}", "source": "tool", "title": "", "preview": "",
"started_at": 1, "message_count": 1}
for i in range(limit)
]
db = _DenyAllDB()
monkeypatch.setattr(server, "_get_db", lambda: db)
resp = server.handle_request(
{"id": "1", "method": "session.list", "params": {"sources": ["tui", "cli"], "limit": 5}}
)
result = resp["result"]
assert result["sessions"] == []
assert result["truncated"] is True
assert max(db.offsets) <= 10_000
def test_session_list_truncated_false_on_normal_paths(monkeypatch):
db = _picker_db()
monkeypatch.setattr(server, "_get_db", lambda: db)
legacy = server.handle_request({"id": "1", "method": "session.list", "params": {}})
assert legacy["result"]["truncated"] is False
filtered = server.handle_request(
{"id": "2", "method": "session.list", "params": {"sources": ["tui"], "limit": 5}}
)
assert filtered["result"]["truncated"] is False

View file

@ -3354,6 +3354,7 @@ def _(rid, params: dict) -> dict:
for s in db.list_sessions_rich(source=None, limit=fetch_limit)
if (s.get("source") or "").strip().lower() not in deny
][:limit]
list_truncated = False
else:
# Filtered/paginated path. Single source pushes into SQL; the
# deny-list and multi-source filter run gateway-side, so keep
@ -3372,7 +3373,14 @@ def _(rid, params: dict) -> dict:
collected: list = []
db_offset = 0
page = max(wanted * 2, 200)
while len(collected) < wanted and db_offset < 10_000:
scan_capped = False
while len(collected) < wanted:
if db_offset >= 10_000:
# Safety cap hit with the window still unfilled — report it
# honestly (``truncated``) so the client can say "results
# truncated" instead of silently serving an empty page.
scan_capped = True
break
batch = db.list_sessions_rich(
source=source_arg,
limit=page,
@ -3387,6 +3395,7 @@ def _(rid, params: dict) -> dict:
break
db_offset += page
rows = collected[offset : offset + limit]
list_truncated = scan_capped and len(collected) < wanted
return _ok(
rid,
@ -3409,7 +3418,11 @@ def _(rid, params: dict) -> dict:
"model": s.get("model") or None,
}
for s in rows
]
],
# True when the bounded multi-source scan hit its safety cap
# before filling the requested window — the client should show
# "results truncated" instead of treating the page as final.
"truncated": list_truncated,
},
)
except Exception as e: