diff --git a/loopx/capabilities/manager_context/roundtrip.py b/loopx/capabilities/manager_context/roundtrip.py index bf0e2c56bd..d18889a854 100644 --- a/loopx/capabilities/manager_context/roundtrip.py +++ b/loopx/capabilities/manager_context/roundtrip.py @@ -910,7 +910,10 @@ def record_attempt(value): next_state = _retry_state( state, now, - error="original_route_or_return_delivery_unavailable", + # Before a provider write, preserve the Core-classified + # resolution cause without changing its retry lifecycle. + error=exc.reason if isinstance(exc, ReturnResolutionBlocked) + else "original_route_or_return_delivery_unavailable", ) try: _write_exact_return_state( @@ -1305,7 +1308,8 @@ def record_attempt(value): { "status": "retry_pending", "attempts": attempts, - "error": "original_route_or_return_delivery_unavailable", + "error": exc.reason if isinstance(exc, ReturnResolutionBlocked) + else "original_route_or_return_delivery_unavailable", "retry_at": ( now + timedelta(seconds=min(300, 5 * 2 ** min(attempts, 6))) ).isoformat(), diff --git a/tests/control_plane/test_bound_owner_web_input.py b/tests/control_plane/test_bound_owner_web_input.py index 28a68e53fc..b39e5d98ac 100644 --- a/tests/control_plane/test_bound_owner_web_input.py +++ b/tests/control_plane/test_bound_owner_web_input.py @@ -273,6 +273,57 @@ def test_http_bound_steward_keeps_web_provenance_at_coordination_root(bound_chat external_source_policy(store.root.parent, session, turn) +def test_http_web_handoff_reports_missing_provider_source_without_retargeting(bound_chat): + from types import SimpleNamespace + from loopx.capabilities.manager_context import acknowledge + from loopx.capabilities.manager_context.roundtrip import drain, report, reply_status + from loopx.extensions.lark.manager_returns import LarkManagerReturnTransport + from loopx.extensions.lark.private_conversations import LarkPrivateConversations + + store, runtime, _, _, post, open_bound, capture, canonical = bound_chat + fake = Path(runtime.codex_bin) + # Only model output is synthetic; HTTP admission, source grants, handoff, + # canonical inbox, Chat storage and the Lark return owner remain real. + needle = '"message": "Runtime response.",' + source = fake.read_text() + assert source.count(needle) == 1 + fake.write_text(source.replace(needle, needle + + '\n "context_handoff": {"goal_id": "goal-a", "agent_id": "agent-a"},')) + _, session = open_bound("steward") + sid = session["session_id"] + message = "Ask the registered worker to check this source and return here." + status, accepted = post(session, {"message": message, "client_turn_id": "local-worker-request"}) + assert status == 202, accepted + turn = runtime.wait_for_turn(session_id=sid, turn_id=accepted["turn_id"], timeout_sec=15) + assert turn["status"] == "completed", turn + receipt = turn["response"]["context_handoff_receipt"] + assert receipt["status"] == "delivered" + assert turn["origin"] == "web" and canonical != store.root.parent + acknowledge(canonical, "goal-a", "agent-a", receipt["request_id"], "adopt", "Checked source", + registry=runtime.registry_path, caller_goal_ref=receipt.get("goal_ref")) + report(canonical, "goal-a", "agent-a", receipt["request_id"], "conclusion", "Checked the original source.", + registry=runtime.registry_path, caller_goal_ref=receipt.get("goal_ref")) + + def forbidden_provider(*_): + pytest.fail("A Web request cannot fabricate a provider event or guess a prior message") + + private = LarkPrivateConversations(controller=runtime, runtime_root=store.root.parent, + runner=forbidden_provider, cli_bin="synthetic-provider") + server = SimpleNamespace(registry_path=runtime.registry_path, lark_private_conversations=private, + lark_goal_topic_runtime=SimpleNamespace(snapshot_provider=forbidden_provider)) + drain(canonical, runtime.registry_path, store, LarkManagerReturnTransport(server, canonical)) + delivery = reply_status(canonical, receipt)[0] + assert delivery["status"] == "retry_pending" + assert delivery["error"] == "original_route_unavailable" + returned = [row for row in store.messages(sid) if row.get("origin") == "manager_followup"] + assert len(returned) == 1 and returned[0]["turn_id"] == turn["turn_id"] + status, replay = post(session, {"message": message, "client_turn_id": "local-worker-request"}) + assert status == 202 and not replay["created"] and replay["turn_id"] == turn["turn_id"] + assert store.load_session(sid)["upstream_thread_id"] == session["upstream_thread_id"] + calls = [json.loads(line) for line in capture.read_text().splitlines()] + assert sum(row.get("method") == "turn/start" for row in calls) == 1 + + @pytest.mark.parametrize("mutation", ["operator_ref", "provider_ref", "verified", "workspace", "revoked", "closed", "body_origin"]) def test_local_input_denies_lost_source_and_client_supplied_identity(bound_chat, mutation): store, runtime, bindings, observed, post, open_bound, _, canonical = bound_chat diff --git a/tests/control_plane/test_peer_source_store.py b/tests/control_plane/test_peer_source_store.py index 933b59716f..c82eb2d2ad 100644 --- a/tests/control_plane/test_peer_source_store.py +++ b/tests/control_plane/test_peer_source_store.py @@ -176,6 +176,64 @@ def test_split_store_handoff_can_forward_source_context(tmp_path, monkeypatch, p assert not (root / "chat").exists(), "A context read must not create another Chat store" +@pytest.mark.parametrize("provider", ["file", "sqlite", "source_session"]) +@pytest.mark.parametrize("reason,typed", [ + ("original_route_unavailable", True), + ("return_authorization_unavailable", True), + ("initial_delivery_receipt_unavailable", True), + ("original_route_unavailable", False), +]) +def test_prewrite_return_retains_typed_cause_and_recovers_original_result( + tmp_path, monkeypatch, provider, reason, typed, +): + from datetime import datetime, timedelta, timezone + from loopx.capabilities.manager_context.roundtrip import ReturnResolutionBlocked, reply_status + + root, registry, store, session, turn, receipt = source_request(tmp_path, monkeypatch, provider=provider) + rid = receipt["request_id"] + cli(root, registry, "agent-a", "acknowledge", "--request-id", rid, + "--decision", "adopt", "--reason", "Read the original source") + cli(root, registry, "agent-a", "report", "--request-id", rid, + "--reply-text", "Checked the original source and its result.") + failed_reads, writes = [], [] + unavailable = True + + def sender(*_): + if unavailable: + failed_reads.append(reason) + # Prose with the same spelling is still an untyped outage. Only + # the adapter's Core-classified reason may enter public readback. + if typed: + raise ReturnResolutionBlocked(reason, "Synthetic source is temporarily unavailable") + raise ValueError(reason) + writes.append("original audience") + return {"reply_verified": True, "idempotency_key": "synthetic-provider-receipt"} + + now = datetime.now(timezone.utc) + expected = reason if typed else "original_route_or_return_delivery_unavailable" + for offset in (0, 10): + drain(root, registry, store, sender, now=now + timedelta(minutes=offset)) + status = reply_status(root, receipt)[0] + assert status["status"] == "retry_pending" and status["error"] == expected + public = project_chat_session_snapshot(root, store, session["session_id"], registry=registry) + returned = [row for row in public["messages"] if row.get("origin") == "manager_followup"] + assert len(returned) == 1 and returned[0]["turn_id"] == turn["turn_id"] + assert returned[0]["return_delivery"]["error"] == expected + assert failed_reads == [reason, reason] and writes == [] + # Source repair resumes the original retry clock and conclusion. Keeping + # its typed cause must not terminalize a pending initial provider receipt. + unavailable = False + recovered = ChatSessionStore(store.root.parent) + drain(root, registry, recovered, sender, now=now + timedelta(minutes=20)) + drain(root, registry, recovered, sender, now=now + timedelta(minutes=30)) + assert reply_status(root, receipt)[0]["status"] == "delivered" + assert writes == ["original audience"] + returned = [row for row in recovered.messages(session["session_id"]) + if row.get("origin") == "manager_followup"] + assert len(returned) == 1 and returned[0]["turn_id"] == turn["turn_id"] + assert recovered.turn_for_client(session["session_id"], turn["client_turn_id"])["turn_id"] == turn["turn_id"] + + def route_path(root, receipt): return _root(root) / "roundtrips" / (receipt["request_id"] + ".json")