fix(chat): reconcile worker returns with original App mirror receipts

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")}

loopx/extensions/lark/manager_returns.py

@@ -75,22 +75,49 @@ 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]:
     project = root
     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,

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