From 3108bafa0506830cf4a30347c4459ef0f7c2b0cd Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Mon, 19 Oct 2026 21:46:48 +0800 Subject: [PATCH] fix(lark): retain owner messages through identity outages Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> --- .../native_chat/external_conversations.py | 11 +- .../extensions/lark/private_conversations.py | 29 ++++- tests/test_lark_private_conversations.py | 117 ++++++++++++++++++ tests/test_native_steward_private.py | 20 +++ 4 files changed, 170 insertions(+), 7 deletions(-) diff --git a/loopx/capabilities/native_chat/external_conversations.py b/loopx/capabilities/native_chat/external_conversations.py index 1326ba2dc0..6e6fb8bc6b 100644 --- a/loopx/capabilities/native_chat/external_conversations.py +++ b/loopx/capabilities/native_chat/external_conversations.py @@ -168,7 +168,7 @@ def _admit_prepared(self, path: Path, row: dict[str, Any], selected: dict[str, A controller = self.controller self.bindings.ensure_delivery_scope(selected) if not row.get("routing_recorded"): - choices = [item for item in self.pending() if item["binding_id"] == row["binding_id"] + choices = [item for item in self.pending() if item.get("binding_id") == row["binding_id"] and item["source"]["source_ref"] == row["source"]["source_ref"] and item.get("selection_recorded")] previous = max(choices, key=lambda item: item.get("selection_order", 0)) if choices else None @@ -262,7 +262,7 @@ def _admit_prepared(self, path: Path, row: dict[str, Any], selected: dict[str, A row["response"] = "当前没有已授权且可用的 Agent。请在本机 Chat → 设置 → Lark，为此 App 明确授权已有 attached Session；名称相似不会获得授权。" if not choices else "可选 Agent（使用确切命令）：\n" + "\n".join(f"{item['agent_id']} · {item['goal_id']}\n/agent {item['target_ref']}" for item in choices) if row.get("selection_recorded"): row["selection_order"] = 1 + max((item.get("selection_order", 0) for item in self.pending() - if item["binding_id"] == row["binding_id"] and item["source"]["source_ref"] == row["source"]["source_ref"]), default=0) + if item.get("binding_id") == row["binding_id"] and item["source"]["source_ref"] == row["source"]["source_ref"]), default=0) row.update(status="command_completed", session_id=None, turn_id=None) elif operation == "steward_action": parsed = self.bindings._core("collaboration.steward.command", {"command": row["command"], "message": row["message"]}) @@ -385,8 +385,9 @@ def commission_evidence(self, session: dict[str, Any]) -> list[dict[str, Any]]: latest: dict[str, dict[str, Any]] = {} for row in sorted(self.pending(), key=lambda row: row.get("created_at", "")): resources = row.get("commission_resources") or {} - if (row["binding_id"] == saved["binding_id"] and row["source"]["source_ref"] == saved["source_ref"] - and resources.get("goal_id") in allowed): + if (resources.get("goal_id") in allowed + and row["binding_id"] == saved["binding_id"] + and row["source"]["source_ref"] == saved["source_ref"]): latest[resources["goal_id"]] = row facts = [] for goal_id, row in latest.items(): @@ -603,6 +604,8 @@ def record_delivery(self, request_ref: str, *, session_id: str | None, turn_id: path = self.root / f"{request_ref}.json" with exclusive_file_lock(path, operation="record_external_chat_delivery"): row = _read_json(path) + if row.get("request_ref") != request_ref: + raise ValueError("delivery request journal unavailable") if row.get("commission_adoption_pending"): raise ValueError("commission adoption is pending; notification cannot settle the effect") if (row.get("session_id"), row.get("turn_id")) != (session_id, turn_id): diff --git a/loopx/extensions/lark/private_conversations.py b/loopx/extensions/lark/private_conversations.py index 5e2fbc5fa0..bfae758b01 100644 --- a/loopx/extensions/lark/private_conversations.py +++ b/loopx/extensions/lark/private_conversations.py @@ -558,13 +558,28 @@ def admit(self, profile: str, event: dict[str, Any], *, request = identity_ref(binding["provider_ref"], event["message_id"]) control_root = self._admitted_control_root({"request_ref": request, "source": source, "event": event}, binding) - selected = (None if control_root else self.bindings.resolve( - binding_id=binding["binding_id"], **source, include_observation=True)) except (KeyError, StopIteration, ValueError): return {"status": "audience_rejected"} request = identity_ref(binding["provider_ref"], event["message_id"]) path = self.root / f"{request}.json" with exclusive_file_lock(path, operation="capture_private_chat_request"): + # Retain a candidate from the locally bound owner before provider + # identity IO can fail. This private receipt grants no execution: + # recovery still verifies the App, current owner, source and Core + # grant before creating a Turn. Groups and other senders keep their + # existing pre-capture privacy boundary. + if (not path.exists() and binding.get("enabled") is True + and binding.get("audience", "owner") == "owner" + and source.get("private_human_message") is True + and source["sender_ref"] == binding["operator_ref"]): + _atomic_write_json(path, {"schema_version": "lark_private_chat_delivery_v0", + "request_ref": request, "profile": profile, "binding_id": binding["binding_id"], + "source": source, "event": event, "deliveries": {}, "status": "captured"}) + try: + selected = (None if control_root else self.bindings.resolve( + binding_id=binding["binding_id"], **source, include_observation=True)) + except (KeyError, StopIteration, ValueError): + return {"status": "audience_rejected"} if path.exists(): record = _read_json(path) # event_id can change on redelivery; canonical message content @@ -578,6 +593,11 @@ def admit(self, profile: str, event: dict[str, Any], *, "event": event, "deliveries": {}, "status": "captured"} if binding.get("audience") != "group": _atomic_write_json(path, record) + # A transport-only rejection never owned a Core request. Its + # verified notice stays terminal on provider redelivery too. + if (record["status"] == "delivered" + and "session_id" not in record and "turn_id" not in record): + return {"status": "command_rejected"} if control_root: result = self._admit_group_control(path, record, binding, control_root, defer_background_stop=defer_background_stop) @@ -1163,7 +1183,10 @@ def inbox() -> Path: config = inbox() acknowledge_lark_event_inbox(project=self.runtime_root, config_path=config, message_ids=[record["event"]["message_id"]], execute=True) - self.core.record_delivery(record["request_ref"], session_id=record.get("session_id"), turn_id=record.get("turn_id")) + if native_path.exists(): + self.core.record_delivery(record["request_ref"], session_id=record.get("session_id"), turn_id=record.get("turn_id")) + elif record["status"] != "rejected": + raise ValueError("delivery request journal unavailable") record["status"] = "delivered" _atomic_write_json(path, record) return 1 diff --git a/tests/test_lark_private_conversations.py b/tests/test_lark_private_conversations.py index 6eb7030703..0f5e0fd65b 100644 --- a/tests/test_lark_private_conversations.py +++ b/tests/test_lark_private_conversations.py @@ -305,6 +305,123 @@ def unavailable(**kwargs): runtime.close() +@pytest.mark.parametrize("unavailable", ["bot_ready", "owner_ready"]) +def test_owner_event_survives_identity_outage_and_recovers_once(ordinary, unavailable): # noqa: F811 + store, runtime, provider, transport = connect(ordinary) + event = provider.event("notes-app", "identity_outage", "Explain this project") + try: + setattr(provider, unavailable, False) + assert transport.admit("notes-app", event)["status"] == "audience_rejected" + path = next(transport.root.glob("*.json")) + captured = json.loads(path.read_text()) + assert captured["status"] == "captured" and captured["event"] == event + assert transport.pending_delivery_paths() == [path] + assert store.list_sessions() == [] and transport.core.pending() == [] + assert provider.writes == [] and provider.reaction_creates == [] + assert transport.reconcile_request(path) == 0 + assert json.loads(path.read_text()) == captured + + # The normal scanner can recover the same receipt after a restart; + # no source-history replay or second event consumer is involved. + restored = LarkPrivateConversations(controller=runtime, runtime_root=store.root.parent, + runner=provider, cli_bin="lark-cli") + setattr(provider, unavailable, True) + restored.reconcile_request(path) + row = restored.core.pending()[0] + result = runtime.wait_for_turn(session_id=row["session_id"], turn_id=row["turn_id"], timeout_sec=10) + assert result["status"] == "completed" + restored.reconcile() + delivered = json.loads(path.read_text()) + assert delivered["status"] == "delivered" and delivered["event"] == event + assert delivered["deliveries"]["terminal"]["verified"] is True + assert len(store.list_sessions()) == 1 + assert [m["text"] for m in store.messages(row["session_id"]) if m["role"] == "user"] == [event["content"]] + provider_messages = json.loads(json.dumps(provider.messages)) + provider_writes = list(provider.writes) + restored.admit("notes-app", event) + restored.reconcile() + assert provider.messages == provider_messages and provider.writes == provider_writes + packets = [json.loads(line) for line in ordinary[4].read_text().splitlines()] + assert sum(packet["method"] == "turn/start" for packet in packets) == 1 + finally: + runtime.close() + + +@pytest.mark.parametrize("change", ["revoked", "source_sender_changed", "owner_changed"]) +def test_unverified_owner_receipt_does_not_grant_later_execution(ordinary, monkeypatch, change): # noqa: F811 + store, runtime, provider, transport = connect(ordinary) + event = provider.event("notes-app", "unverified_boundary", "Explain this project") + try: + provider.owner_ready = False + assert transport.admit("notes-app", event)["status"] == "audience_rejected" + path = next(transport.root.glob("*.json")) + captured = json.loads(path.read_text()) + provider.owner_ready = True + if change == "revoked": + transport.bindings.disconnect(captured["binding_id"], + expected_revision=transport.bindings.read()["revision"]) + elif change == "source_sender_changed": + provider.messages[event["message_id"]]["sender"]["id"] = "ou_other" + else: + observe = transport.bindings.observe + monkeypatch.setattr(transport.bindings, "observe", lambda profile: { + **observe(profile), "operator_ref": "f" * 24}) + assert transport.reconcile_request(path) == 0 + assert store.list_sessions() == [] and transport.core.pending() == [] + assert provider.writes == [] and provider.reaction_creates == [] + assert json.loads(path.read_text())["status"] == "captured" + finally: + runtime.close() + + +@pytest.mark.parametrize("group", [False, True]) +def test_identity_outage_does_not_precapture_groups_or_other_owners(ordinary, group): # noqa: F811 + store, runtime, provider, transport = (connect_group if group else connect)(ordinary) + event = provider.topic("unverified_group", "Explain this project") if group else provider.event( + "notes-app", "unverified_other_owner", "Explain this project") + if not group: + event["sender_id"] = "ou_other" + try: + provider.bot_ready = False + assert transport.admit("notes-app", event)["status"] == "audience_rejected" + assert list(transport.root.glob("*.json")) == [] + assert store.list_sessions() == [] and transport.core.pending() == [] + assert provider.writes == [] and provider.reaction_creates == [] + finally: + runtime.close() + + +def test_rejected_source_returns_notice_without_creating_core_execution(ordinary, monkeypatch): # noqa: F811 + store, runtime, provider, transport = connect(ordinary) + event = provider.event("notes-app", "rejected_before_core", "Explain this project") + try: + admit = transport.core.admit + def reject(**_kwargs): + raise ValueError("identity changed before Core admission") + + monkeypatch.setattr(transport.core, "admit", reject) + assert transport.admit("notes-app", event)["status"] == "command_rejected" + path = next(transport.root.glob("*.json")) + captured = json.loads(path.read_text()) + native_path = transport.core.root / f"{captured['request_ref']}.json" + assert not native_path.exists() + assert transport.reconcile_request(path) == 1 + delivered = json.loads(path.read_text()) + assert delivered["status"] == "delivered" + assert delivered["deliveries"]["terminal"]["verified"] is True + assert len(provider.writes) == 1 and not native_path.exists() + assert store.list_sessions() == [] + with pytest.raises(ValueError, match="request journal unavailable"): + transport.core.record_delivery(captured["request_ref"], session_id=None, turn_id=None) + assert not native_path.exists() + monkeypatch.setattr(transport.core, "admit", admit) + assert transport.admit("notes-app", event)["status"] == "command_rejected" + assert not native_path.exists() and store.list_sessions() == [] + assert len(provider.writes) == 1 + finally: + runtime.close() + + @pytest.mark.parametrize("group", [False, True]) def test_completion_during_processing_receipt_returns_final_without_stale_draft(ordinary, monkeypatch, tmp_path, group): # noqa: F811 store, runtime, provider, transport = (connect_group if group else connect)(ordinary) diff --git a/tests/test_native_steward_private.py b/tests/test_native_steward_private.py index c4697fa787..ef12e320df 100644 --- a/tests/test_native_steward_private.py +++ b/tests/test_native_steward_private.py @@ -114,6 +114,26 @@ def test_personal_steward_uses_registered_portfolio_and_configured_inbox_root(st assert not (workspace / "ACTIVE_GOAL_STATE.md").exists() +def test_legacy_delivery_only_receipt_does_not_block_steward_conversation(steward): + store, runtime, provider, transport, _, capture, _ = steward + orphan = transport.core.root / ("f" * 24 + ".json") + orphan.parent.mkdir(parents=True, exist_ok=True) + orphan.write_text(json.dumps({"delivery_verified": True})) + before = orphan.read_bytes() + event = provider.event("steward-app", "after_legacy_receipt", "Check the authorized portfolio") + assert transport.admit("steward-app", event)["status"] == "durably_accepted" + row = next(r for r in transport.core.pending() if r.get("request_ref") == identity_ref( + transport.bindings.read()["bindings"][1]["provider_ref"], event["message_id"])) + assert finish(runtime, row)["status"] == "completed" + session = store.load_session(row["session_id"]) + assert transport.core.commission_evidence(session) == [] + selection = provider.event("notes-app", "select_after_legacy_receipt", "/project") + assert transport.admit("notes-app", selection)["status"] == "command_recorded" + assert orphan.read_bytes() == before + packets = [json.loads(line) for line in capture.read_text().splitlines()] + assert sum(p["method"] == "turn/start" for p in packets) == 1 + + def test_verified_empty_steward_is_not_missing_authorization_and_does_not_inherit(steward): store, runtime, provider, transport, binding, _, _ = steward transport.admit("steward-app", provider.event("steward-app", "empty", "What new work do you manage?"))