Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
41 changes: 39 additions & 2 deletions loopx/capabilities/manager_context/roundtrip.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
91 changes: 91 additions & 0 deletions tests/test_manager_context_roundtrip.py
Original file line number Diff line number Diff line change
Expand Up @@ -4,6 +4,7 @@
import threading
from concurrent.futures import ThreadPoolExecutor
from datetime import datetime, timedelta, timezone
from pathlib import Path

import pytest

Expand Down Expand Up @@ -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)
Expand Down
Loading