From 75d012e7fe3f19bab3bb040b7a885466dace9f9b Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Sun, 8 Nov 2026 16:51:32 +0800 Subject: [PATCH] perf(chat): read delivery state only for visible history results Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> --- .../history-recovery/README.md | 20 ++++- .../capabilities/manager_context/roundtrip.py | 9 +- loopx/control_plane/collaboration/inbox.py | 9 +- tests/test_manager_context_roundtrip.py | 83 +++++++++++++++++++ 4 files changed, 118 insertions(+), 3 deletions(-) diff --git a/docs/assets/personal-workspace/history-recovery/README.md b/docs/assets/personal-workspace/history-recovery/README.md index c0ba1eb701..682b826d66 100644 --- a/docs/assets/personal-workspace/history-recovery/README.md +++ b/docs/assets/personal-workspace/history-recovery/README.md @@ -16,7 +16,7 @@ the changed reader avoids the repeated prefix. The regression belongs to These captures use the built Chat bundle and the production HTTP/SQLite/Core binding owner in a disposable synthetic runtime. Its 464 original rows include -a 19,000-character body. The packaged UI recovered automatically through both +a 20,000-character body. The packaged UI recovered automatically through both interruptions, displayed the complete body once, retained the draft, and kept Send disabled until the complete history was verified. Independent store readback remained identical, with no message/Session creation or model Turns. @@ -36,3 +36,21 @@ PNG pixels and metadata were inspected. This qualifies the packaged browser reader, not the installed native App, live IM transport, server latency or business completion. Current source authorization and every page's frozen prefix/identity checks remain mandatory; checkpoints do not grant authority. + +The server also selects the current page's immutable result keys before reading +answer bodies and delivery sidecars. This matters when an unmatched legacy row +requires the existing exhaustive route fallback: off-page results no longer +need to be deserialized. The fallback still examines every candidate route; +Session/source checks, current status validation and unreadable-state handling +remain unchanged. No delivery state or authorization is cached between reads. + +A controlled production HTTP/SQLite/Core fixture contains 404 messages, 96 +request routes and 768 complete 20,000-character results. In alternating +base/head/head/base runs (seven uninstrumented samples each), the first page's +result-file reads fell from 1,536 to 80. Per-run median HTTP latency was +0.17–0.26 seconds on the base and 0.04–0.05 seconds after the change. Both arms +returned the identical complete history and left the original source unchanged. +These local synthetic measurements are neither a production p95 nor evidence +that other service contention is fixed. The durable HTTP and fallback +regressions live in `tests/test_manager_context_roundtrip.py`; they also retain +late verification changes, malformed state and the original full long body. diff --git a/loopx/capabilities/manager_context/roundtrip.py b/loopx/capabilities/manager_context/roundtrip.py index 8431f52edf..41db7b6d09 100644 --- a/loopx/capabilities/manager_context/roundtrip.py +++ b/loopx/capabilities/manager_context/roundtrip.py @@ -310,7 +310,14 @@ def candidate_paths(): request_id = str(route.get("request_id") or "") if path.stem != request_id or not BARE_SHA256_PATTERN.fullmatch(request_id): continue - for item in reply_status(root, route): + # A legacy/unmatched row can require the exhaustive route fallback. + # Match immutable result keys before reading potentially long reply + # bodies and delivery sidecars for off-page results. The existing + # result owner still validates every selected status on this read. + for item in reply_status( + root, route, + include_result_key=lambda key: "handoff." + _hash([request_id, key]) in pending_message_ids, + ): message_id = "handoff." + _hash([request_id, item.get("result_key", item["phase"])]) if message_id not in pending_message_ids: continue diff --git a/loopx/control_plane/collaboration/inbox.py b/loopx/control_plane/collaboration/inbox.py index 2e3107327a..29f1011caf 100644 --- a/loopx/control_plane/collaboration/inbox.py +++ b/loopx/control_plane/collaboration/inbox.py @@ -538,9 +538,16 @@ def iter_result_paths(folder: Path): DELIVERY_ERRORS = set(_delivery_contract["delivery_errors"]) -def reply_status(root, row): +def reply_status(root, row, *, include_result_key=None): + """Read current delivery state, optionally selecting result identities first. + + The selector narrows IO only; retained results use the same validation and + unreadable-state handling. No result content or status is cached. + """ result = [] for path in result_paths(_root(root) / "replies" / row["request_id"]): + if include_result_key is not None and not include_result_key(path.stem): + continue phase = "decision" if path.stem == "decision" or path.stem.startswith("decision-") else "conclusion" state_path = path.with_name(path.stem + ".delivery.json") try: diff --git a/tests/test_manager_context_roundtrip.py b/tests/test_manager_context_roundtrip.py index 7dce9ea044..7d82311906 100644 --- a/tests/test_manager_context_roundtrip.py +++ b/tests/test_manager_context_roundtrip.py @@ -1153,6 +1153,89 @@ def test_history_original_receipt_is_only_a_locator_not_route_authority(returned assert project_chat_return_deliveries(root, sid, messages, source_store=store) == messages +def test_history_fallback_reads_only_visible_result_states(returned_history, monkeypatch): + from loopx.control_plane.collaboration import inbox + + root, _, store, session, turn, receipt = returned_history + sid, rid = session["session_id"], receipt["request_id"] + # Lost locators and an unmatched legacy message must retain the exhaustive + # fallback, without deserializing all other requests' complete answer bodies. + store.update_turn(sid, turn["turn_id"], response=None) + store.append_message(sid, role="agent", origin="manager_followup", + message_id="handoff." + "0" * 64, text="Legacy result without a known receipt") + replies = _root(root) / "replies" + for index in range(30): + other = f"{index:064x}" + _write(_root(root) / "roundtrips" / f"{other}.json", {"request_id": other, "session_id": sid}) + for version in range(4): + _write(replies / other / f"conclusion-{version:08}.json", { + "phase": "conclusion", "created_at": "2026-09-14T00:00:00+00:00", + "text": "A complete off-page answer. " * 1000, + }) + expected = {"schema_version": "manager_return_delivery_status_v0", **reply_status(root, receipt)[0]} + before = {path: path.read_bytes() for path in root.rglob("*.json")} + original_read, reads = inbox._read, [] + + def read(path): + if path.parent.parent == replies: + reads.append(path) + return original_read(path) + + monkeypatch.setattr(inbox, "_read", read) + messages = store.messages(sid) + projected = project_chat_return_deliveries(root, sid, messages, source_store=store) + returned = [message for message in projected if message.get("return_delivery")] + assert len(returned) == 1 and returned[0]["return_delivery"] == expected + assert set(reads) == {replies / rid / "conclusion.json", replies / rid / "conclusion.delivery.json"} + assert {path: path.read_bytes() for path in root.rglob("*.json")} == before + assert store.messages(sid) == messages + + +def test_paged_history_filters_result_io_and_rereads_late_state(frozen_private_reply, private_snapshot, monkeypatch): + from loopx.control_plane.collaboration import inbox + + store, session, turn, *_ = frozen_private_reply + root, sid, rid = store.root.parent, session["session_id"], "e" * 64 + replies = _root(root) / "replies" / rid + _write(_root(root) / "roundtrips" / f"{rid}.json", {"request_id": rid, "session_id": sid}) + body = "Original full body. " * 1000 + keys = ["decision", "conclusion-00000001"] + for index, key in enumerate(keys): + _write(replies / f"{key}.json", {"created_at": "2026-09-14T00:00:00+00:00", "text": body}) + _write(replies / f"{key}.delivery.json", {"status": "delivered"}) + store.append_message(sid, role="agent", origin="manager_followup", + message_id="handoff." + _hash([rid, key]), turn_id=turn["turn_id"], text=body) + if index == 0: + for filler in range(205): + store.append_message(sid, role="agent", message_id=f"filler-{filler}", text=f"Earlier row {filler}") + original_read, reads = inbox._read, [] + + def read(path): + if path.parent == replies: + reads.append(path.name) + return original_read(path) + + monkeypatch.setattr(inbox, "_read", read) + first = private_snapshot(message_limit=200) + assert set(reads) == {"decision.json", "decision.delivery.json"} + assert next(message for message in first["messages"] if message.get("return_delivery"))["text"] == body + _write(replies / "conclusion-00000001.delivery.json", { + "status": "verification_required", "error": "provider_delivery_unverified", + }) + reads.clear() + tail = private_snapshot(message_limit=200, message_cursor=first["message_page"]["next_cursor"]) + returned = next(message for message in tail["messages"] if message.get("return_delivery")) + assert returned["text"] == body + assert returned["return_delivery"]["result_key"] == "conclusion-00000001" + assert returned["return_delivery"]["status"] == "verification_required" + assert set(reads) == {"conclusion-00000001.json", "conclusion-00000001.delivery.json"} + (replies / "conclusion-00000001.delivery.json").write_text("{") + fresh = private_snapshot(message_limit=200, message_cursor=first["message_page"]["next_cursor"]) + failed = next(message for message in fresh["messages"] if message.get("return_delivery")) + assert failed["return_delivery"]["status"] == "explicit_unverified" + assert failed["return_delivery"]["error"] == "delivery_state_unreadable" + + def test_provider_verification_outage_retries_read_only_without_resend(flow): root, registry, store, create = flow _, _, receipt = create(True)