From 8d6a38826eae57da96653137762a2068cab86f3b Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Wed, 11 Nov 2026 13:49:39 +0800 Subject: [PATCH] Avoid redundant Turn reads for dense Chat return history Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> --- .../capabilities/manager_context/roundtrip.py | 18 ++++++++++ tests/test_manager_context_roundtrip.py | 36 +++++++++++++++++++ 2 files changed, 54 insertions(+) diff --git a/loopx/capabilities/manager_context/roundtrip.py b/loopx/capabilities/manager_context/roundtrip.py index 7b435f4ea8..9ffe1afa5f 100644 --- a/loopx/capabilities/manager_context/roundtrip.py +++ b/loopx/capabilities/manager_context/roundtrip.py @@ -9,9 +9,11 @@ import logging import json import re +from fnmatch import fnmatch from hashlib import sha256 import threading from itertools import islice +from os import scandir from contextlib import contextmanager, ExitStack, nullcontext from urllib.parse import quote from datetime import datetime, timezone @@ -332,6 +334,22 @@ def candidate_paths(): if message.get("message_id") in pending_message_ids and isinstance(message.get("turn_id"), str) and message.get("turn_id") ) + # A cursor page holds at most 200 rows. Wider reads with a dense + # set of returns can skip opening each original Turn: the complete + # route scan costs at most a Turn and route lookup per visible Turn. + # Only choose locators here; the same route and result checks apply. + if len(turn_ids) > 200: + lookup_limit = 2 * len(turn_ids) + try: + with scandir(routes) as entries: + names = list(islice((entry.name for entry in entries if fnmatch(entry.name, "*.json")), + lookup_limit + 1)) + except OSError: + pass # A direct locator may still be readable. + else: + if 0 < len(names) <= lookup_limit: + yield from sorted(routes / name for name in names) + return for turn_id in turn_ids: try: turn = source_store.load_turn(session_id, turn_id) or {} diff --git a/tests/test_manager_context_roundtrip.py b/tests/test_manager_context_roundtrip.py index 7dbc80d149..3f68788037 100644 --- a/tests/test_manager_context_roundtrip.py +++ b/tests/test_manager_context_roundtrip.py @@ -1472,6 +1472,42 @@ def read(path): assert store.messages(sid) == messages +def test_dense_return_history_reads_routes_once_without_opening_each_turn(frozen_private_reply, monkeypatch): + store, session, *_ = frozen_private_reply + root, sid = store.root.parent, session["session_id"] + messages = [] + for i in range(202): + rid, tid = f"{i:064x}", f"{i:032x}" + source = root if i != 201 else root / "other-chat-host" + _write(_root(root) / "roundtrips" / f"{rid}.json", { + "request_id": rid, "session_id": sid, "source_chat_runtime_root": str(source), + }) + replies = _root(root) / "replies" / rid + _write(replies / "conclusion.json", {"text": f"Original result {i}"}) + _write(replies / "conclusion.delivery.json", {"status": "delivered"}) + messages.append({ + "message_id": "handoff." + _hash([rid, "conclusion"]), + "turn_id": tid, "role": "agent", "origin": "manager_followup", "text": f"Original result {i}", + }) + load_turn, opened = store.load_turn, [] + + def track_turn(session_id, turn_id): + opened.append(turn_id) + return load_turn(session_id, turn_id) + + monkeypatch.setattr(store, "load_turn", track_turn) + projected = project_chat_return_deliveries(root, sid, messages, source_store=store) + assert [row["text"] for row in projected] == [row["text"] for row in messages] + assert all(row["return_delivery"]["status"] == "delivered" for row in projected[:-1]) + assert projected[-1] == messages[-1] + assert not opened + + first = _root(root) / "replies" / ("0" * 64) / "conclusion.delivery.json" + _write(first, {"status": "verification_required", "error": "provider_delivery_unverified"}) + reread = project_chat_return_deliveries(root, sid, messages, source_store=store) + assert reread[0]["return_delivery"]["status"] == "verification_required" + + def test_paged_history_filters_result_io_and_rereads_late_state(frozen_private_reply, private_snapshot, monkeypatch): from loopx.control_plane.collaboration import inbox