diff --git a/docs/architecture/rfcs/app-conversation-and-async-inbox-v0.md b/docs/architecture/rfcs/app-conversation-and-async-inbox-v0.md index 9959e623bf..571519e788 100644 --- a/docs/architecture/rfcs/app-conversation-and-async-inbox-v0.md +++ b/docs/architecture/rfcs/app-conversation-and-async-inbox-v0.md @@ -112,6 +112,14 @@ target that topic. Admission feedback, streamed posts, terminal edits and receip recovery use the existing delivery path with `source_thread` placement. A revoked binding or unreadable source prevents later result delivery. +A completed, target-frozen stop on a locally enabled group binding reads its two +original provider messages while Core freshly resolves the binding and App +identity. Both readonly observations must finish successfully before source, +ancestry and root-mention validation can authorize delivery. Only that invocation +reuses the verified identity; no observation is persisted or reused by a later +verification. Changed bindings discard the buffered message observation; ordinary +input, private conversations and unknown stop scope keep their existing paths. + When an ordinary Codex topic has no active Turn, `/stop` also covers background commands left by completed Turns in that same canonical Session, even after a later answer. Core freezes the original Session and canonical candidate Turn diff --git a/docs/architecture/rfcs/app-conversation-and-async-inbox-v0.zh-CN.md b/docs/architecture/rfcs/app-conversation-and-async-inbox-v0.zh-CN.md index 19f87afc2f..a35f2f971c 100644 --- a/docs/architecture/rfcs/app-conversation-and-async-inbox-v0.zh-CN.md +++ b/docs/architecture/rfcs/app-conversation-and-async-inbox-v0.zh-CN.md @@ -81,6 +81,11 @@ listener、队列或模型 runner。一个 App 仍只有一个 binding owner， 编辑和回执恢复复用现有投递路径，使用 `source_thread` 落点。解绑或原消息不可读后， 即使执行完成，也不得继续发送结果。 +本机群绑定仍启用且 Core 已完成、固定原目标的停止请求，可在 Core 重新核验绑定和 +App 身份时，并行读取两条原 provider 消息。两项只读观测全部成功后，仍须核对来源、 +关联关系和根消息提及，才能投递；仅该次校验复用身份观测，不持久化或跨次复用。 +绑定发生变化时丢弃预读消息；普通输入、本人私聊和未知停止范围保持原路径。 + 普通 Codex 话题空闲时，`/stop` 也覆盖同一原生 Session 中已完成 Turn 遗留的后台命令。 Core 先在原请求日志固定 Session 与候选 Turn，原 listener 随即继续受理消息；现有 回复工作池再恢复这条 prepared 请求，读取原生命令清单并取消原目标。恢复只能缩小 diff --git a/loopx/extensions/lark/private_conversations.py b/loopx/extensions/lark/private_conversations.py index c2c0cc7548..a5dead1846 100644 --- a/loopx/extensions/lark/private_conversations.py +++ b/loopx/extensions/lark/private_conversations.py @@ -121,6 +121,25 @@ def _completed_stop_sources(self, profile: str, message_id: str, messages = {item["message_id"]: item for item in items} return messages[message_id], messages[root_id] + def _completed_stop_root(self, record: Mapping[str, Any], binding: Mapping[str, Any]) -> str | None: + """Read scheduling only; fresh Core resolution still owns authorization.""" + event = record["event"] + root = str(event.get("root_id") or event["message_id"]) + path = self.core.root / f"{record['request_ref']}.json" + native = _read_json(path) if path.exists() else {} + if (binding.get("enabled") is True and binding.get("audience") == "group" + and binding.get("binding_id") == record["binding_id"] + and binding.get("transport_ref") == record["profile"] + and root != event["message_id"] and native.get("command") == "stop" + and native.get("status") == "command_completed" and native.get("target_recorded") is True + and native.get("binding_id") == record["binding_id"] + and native.get("source") == record["source"] == lark_private_source( + provider_ref=binding["provider_ref"], event=event) + and identity_ref(binding["provider_ref"], event["message_id"]) + == record["request_ref"] == native.get("request_ref")): + return root + return None + def _source_message(self, record: dict[str, Any], *, selected: dict[str, Any] | None = None, render_content: bool = True) -> Mapping[str, Any] | None: """Read the exact source under this App, then recheck the Core audience. @@ -132,8 +151,30 @@ def _source_message(self, record: dict[str, Any], *, selected: dict[str, Any] | result: Mapping[str, Any] message: Mapping[str, Any] | None try: + prefetched = None + read_binding: Mapping[str, Any] | None = None if selected is None: - selected = self.bindings.resolve(binding_id=record["binding_id"], **record["source"]) + # Only an existing, locally enabled binding and a completed, + # target-frozen Core stop can schedule these original two IDs. + # Buffer readonly IO while Core freshly verifies identity and + # authority; neither result is usable until both have joined. + if not render_content: + read_binding = next((row for row in self.bindings.read()["bindings"] + if row.get("binding_id") == record["binding_id"]), {}) + root = self._completed_stop_root(record, read_binding) if read_binding else None + if root: + with ThreadPoolExecutor(max_workers=1) as reads: + envelopes = reads.submit(self._completed_stop_sources, + record["profile"], event["message_id"], root) + selected = self.bindings.resolve(binding_id=record["binding_id"], + include_observation=True, **record["source"]) + prefetched = envelopes.result() + current = next((row for row in self.bindings.read()["bindings"] + if row.get("binding_id") == record["binding_id"]), None) + if current != read_binding or current != selected["binding"]: + return None # Revocation/replacement while reads ran is not a delivery grant. + else: + selected = self.bindings.resolve(binding_id=record["binding_id"], **record["source"]) native_path = self.core.root / f"{record['request_ref']}.json" native = _read_json(native_path) if native_path.exists() else {} if native.get("agent_target"): @@ -144,7 +185,9 @@ def rendered() -> Mapping[str, Any]: "--as", "bot", "--no-reactions", "--format", "json"])) group = selected["binding"].get("audience") == "group" - observation = None + # Resolve has just verified this App for this invocation. Reuse + # that observation for mention checks, never across verifications. + observation = selected.get("observation") stop_root: Mapping[str, Any] | None = None completed_stop = False if group: @@ -155,7 +198,6 @@ def rendered() -> Mapping[str, Any]: # Admission already verified this exact App. Reuse only # that call's ephemeral observation; Core still rechecks # under its fences, and outbound verification stays fresh. - observation = selected.get("observation") # An untrusted event hint only changes read scheduling. # Only authenticated provider text can select a control; # ordinary input still needs its display projection. @@ -188,25 +230,24 @@ def rendered() -> Mapping[str, Any]: # Only batch IDs already fixed by a target-frozen Core stop. # The current binding above still authorizes this read. # Event hints alone keep the source-first lookup. - completed_stop = (root_id != event["message_id"] - and native.get("command") == "stop" and native.get("status") == "command_completed" - and native.get("target_recorded") is True - and native.get("binding_id") == record["binding_id"] - and native.get("source") == record["source"] == lark_private_source( - provider_ref=selected["binding"]["provider_ref"], event=event) - and identity_ref(selected["binding"]["provider_ref"], event["message_id"]) - == record["request_ref"] == native.get("request_ref")) - # Delivery still needs both fresh observations. Their IO - # is independent; join it before checking ancestry and @. + completed_stop = self._completed_stop_root(record, selected["binding"]) == root_id + # Delivery reuses only its own binding observation. An + # explicitly supplied selection without one still needs a + # fresh identity; join that read before ancestry and @. def sources() -> tuple[Mapping[str, Any] | None, Mapping[str, Any] | None]: if completed_stop: + if prefetched is not None and read_binding == selected["binding"]: + return prefetched return self._completed_stop_sources(record["profile"], event["message_id"], root_id) return self._group_source_message(record["profile"], event["message_id"]), None - with ThreadPoolExecutor(max_workers=2) as reads: - envelopes = reads.submit(sources) - identity = reads.submit(self.bindings.observe, record["profile"]) - (canonical, stop_root), observation = envelopes.result(), identity.result() + if observation is not None: + canonical, stop_root = sources() + else: + with ThreadPoolExecutor(max_workers=2) as reads: + envelopes = reads.submit(sources) + identity = reads.submit(self.bindings.observe, record["profile"]) + (canonical, stop_root), observation = envelopes.result(), identity.result() result, message = {"returncode": 0 if canonical else 1}, canonical else: result = rendered() diff --git a/tests/test_lark_private_conversations.py b/tests/test_lark_private_conversations.py index 6f7b7025c9..7a6db17dc1 100644 --- a/tests/test_lark_private_conversations.py +++ b/tests/test_lark_private_conversations.py @@ -1171,14 +1171,14 @@ def completed_group_stop(ordinary): # noqa: F811 runtime.close() -def test_completed_stop_batches_only_its_core_bound_source_and_root(completed_group_stop): +def test_completed_stop_batches_its_core_bound_source_and_root_with_one_current_identity(completed_group_stop): _, provider, transport, root, event, record = completed_group_stop assert transport._source_verified(record) calls = [a for a in provider.calls if "GET" in a] assert len(calls) == 1 and calls[0][calls[0].index("GET") + 1].endswith("/mget") assert json.loads(calls[0][calls[0].index("--params") + 1]) == { "message_ids": [event["message_id"], root["message_id"]]} - assert sum("auth" in a for a in provider.calls) == 2 + assert sum("auth" in a for a in provider.calls) == 1 provider.messages[root["message_id"]]["mentions"] = [] assert not transport._source_verified(record) # Re-read, never cache the old root. assert not provider.writes and not provider.reaction_creates @@ -1353,8 +1353,32 @@ def runner(args, cwd=None, timeout=None): assert not provider.writes and not provider.reaction_creates +@pytest.mark.parametrize("change", ["unverified_app", "other_app", "root_mention", "source_parent"]) +def test_completed_stop_source_verification_reuses_one_current_observation_only_within_the_call( + completed_group_stop, change, +): + _, provider, transport, _, _, record = completed_group_stop + provider.calls.clear() + assert transport._source_verified(record) + # The binding check already verified this App. Its observation also checks + # the root mention; a second network identity probe adds no new authority. + assert sum("auth" in args for args in provider.calls) == 1 + if change == "unverified_app": + provider.bot_ready = False + elif change == "other_app": + provider.profile_apps[record["profile"]] = "cli_other_app" + elif change == "root_mention": + provider.messages[record["event"]["root_id"]]["mentions"] = [] + else: + provider.messages[record["event"]["message_id"]]["parent_id"] = "om_other_parent" + provider.calls.clear() + assert not transport._source_verified(record) + assert sum("auth" in args for args in provider.calls) == 1 + assert not provider.writes and not provider.reaction_creates + + @pytest.mark.parametrize("identity_fails", [False, True]) -def test_completed_stop_batch_joins_a_separate_fresh_identity(completed_group_stop, monkeypatch, identity_fails): +def test_completed_stop_preselected_without_observation_joins_a_fresh_identity(completed_group_stop, monkeypatch, identity_fails): _, provider, transport, _, _, record = completed_group_stop started, observed, release = threading.Event(), threading.Event(), threading.Event() original_sources, original_observe = transport._completed_stop_sources, transport.bindings.observe @@ -1376,18 +1400,107 @@ def observe(*args, **kwargs): monkeypatch.setattr(transport, "_completed_stop_sources", sources) monkeypatch.setattr(transport.bindings, "observe", observe) + selected = transport.bindings.resolve(binding_id=record["binding_id"], **record["source"]) with ThreadPoolExecutor(max_workers=1) as caller: - pending = caller.submit(transport._source_verified, record) + pending = caller.submit(transport._source_message, record, selected=selected, render_content=False) try: assert started.wait(2) and observed.wait(2) assert not pending.done() finally: release.set() - assert pending.result(timeout=3) is not identity_fails + assert bool(pending.result(timeout=3)) is not identity_fails assert len(observations) == 2 assert not provider.writes and not provider.reaction_creates +@pytest.mark.parametrize("failure", [None, "identity", "source", "revoked"]) +def test_completed_stop_default_overlaps_readonly_observations_but_joins_and_rechecks_authority( + completed_group_stop, monkeypatch, failure, +): + _, provider, transport, _, _, record = completed_group_stop + source_started, identity_started, release = threading.Event(), threading.Event(), threading.Event() + original_sources, original_observe = transport._completed_stop_sources, transport.bindings.observe + + def sources(*args): + source_started.set() + assert release.wait(3) + return (None, None) if failure == "source" else original_sources(*args) + + def observe(*args, **kwargs): + observation = original_observe(*args, **kwargs) + identity_started.set() + assert release.wait(3) + if failure == "identity": + raise ValueError("current App observation unavailable") + return observation + + monkeypatch.setattr(transport, "_completed_stop_sources", sources) + monkeypatch.setattr(transport.bindings, "observe", observe) + with ThreadPoolExecutor(max_workers=1) as caller: + pending = caller.submit(transport._source_verified, record) + try: + assert source_started.wait(2) and identity_started.wait(2) + assert not pending.done() and not provider.writes + if failure == "revoked": + state = transport.bindings.read() + transport.bindings.disconnect(binding_id=record["binding_id"], expected_revision=state["revision"]) + finally: + release.set() + assert pending.result(timeout=3) is (failure is None) + assert not provider.writes and not provider.reaction_creates + + +@pytest.mark.parametrize("revoke", [False, True]) +def test_completed_stop_reply_worker_waits_for_both_reads_and_does_not_send_after_revocation( + completed_group_stop, monkeypatch, revoke, +): + _, provider, transport, _, event, record = completed_group_stop + checking_source, source_started, identity_started, release = ( + threading.Event() for _ in range(4) + ) + original_verify = transport._source_verified + original_sources, original_observe = transport._completed_stop_sources, transport.bindings.observe + + def verify(value): + checking_source.set() + return original_verify(value) + + def sources(*args): + source_started.set() + assert release.wait(3) + return original_sources(*args) + + def observe(*args, **kwargs): + observation = original_observe(*args, **kwargs) + if checking_source.is_set(): + identity_started.set() + assert release.wait(3) + return observation + + monkeypatch.setattr(transport, "_source_verified", verify) + monkeypatch.setattr(transport, "_completed_stop_sources", sources) + monkeypatch.setattr(transport.bindings, "observe", observe) + path = transport.root / f"{record['request_ref']}.json" + with ThreadPoolExecutor(max_workers=1) as caller: + pending = caller.submit(transport.reconcile_request, path) + try: + assert source_started.wait(2) and identity_started.wait(2) + assert not pending.done() and not provider.writes + if revoke: + state = transport.bindings.read() + transport.bindings.disconnect(binding_id=record["binding_id"], expected_revision=state["revision"]) + finally: + release.set() + assert pending.result(timeout=3) == (0 if revoke else 1) + delivered = json.loads(path.read_text()) + if revoke: + assert not provider.writes and not delivered["deliveries"].get("terminal", {}).get("verified") + else: + assert delivered["deliveries"]["terminal"]["verified"] + assert len(provider.writes) == 1 + assert provider.messages[event["message_id"]]["root_id"] == event["root_id"] + + @pytest.mark.parametrize("followup", [False, True]) @pytest.mark.parametrize("identity_fails", [False, True]) def test_group_outbound_joins_current_source_and_identity_before_root_read(ordinary, monkeypatch, followup, identity_fails): # noqa: F811