fix(lark): retain owner messages through identity outages

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):

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

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)

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