From fc35010125da117ed7729ce9c692ff6e4b24c911 Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Tue, 3 Nov 2026 05:35:17 +0800 Subject: [PATCH] perf(chat): resolve history returns through original request receipts Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> --- .../capabilities/manager_context/roundtrip.py | 41 ++++++++- tests/test_manager_context_roundtrip.py | 91 +++++++++++++++++++ 2 files changed, 130 insertions(+), 2 deletions(-) diff --git a/loopx/capabilities/manager_context/roundtrip.py b/loopx/capabilities/manager_context/roundtrip.py index a5bb1a07a5..4d82705088 100644 --- a/loopx/capabilities/manager_context/roundtrip.py +++ b/loopx/capabilities/manager_context/roundtrip.py @@ -258,8 +258,45 @@ def project_chat_return_deliveries(root, session_id, messages, *, source_store=N if not pending_message_ids: return list(messages) statuses = {} - paths = sorted((_root(root) / "roundtrips").glob("*.json")) - for path in paths: + routes = _root(root) / "roundtrips" + + def candidate_paths(): + # The original Turn normally names its requests. Prefer those locators + # over scanning every conversation's routes for every history page. + # They are only hints: the route, source host and result identity below + # still have to match. Lost/older receipts retain the exhaustive fallback. + known = set() + if source_store is not None: + turn_ids = dict.fromkeys( + message.get("turn_id") for message in messages + if message.get("message_id") in pending_message_ids + and isinstance(message.get("turn_id"), str) and message.get("turn_id") + ) + for turn_id in turn_ids: + try: + turn = source_store.load_turn(session_id, turn_id) or {} + if turn.get("session_id") != session_id or turn.get("turn_id") != turn_id: + continue + response = turn.get("response") or {} + if not isinstance(response, dict): + continue + receipts = response.get("context_handoff_receipts") + if receipts is None: + receipts = [response.get("context_handoff_receipt")] + for receipt in receipts if isinstance(receipts, list) else []: + request_id = receipt.get("request_id") if isinstance(receipt, dict) else None + if (isinstance(request_id, str) and BARE_SHA256_PATTERN.fullmatch(request_id) + and request_id not in known): + known.add(request_id) + yield routes / (request_id + ".json") + except (OSError, ValueError, KeyError, TypeError): + continue + # This part is lazy: a fully resolved page never enumerates the directory. + for path in sorted(routes.glob("*.json")): + if path.stem not in known: + yield path + + for path in candidate_paths(): try: route = _read(path) if (route.get("session_id") != session_id diff --git a/tests/test_manager_context_roundtrip.py b/tests/test_manager_context_roundtrip.py index 29f756d0b3..23409095db 100644 --- a/tests/test_manager_context_roundtrip.py +++ b/tests/test_manager_context_roundtrip.py @@ -4,6 +4,7 @@ import threading from concurrent.futures import ThreadPoolExecutor from datetime import datetime, timedelta, timezone +from pathlib import Path import pytest @@ -924,6 +925,96 @@ def test_chat_snapshot_keeps_current_session_delivery_after_unrelated_route_limi assert returned == expected +@pytest.fixture +def returned_history(flow): + root, registry, store, create = flow + session, turn, receipt = create() + acknowledge(root, "research", "worker", receipt["request_id"], "adopt", "Checked") + report(root, "research", "worker", receipt["request_id"], "conclusion", "Completed the original request.") + assert drain(root, registry, store, None) == 1 + return root, registry, store, session, turn, receipt + + +@pytest.mark.parametrize("plural", [False, True]) +def test_history_delivery_reads_original_route_without_global_scan_and_stays_fresh(returned_history, monkeypatch, plural): + root, _, store, session, turn, receipt = returned_history + sid, rid = session["session_id"], receipt["request_id"] + if plural: + store.update_turn(sid, turn["turn_id"], response={"context_handoff_receipts": [receipt]}) + messages = store.messages(sid) + followups = [message for message in messages if message.get("origin") == "manager_followup"] + routes = _root(root) / "roundtrips" + for index in range(10): + _write(routes / f"{index:064x}.json", {"request_id": f"{index:064x}", "session_id": "unrelated"}) + original_glob = Path.glob + + def no_global_routes(path, pattern, **kwargs): + assert path != routes, "A known original receipt must not enumerate all conversations" + return original_glob(path, pattern, **kwargs) + + monkeypatch.setattr(Path, "glob", no_global_routes) + before = {path: path.read_bytes() for path in root.rglob("*.json")} + projected = project_chat_return_deliveries(root, sid, followups, source_store=store) + assert len(projected) == len(followups) == 1 + assert projected[0]["return_delivery"]["status"] == "delivered" + assert {path: path.read_bytes() for path in root.rglob("*.json")} == before + # Delivery state remains current even when the original transcript is unchanged. + _write(_root(root) / "replies" / rid / "conclusion.delivery.json", { + "status": "verification_required", "error": "provider_delivery_unverified", + }) + latest = project_chat_return_deliveries(root, sid, followups, source_store=store) + assert latest[0]["return_delivery"]["status"] == "verification_required" + assert latest[0]["return_delivery"]["error"] == "provider_delivery_unverified" + assert store.messages(sid) == messages + + +@pytest.mark.parametrize("receipt_state", ["lost", "malformed", "unreadable_turn", "unmatched"]) +def test_history_delivery_keeps_fallback_for_missing_original_locators(returned_history, monkeypatch, receipt_state): + root, _, store, session, turn, receipt = returned_history + sid = session["session_id"] + if receipt_state == "unreadable_turn": + (store.root / "sessions" / sid / "turns" / (turn["turn_id"] + ".json")).write_text("{") + else: + response = {"context_handoff_receipts": "unsupported"} if receipt_state == "malformed" else ( + {"context_handoff_receipts": [None, {"request_id": "../outside"}, {"request_id": "0" * 64}]} + if receipt_state == "unmatched" else None + ) + store.update_turn(sid, turn["turn_id"], response=response) + messages = store.messages(sid) + routes, scanned = _root(root) / "roundtrips", [] + original_glob = Path.glob + + def observe(path, pattern, **kwargs): + if path == routes: + scanned.append(path) + return original_glob(path, pattern, **kwargs) + + monkeypatch.setattr(Path, "glob", observe) + projected = project_chat_return_deliveries(root, sid, messages, source_store=store) + assert len(scanned) == 1 + assert len(projected) == len(messages) + returned = next(message for message in projected if message.get("origin") == "manager_followup") + assert returned["return_delivery"] == {"schema_version": "manager_return_delivery_status_v0", **reply_status(root, receipt)[0]} + assert store.messages(sid) == messages + + +@pytest.mark.parametrize("drift", ["session", "host", "request"]) +def test_history_original_receipt_is_only_a_locator_not_route_authority(returned_history, drift): + root, _, store, session, _, receipt = returned_history + sid, rid = session["session_id"], receipt["request_id"] + path = _root(root) / "roundtrips" / (rid + ".json") + route = json.loads(path.read_text()) + if drift == "session": + route["session_id"] = "another-session" + elif drift == "host": + route["source_chat_runtime_root"] = str(root / "another-host") + else: + route["request_id"] = "0" * 64 + _write(path, route) + messages = store.messages(sid) + assert project_chat_return_deliveries(root, sid, messages, source_store=store) == messages + + def test_provider_verification_outage_retries_read_only_without_resend(flow): root, registry, store, create = flow _, _, receipt = create(True)