diff --git a/loopx/extensions/lark/manager_returns.py b/loopx/extensions/lark/manager_returns.py index ff63fa5079..2dd1c34f2c 100644 --- a/loopx/extensions/lark/manager_returns.py +++ b/loopx/extensions/lark/manager_returns.py @@ -16,6 +16,8 @@ from .outbound import present_lark_conversation_text from ...capabilities.manager_context import authority from ...capabilities.manager_context.roundtrip import ReturnResolutionBlocked +from ...file_lock import exclusive_file_lock +from .conversation_identity import identity_ref def _resolve_return( @@ -89,6 +91,55 @@ def _private_return_authority(*, root: Path, registry: Path, route: dict[str, An ) +def _verify_web_mirror(*, route: dict[str, Any], session: dict[str, Any], turn: dict[str, Any], + text: str, private_transport: Any, attempt: Mapping[str, str | None] | None, + record_attempt: Callable[[Mapping[str, str | 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. + """ + saved = session["steward_context"] + key = identity_ref(saved["binding_id"], session["session_id"]) + path = private_transport.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 = private_transport._read_web_projection_journal(path) + canonical = private_transport.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 private_transport._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 present_lark_conversation_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, present_lark_conversation_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=private_transport.runtime_root, config_path=private_transport._inbox(row), + message_id=row["event"]["message_id"], text=phase["text"], attempt=mirrored, + reply_placement="chat_root", delivery_id=private_transport._web_projection_delivery_id(row, message), + runner=private_transport._reply_runner, finalize_reactions=False, resume_parts=False, + source_authority_verifier=lambda profile, chat, ref: private_transport._source_authority(row, profile, chat, ref)) + if message not in private_transport._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_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, @@ -97,8 +148,8 @@ def _web_return(*, root: Path, registry: Path, route: dict[str, Any], session: d 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) + result = _verify_web_mirror(route=route, session=session, turn=turn, text=text, + private_transport=private_transport, attempt=attempt, record_attempt=record_attempt) except ReturnResolutionBlocked: raise except (KeyError, ValueError, OSError) as exc: diff --git a/loopx/extensions/lark/private_conversations.py b/loopx/extensions/lark/private_conversations.py index e7d2f844a7..0466d0a5d3 100644 --- a/loopx/extensions/lark/private_conversations.py +++ b/loopx/extensions/lark/private_conversations.py @@ -1447,8 +1447,7 @@ def pending_body_paths(self) -> list[Path]: """ paths = [path for path, row, _ in self._body_discovery.rows() if row["status"] != "delivered"] - return [*paths, *(path for path in sorted((self.root / "web-projections").glob("*.json")) - if _read_json(path).get("status") == "pending")] + return [*paths, *self._pending_web_projection_paths()] @effect_runtime_request_scope() def pending_delivery_paths(self) -> list[Path]: @@ -1476,6 +1475,31 @@ def is_recovery_path(self, path: Path) -> bool: """ return path.parent != self.root / "web-projections" and _read_json(path).get("status") == "delivered" + def _pending_web_projection_paths(self) -> list[Path]: + paths = [] + for path in sorted((self.root / "web-projections").glob("*.json")): + try: + if _read_json(path).get("status") == "pending": + paths.append(path) + except (OSError, ValueError): + # An unreadable journal remains unresolved, but must not stop + # discovery of other original conversations' replies. + continue + return paths + + @staticmethod + def _read_web_projection_journal(path: Path) -> dict[str, Any]: + row = _read_json(path) + if (row.get("schema_version") != "lark_private_web_projection_v0" + or row.get("projection_id") != path.stem + or row.get("status") not in ("pending", "delivered", "partial") + or any(not isinstance(row.get(key), str) or not row[key] + for key in ("request_ref", "binding_id", "profile", "session_id")) + or any(not isinstance(row.get(key), dict) for key in ("source", "event", "deliveries")) + or any(not isinstance(phase, dict) for phase in row["deliveries"].values())): + raise ValueError("Original App mirror journal is unavailable") + return row + def _web_projection_paths(self) -> list[Path]: """Discover committed App messages, using the existing reply workers. @@ -1512,7 +1536,7 @@ def _web_projection_paths(self) -> list[Path]: selected = self._web_projection_transcript(sid) if selected: with exclusive_file_lock(path, operation="discover_web_conversation_projection"): - row = _read_json(path) if path.exists() else { + row = self._read_web_projection_journal(path) if path.exists() else { "schema_version": "lark_private_web_projection_v0", "projection_id": key, "request_ref": source["request_ref"], "binding_id": source["binding_id"], "profile": source["profile"], @@ -1527,7 +1551,7 @@ def _web_projection_paths(self) -> list[Path]: # An unreadable source is unknown. Keep any prior outbox; # it retains fresh source and canonical-message checks. continue - return [path for path in sorted(root.glob("*.json")) if _read_json(path).get("status") == "pending"] + return self._pending_web_projection_paths() def _web_projection_transcript(self, session_id: str, *, inputs_only: bool = False, verify_turns: bool = False) -> list[dict[str, Any]]: @@ -1588,58 +1612,6 @@ 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 @@ -1723,7 +1695,12 @@ def _reconcile_web_projection(self, path: Path) -> int: # The existing worker pool serializes this outbox with a durable fence. # Recovery reads the same provider attempt; it never starts a Core Turn. with exclusive_file_lock(path, operation="deliver_web_conversation_projection"): - row = _read_json(path) + try: + row = self._read_web_projection_journal(path) + except (OSError, ValueError): + # Never overwrite the sole evidence of an unknown send with + # a blocker or a new outbox when its journal cannot be read. + return 0 try: messages = self._web_projection_messages(row) pending = next((message for message in messages @@ -1772,8 +1749,11 @@ def reconcile_app_feedback(self) -> int: return 0 for path in sorted((self.root / "web-projections").glob("*.json")): with exclusive_file_lock(path.with_suffix(".feedback"), operation="reconcile_web_input_feedback"): - with exclusive_file_lock(path, operation="read_web_input_feedback"): - row = _read_json(path) + try: + with exclusive_file_lock(path, operation="read_web_input_feedback"): + row = self._read_web_projection_journal(path) + except (OSError, ValueError): + continue try: due = self._has_app_feedback_work(row) except (OSError, ValueError, KeyError): @@ -1796,7 +1776,10 @@ def reconcile_app_feedback(self) -> int: if "feedback" not in phase or phase["feedback"] == prior_feedback: continue with exclusive_file_lock(path, operation="record_web_input_feedback"): - current = _read_json(path) + try: + current = self._read_web_projection_journal(path) + except (OSError, ValueError): + continue saved = current["deliveries"].get(phase_key, {}) if (all(current.get(key) == row.get(key) for key in ("request_ref", "binding_id", "profile", "source", "event", "session_id")) diff --git a/tests/test_lark_private_conversations.py b/tests/test_lark_private_conversations.py index d9630e8153..dec93d2c03 100644 --- a/tests/test_lark_private_conversations.py +++ b/tests/test_lark_private_conversations.py @@ -2692,6 +2692,54 @@ def test_app_projection_recovers_original_receipt_and_revocation_blocks_remainin runtime.close() +@pytest.mark.parametrize("unreadable", [b'{"status":', b"\xff"]) +def test_unreadable_app_mirror_does_not_hold_other_private_replies(ordinary, unreadable): # noqa: F811 + store, runtime, provider, transport, original, turns, _ = _app_projection_journey(ordinary) + try: + transport._web_projection_scan_at = 0 + [mirror] = transport._web_projection_paths() + complete_receipt = mirror.read_bytes() + mirror.write_bytes(unreadable) + + event = provider.event("steward-app", "independent-private-reply", "An independent original request") + transport.admit("steward-app", event) + ref = identity_ref(transport._binding("steward-app")["provider_ref"], event["message_id"]) + native = transport.core.read_request(ref) + assert runtime.wait_for_turn(session_id=native["session_id"], turn_id=native["turn_id"], timeout_sec=10)["status"] == "completed" + healthy = transport.root / f"{ref}.json" + for discover in (transport.pending_body_paths, transport.pending_delivery_paths): + paths = discover() + assert healthy in paths and mirror not in paths + + # A previously scheduled worker can still see damage after discovery. + before = list(provider.writes) + assert transport.reconcile_request(mirror) == 0 + assert mirror.read_bytes() == unreadable + assert provider.writes == before + + requests = ordinary[4].read_bytes() + transport.reconcile() + terminal = json.loads(healthy.read_text())["deliveries"]["terminal"] + assert terminal["verified"] + assert provider.messages[terminal["attempt"]["message_ref"]]["chat_id"] == event["chat_id"] + assert mirror.read_bytes() == unreadable + assert not any("（App）" in text for _, text in provider.writes) + + # Restoring the exact receipt in the disposable fixture makes the + # original App work available to a newly constructed worker again. + mirror.write_bytes(complete_receipt) + restarted = LarkPrivateConversations(controller=runtime, runtime_root=transport.runtime_root, + runner=provider, cli_bin="lark-cli") + expected = [row for row in store.messages(original["session_id"]) if row.get("turn_id") in turns] + _drain_app_projection(restarted, lambda: sum("（App）" in text for _, text in provider.writes) == len(expected)) + sent = list(provider.writes) + restarted.reconcile() + assert provider.writes == sent + assert ordinary[4].read_bytes() == requests + finally: + runtime.close() + + def test_app_long_return_keeps_all_parts_and_does_not_repeat_identical_parts(ordinary): # noqa: F811 store, runtime, provider, transport, original, turns, before = _app_projection_journey(ordinary) text = ("A repeated output paragraph with its complete evidence.\n" * 4000) + "UNIQUE FINAL END"