Skip to content

Keep private replies moving when an App mirror journal is unreadable #8562

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
huangruiteng merged 1 commit into from
Nov 11, 2026
Merged

Keep private replies moving when an App mirror journal is unreadable #8562

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
55 changes: 53 additions & 2 deletions loopx/extensions/lark/manager_returns.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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(
Expand Down Expand Up @@ -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,
Expand All @@ -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:
Expand Down
103 changes: 43 additions & 60 deletions loopx/extensions/lark/private_conversations.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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]:
Expand Down Expand Up @@ -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.

Expand Down Expand Up @@ -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"],
Expand All @@ -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]]:
Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -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):
Expand All @@ -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"))
Expand Down
48 changes: 48 additions & 0 deletions tests/test_lark_private_conversations.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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"
Expand Down
Loading