From 116c5929b4755a96132d4455d6f98474f0675dc4 Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Thu, 5 Nov 2026 04:29:27 +0800 Subject: [PATCH] fix(chat): reconcile worker returns with original App mirrors Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> --- .../capabilities/manager_context/roundtrip.py | 8 + loopx/extensions/lark/manager_returns.py | 53 ++++- .../extensions/lark/private_conversations.py | 52 +++++ .../test_lark_private_manager_returns.py | 218 ++++++++++++++++++ 4 files changed, 322 insertions(+), 9 deletions(-) diff --git a/loopx/capabilities/manager_context/roundtrip.py b/loopx/capabilities/manager_context/roundtrip.py index 4d76d93003..8431f52edf 100644 --- a/loopx/capabilities/manager_context/roundtrip.py +++ b/loopx/capabilities/manager_context/roundtrip.py @@ -450,6 +450,9 @@ def _return_transaction(host, path, phase, request): def _return_provider(plan, source, sender, record_attempt): route, session, turn = (source[key] for key in ("route", "session", "turn")) + # The typed transaction owns the canonical result identity. Transports may + # verify a mirror of this message, never guess its identity from its body. + route = {**route, "return_message_id": plan["message_id"]} attachments = plan.get("attachments") or [] if attachments: if getattr(sender, "supports_result_files", False) is not True: @@ -472,6 +475,11 @@ def _return_provider(plan, source, sender, record_attempt): sent = sender(route, session, turn, plan["text"]) else: return {"kind": "sender_unavailable"} + if isinstance(sent.get("verification"), dict) and sent.get("external_write_performed") is False: + # A provider outbox may already own publication. Its adapter records the + # existing attempt before returning a readback, including unavailable + # or mismatched results, through the ordinary verification state machine. + return {"kind": "verified", "verification": sent["verification"]} return {"kind": "sent", "reply_verified": sent.get("reply_verified"), "external_write_performed": sent.get("external_write_performed"), "idempotency_key": sent.get("idempotency_key")} diff --git a/loopx/extensions/lark/manager_returns.py b/loopx/extensions/lark/manager_returns.py index 00bc1f7635..ff63fa5079 100644 --- a/loopx/extensions/lark/manager_returns.py +++ b/loopx/extensions/lark/manager_returns.py @@ -75,6 +75,40 @@ def _resolve_return( return snapshot, routing, target_config +def _private_return_authority(*, root: Path, registry: Path, route: dict[str, Any], + session: dict[str, Any], turn: dict[str, Any], + cancelled: Callable[[], bool], private_transport: Any) -> None: + if cancelled(): + raise ValueError("manager return service stopped") + grant = authority(root, registry, session, turn) + target = {k: route[k] for k in ("goal_id", "agent_id")} + if (target not in grant["targets"] or grant.get("source_id") != route["source_id"] + or private_transport is None): + raise ReturnResolutionBlocked( + "return_authorization_unavailable", "private return authority unavailable" + ) + + +def _web_return(*, root: Path, registry: Path, route: dict[str, Any], session: dict[str, Any], + turn: dict[str, Any], text: str, cancelled: Callable[[], bool], private_transport: Any, + attempt: Mapping[str, str | None] | None = None, + record_attempt: Callable[[Mapping[str, str | None]], None] | None = None) -> dict[str, Any]: + arguments = dict(root=root, registry=registry, route=route, session=session, turn=turn, + cancelled=cancelled, private_transport=private_transport) + _private_return_authority(**arguments) + try: + result = private_transport.verify_web_return(route=route, session=session, turn=turn, + text=text, attempt=attempt, record_attempt=record_attempt) + except ReturnResolutionBlocked: + raise + except (KeyError, ValueError, OSError) as exc: + raise ReturnResolutionBlocked("original_route_unavailable", "original App return source unavailable") from exc + # Provider readback can outlast a grant or a stop. It grants no publication + # authority and must not mark a revoked request as delivered. + _private_return_authority(**arguments) + return result + + def _return_inbox(*, root: Path, registry: Path, snapshot_provider: Callable[[], dict[str, Any]], route: dict[str, Any], session: dict[str, Any], turn: dict[str, Any], cancelled: Callable[[], bool], private_transport: Any) -> tuple[Path, Path, Callable[[], bool] | None, object, str]: @@ -82,15 +116,8 @@ def _return_inbox(*, root: Path, registry: Path, snapshot_provider: Callable[[], source_verifier: Callable[[], bool] | None destination: object if isinstance(session.get("steward_context"), dict): - if cancelled(): - raise ValueError("manager return service stopped") - grant = authority(root, registry, session, turn) - target = {k: route[k] for k in ("goal_id", "agent_id")} - if (target not in grant["targets"] or grant.get("source_id") != route["source_id"] - or private_transport is None): - raise ReturnResolutionBlocked( - "return_authorization_unavailable", "private return authority unavailable" - ) + _private_return_authority(root=root, registry=registry, route=route, session=session, + turn=turn, cancelled=cancelled, private_transport=private_transport) try: config_path, record = private_transport.return_inbox(route=route, session=session, turn=turn) # The original Inbox belongs to the private Chat transport. Its @@ -140,6 +167,11 @@ def send_return( private_transport: Any = None, ) -> dict[str, Any]: text = present_lark_conversation_text(text) + if isinstance(session.get("steward_context"), dict) and turn.get("origin") == "web": + result = _web_return(root=root, registry=registry, route=route, session=session, turn=turn, + text=text, cancelled=cancelled, private_transport=private_transport, + record_attempt=delivery_attempt_recorder) + return {**result, "verification": result, "external_write_performed": False} def resolve() -> tuple[Path, Path, Callable[[], bool] | None, object, str]: return _return_inbox( root=root, @@ -229,6 +261,9 @@ def verify_return( private_transport: Any = None, ) -> dict[str, Any]: text = present_lark_conversation_text(text) + if isinstance(session.get("steward_context"), dict) and turn.get("origin") == "web": + return _web_return(root=root, registry=registry, route=route, session=session, turn=turn, + text=text, cancelled=cancelled, private_transport=private_transport, attempt=attempt) project, config_path, source_verifier, destination, message_id = _return_inbox( root=root, registry=registry, diff --git a/loopx/extensions/lark/private_conversations.py b/loopx/extensions/lark/private_conversations.py index 219071662a..a8d0c736f8 100644 --- a/loopx/extensions/lark/private_conversations.py +++ b/loopx/extensions/lark/private_conversations.py @@ -1412,6 +1412,58 @@ def _web_projection_delivery_id(row: dict[str, Any], message: dict[str, Any]) -> return hashlib.sha256(json.dumps([row["binding_id"], row["session_id"], message["message_id"]], separators=(",", ":")).encode()).hexdigest() + def verify_web_return(self, *, route: dict[str, Any], session: dict[str, Any], + turn: dict[str, Any], text: str, + attempt: Mapping[str, str | None] | None = None, + record_attempt: Callable[[Mapping[str, str | None]], None] | None = None) -> dict[str, Any]: + """Read an existing App mirror; its original worker alone owns sends. + + Manager-context supplies its typed canonical result id and owns result + settlement. Neither a matching body nor a historical verified flag is + sufficient: revalidate the source and read the exact provider attempt. + """ + from ...capabilities.manager_context.roundtrip import ReturnResolutionBlocked + + saved = session["steward_context"] + key = identity_ref(saved["binding_id"], session["session_id"]) + path = self.root / "web-projections" / f"{key}.json" + if not path.is_file(): + raise ReturnResolutionBlocked("initial_delivery_receipt_unavailable", "App mirror not yet recorded") + with exclusive_file_lock(path, operation="verify_web_conversation_return"): + row = _read_json(path) + canonical = self.core.controller.store.load_turn(session["session_id"], turn["turn_id"]) + if (route.get("session_id") != session["session_id"] or row.get("session_id") != session["session_id"] + or not canonical or canonical.get("origin") != "web" + or canonical.get("client_turn_id") != turn.get("client_turn_id") + or route.get("result_attachments")): + raise ValueError("original App return identity or media unavailable") + message = next((item for item in self._web_projection_messages(row) + if item["message_id"] == route.get("return_message_id")), None) + if (not message or message.get("origin") != "manager_followup" or message.get("role") != "agent" + or message.get("turn_id") != turn["turn_id"] or _presentation_text(message["text"]) != text + or message.get("attachments")): + raise ValueError("canonical App result changed") + phase = row["deliveries"].get("web." + message["message_id"], {}) + mirrored = phase.get("attempt") + if phase.get("verified") is not True or not isinstance(mirrored, dict): + raise ReturnResolutionBlocked("initial_delivery_receipt_unavailable", "App mirror remains unverified") + expected = f"**LoopX（App）**\n\n{message['text']}" + if phase.get("text") not in (expected, _presentation_text(expected)) or ( + attempt is not None and dict(attempt) != mirrored): + raise ValueError("App mirror intent changed") + if attempt is None and record_attempt is not None: + # Persist the original locator before IO, so a crash or outage + # resumes verification rather than starting another delivery. + record_attempt(dict(mirrored)) + result = verify_lark_inbox_reply(project=self.runtime_root, config_path=self._inbox(row), + message_id=row["event"]["message_id"], text=phase["text"], attempt=mirrored, + reply_placement="chat_root", delivery_id=self._web_projection_delivery_id(row, message), + runner=self._reply_runner, finalize_reactions=False, resume_parts=False, + source_authority_verifier=lambda profile, chat, ref: self._source_authority(row, profile, chat, ref)) + if message not in self._web_projection_messages(row): + raise ValueError("canonical App result changed during readback") + return {**result, "idempotency_key": mirrored["provider_receipt"], "external_write_performed": False} + def _web_projection_feedback(self, row: dict[str, Any], message: dict[str, Any]) -> None: if not self.reaction_feedback or message["role"] != "user": return diff --git a/tests/extensions/test_lark_private_manager_returns.py b/tests/extensions/test_lark_private_manager_returns.py index 87f6130dd9..2d2661d437 100644 --- a/tests/extensions/test_lark_private_manager_returns.py +++ b/tests/extensions/test_lark_private_manager_returns.py @@ -290,3 +290,221 @@ def test_partial_return_survives_saved_attempt_recovery_until_final_delivery(pri assert store._turn_path(session["session_id"], turn["turn_id"]).read_bytes() == original_turn assert sorted((store.root / "sessions" / session["session_id"] / "turns").glob("*.json")) == turns assert capture.read_bytes() == before_host_calls + + +@pytest.fixture +def web_return(private_return): + """Use the shipped HTTP input, File store, return pump and Lark outbox.""" + import http.client + import threading + from loopx.chat_server import ChatHTTPServer, ChatRequestHandler + from loopx.extensions.lark.conversation_identity import identity_ref + from test_lark_private_conversations import _drain_app_projection + + sender, session, _, original_route, _, provider, transport, _ = private_return + runtime, store = transport.core.controller, transport.core.controller.store + transport.runner = provider + server = ChatHTTPServer(("127.0.0.1", 0), ChatRequestHandler) + server.chat_store, server.runtime_controller = store, runtime + server.runtime_root, server.registry_path = transport.runtime_root, runtime.registry_path + server.selected_goal_id, server.verbose = None, False + server.lark_private_conversations = transport + thread = threading.Thread(target=server.serve_forever, daemon=True) + thread.start() + + def request(path, body=None): + connection = http.client.HTTPConnection(*server.server_address, timeout=15) + try: + connection.request("GET" if body is None else "POST", path, + None if body is None else json.dumps(body), headers={"Content-Type": "application/json"}) + result = connection.getresponse() + return result.status, json.loads(result.read()) + finally: + connection.close() + + try: + path = f"/api/chat/sessions/{session['session_id']}" + status, accepted = request(path + "/turns", { + "message": "Ask the same worker from the App", + "client_turn_id": "d9bf2f43-4c6c-48d0-a8c1-baf000000001"}) + assert status == 202 + turn = finish(runtime, {"session_id": session["session_id"], "turn_id": accepted["turn_id"]}) + assert turn["status"] == "completed" and turn["origin"] == "web" + session = store.load_session(session["session_id"]) + target = {key: original_route[key] for key in ("goal_id", "agent_id")} + result = deliver(sender.root, runtime.registry_path, session=session, turn=turn, + request=target, source_store=store) + route = _read_json(_root(sender.root) / "roundtrips" / (result["request_id"] + ".json")) + assert route["source_id"].startswith("web:") + acknowledge(sender.root, target["goal_id"], target["agent_id"], route["request_id"], "adopt", "Inspect this request") + mirror = transport.root / "web-projections" / ( + identity_ref(session["steward_context"]["binding_id"], session["session_id"]) + ".json") + + def flush(message_id=None): + transport._web_projection_scan_at = 0 + ids = [message_id] if message_id else [item["message_id"] for item in store.messages(session["session_id"]) + if item["turn_id"] == turn["turn_id"]] + _drain_app_projection(transport, lambda: mirror.exists() and all( + _read_json(mirror)["deliveries"].get("web." + ref, {}).get("verified") for ref in ids)) + + flush() + yield sender, session, turn, route, provider, transport, mirror, flush, request + finally: + server.shutdown() + server.server_close() + thread.join(timeout=2) + + +@pytest.mark.parametrize("long_result", [False, True]) +def test_web_result_adopts_original_mirror_and_recovers_readback_without_send(web_return, monkeypatch, long_result): + sender, session, turn, route, provider, transport, mirror, flush, request = web_return + runtime, store = transport.core.controller, transport.core.controller.store + capture = store.root.parent.parent / "requests.jsonl" + starts = capture.read_bytes() + frozen_turn = store._turn_path(session["session_id"], turn["turn_id"]).read_bytes() + remainder = {"summary": "Validate installation", "resume_when": "The same worker has the original installation proof"} + body = ("完整结果与独立证据。\n" * 1200 if long_result else "Concrete original result.") + "UNIQUE_RESULT_END" + report(sender.root, route["goal_id"], route["agent_id"], route["request_id"], "conclusion", body, + remaining_work=remainder) + writes = list(provider.writes) + drain(sender.root, runtime.registry_path, store, sender) + assert provider.writes == writes + assert reply_status(sender.root, route)[0]["error"] == "initial_delivery_receipt_unavailable" + message = next(item for item in store.messages(session["session_id"]) + if item.get("origin") == "manager_followup" and item["turn_id"] == turn["turn_id"]) + flush(message["message_id"]) + original_mirror = mirror.read_bytes() + phase = _read_json(mirror)["deliveries"]["web." + message["message_id"]] + writes = list(provider.writes) + runner = transport.runner + + def unavailable(args, cwd=None, timeout=None): + if "+messages-mget" in args and args[args.index("--message-ids") + 1].startswith("om_out_"): + return {"returncode": 1, "stdout": ""} + return runner(args, cwd, timeout) + + monkeypatch.setattr(transport, "runner", unavailable) + clock = datetime.now(timezone.utc) + timedelta(minutes=10) + drain(sender.root, runtime.registry_path, store, sender, now=clock) + state = _read_json(_root(sender.root) / "replies" / route["request_id"] / "conclusion.delivery.json") + assert state["status"] == "verification_required" and state["attempt"] == phase["attempt"] + assert provider.writes == writes + monkeypatch.setattr(transport, "runner", runner) + recovered = ChatSessionStore(store.root.parent) + restarted = LarkManagerReturnTransport(sender.server, sender.root) + drain(sender.root, runtime.registry_path, recovered, restarted, now=clock + timedelta(minutes=10)) + drain(sender.root, runtime.registry_path, recovered, restarted, now=clock + timedelta(minutes=20)) + status = reply_status(sender.root, route)[0] + assert status["status"] == "delivered" + assert provider.writes == writes and mirror.read_bytes() == original_mirror + assert capture.read_bytes() == starts + assert store._turn_path(session["session_id"], turn["turn_id"]).read_bytes() == frozen_turn + returned = [item for item in recovered.messages(session["session_id"]) + if item.get("origin") == "manager_followup" and item["turn_id"] == turn["turn_id"]] + assert len(returned) == 1 and all(value in returned[0]["text"] for value in remainder.values()) + agenda = read_delegation_turn_context(sender.root, runtime.registry_path, + route["goal_id"], route["agent_id"])["unfinished_delegations"]["items"] + assert next(item for item in agenda if item["request_id"] == route["request_id"])["remaining_work"] == remainder + code, snapshot = request(f"/api/chat/sessions/{session['session_id']}") + assert code == 200 + actual = next(item for item in snapshot["messages"] if item["message_id"] == message["message_id"]) + assert actual["text"] == returned[0]["text"] and actual["return_delivery"]["status"] == "delivered" + + +@pytest.mark.parametrize("fault", ["unknown", "revoked", "revoked-after-read", "scope", "scope-after-read", "audience", "body", "receipt"]) +def test_web_result_keeps_unknown_or_revoked_mirror_without_duplicate_delivery(web_return, monkeypatch, fault): + sender, session, turn, route, provider, transport, mirror, flush, _ = web_return + store, registry = transport.core.controller.store, transport.core.controller.registry_path + report(sender.root, route["goal_id"], route["agent_id"], route["request_id"], "conclusion", "Frozen worker result") + drain(sender.root, registry, store, sender) + message = next(item for item in store.messages(session["session_id"]) + if item.get("origin") == "manager_followup" and item["turn_id"] == turn["turn_id"]) + if fault == "unknown": + provider.lose_next_reply = True + transport._web_projection_scan_at = 0 + transport.reconcile() + phase = _read_json(mirror)["deliveries"]["web." + message["message_id"]] + assert phase["started"] and not phase.get("attempt") + else: + flush(message["message_id"]) + phase = _read_json(mirror)["deliveries"]["web." + message["message_id"]] + + def narrow_scope(): + value = transport.bindings.read() + binding = next(item for item in value["bindings"] if item["binding_id"] == session["steward_context"]["binding_id"]) + binding.update(goal_scope="selected", goal_ids=[]) + _atomic_write_json(transport.bindings.path, value) + + if fault == "revoked": + transport.bindings.disconnect(session["steward_context"]["binding_id"], + expected_revision=transport.bindings.read()["revision"]) + elif fault == "scope": + narrow_scope() + elif fault in {"revoked-after-read", "scope-after-read"}: + runner = transport.runner + + def revoked_after_read(args, cwd=None, timeout=None): + result = runner(args, cwd, timeout) + if "+messages-mget" in args and args[args.index("--message-ids") + 1] == phase["attempt"]["message_ref"]: + if fault == "scope-after-read": + narrow_scope() + else: + transport.bindings.disconnect(session["steward_context"]["binding_id"], + expected_revision=transport.bindings.read()["revision"]) + return result + + monkeypatch.setattr(transport, "runner", revoked_after_read) + elif fault == "receipt": + row = _read_json(mirror) + row["deliveries"]["web." + message["message_id"]]["attempt"] = next( + item["attempt"] for key, item in row["deliveries"].items() + if key != "web." + message["message_id"] and item.get("verified")) + _atomic_write_json(mirror, row) + else: + target = provider.messages[phase["attempt"]["message_ref"]] + if fault == "audience": + target["chat_id"] = "oc_another_audience" + else: + target["body"] = {"content": json.dumps({"zh_cn": {"title": "", "content": [[{"tag": "text", "text": "Different result"}]]}})} + writes = list(provider.writes) + for minutes in [10, 20]: + transport.reconcile() + drain(sender.root, registry, ChatSessionStore(store.root.parent), + LarkManagerReturnTransport(sender.server, sender.root), + now=datetime.now(timezone.utc) + timedelta(minutes=minutes)) + assert provider.writes == writes + assert reply_status(sender.root, route)[0]["status"] != "delivered" + assert len([item for item in store.messages(session["session_id"]) + if item.get("origin") == "manager_followup" and item["turn_id"] == turn["turn_id"]]) == 1 + + +def test_web_result_identical_updates_keep_distinct_mirrors_and_final_scope(web_return): + sender, session, turn, route, provider, transport, mirror, flush, _ = web_return + runtime, store = transport.core.controller, transport.core.controller.store + capture = store.root.parent.parent / "requests.jsonl" + starts = capture.read_bytes() + remainder = {"summary": "Installation remains", "resume_when": "Original installation evidence is available"} + ids, attempts = [], [] + for index, update in enumerate([None, "same-observation", "qualified-final"]): + body = "Repeated original observation" if index < 2 else "Original installation independently verified" + report(sender.root, route["goal_id"], route["agent_id"], route["request_id"], "conclusion", body, + remaining_work=remainder if index < 2 else None, update_id=update) + clock = datetime.now(timezone.utc) + timedelta(minutes=10 * (index + 1)) + drain(sender.root, runtime.registry_path, store, sender, now=clock) + returned = [item for item in store.messages(session["session_id"]) + if item.get("origin") == "manager_followup" and item["turn_id"] == turn["turn_id"]] + assert len(returned) == index + 1 + message = returned[-1] + ids.append(message["message_id"]) + flush(message["message_id"]) + writes = list(provider.writes) + drain(sender.root, runtime.registry_path, store, sender, now=clock + timedelta(minutes=5)) + assert all(item["status"] == "delivered" for item in reply_status(sender.root, route)) + assert provider.writes == writes + attempts.append(_read_json(mirror)["deliveries"]["web." + message["message_id"]]["attempt"]["provider_receipt"]) + assert len(set(ids)) == len(set(attempts)) == 3 + assert len([body for _, body in provider.writes if "Repeated original observation" in body]) == 2 + agenda = read_delegation_turn_context(sender.root, runtime.registry_path, + route["goal_id"], route["agent_id"])["unfinished_delegations"]["items"] + assert not any(item["request_id"] == route["request_id"] for item in agenda) + assert capture.read_bytes() == starts