From 938673b950203d6f62de1c58befa711010333b2a Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Wed, 11 Nov 2026 23:45:18 +0800 Subject: [PATCH 1/3] fix(chat): retain steward admission across executor publication Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> --- .../native_chat/external_conversations.py | 25 +++--- loopx/chat_ingress.py | 48 ++++++++++- tests/test_lark_private_conversations.py | 82 +++++++++++++++++++ 3 files changed, 140 insertions(+), 15 deletions(-) diff --git a/loopx/capabilities/native_chat/external_conversations.py b/loopx/capabilities/native_chat/external_conversations.py index c49ce6c592..8468d7d2e8 100644 --- a/loopx/capabilities/native_chat/external_conversations.py +++ b/loopx/capabilities/native_chat/external_conversations.py @@ -16,6 +16,7 @@ from typing import Any from ...chat_actions import goal_workspace_ref +from ...chat_ingress import ensure_external_admission_scope, resolve_external_admission_source from ...chat_store import _atomic_write_json, _read_json, RESUMABLE_SESSION_STATES from ...chat_attachments import normalize_chat_image_attachments from ...control_plane.quota.automation_cadence_readback import ( @@ -1397,17 +1398,9 @@ def admit(self, *, binding_id: str, source: dict[str, Any], request_ref: str, # it must not hold up other messages while native cancellation runs. with exclusive_file_lock(path, operation="admit_external_chat_request"): with exclusive_file_lock(self.root / "source-fences" / f"{binding_id}.{source['source_ref']}.json", operation="route_external_chat_request"): - # Resolve after waiting for both fences, including exact replay. - # A lock-external provider probe cannot authorize the write and - # would duplicate this fresh authority check on every request. - selected = self.bindings.resolve(binding_id=binding_id, **source, - **({"include_observation": True} if source_preflight is not None else {})) - # Earlier provider IO and App observations cannot authorize - # admission. Validate against this fresh selection inside both fences, - # before creating or replaying any canonical request. - if source_preflight is not None: - if source_preflight(deepcopy(selected)) is not None: - raise ValueError("source preflight must complete verification or raise") + # Resolve and verify the source inside both fences before + # creating or replaying any canonical request. + selected = resolve_external_admission_source(self.bindings, binding_id, source, source_preflight) expected = {"binding_id": binding_id, "source": source, "message": message, "command": command, "attachments": normalize_chat_image_attachments(attachments) or None, "origin": origin} recovering_request = path.exists() @@ -1427,7 +1420,8 @@ def admit(self, *, binding_id: str, source: dict[str, Any], request_ref: str, "created_at": datetime.now(timezone.utc).isoformat()} _atomic_write_json(path, row) try: - row = self._admit_prepared(path, row, selected, recovering_request=recovering_request) + row = self._admit_prepared(path, row, selected, recovering_request=recovering_request, + source_preflight=source_preflight) except ValueError: self._reject_prepared(path, row) raise @@ -1449,12 +1443,15 @@ def _reject_prepared(path: Path, row: dict[str, Any]) -> None: _atomic_write_json(path, row) def _admit_prepared(self, path: Path, row: dict[str, Any], selected: dict[str, Any], - *, recovering_request: bool = False) -> dict[str, Any]: + *, recovering_request: bool = False, + source_preflight: Callable[[dict[str, Any]], None] | None = None) -> dict[str, Any]: controller = self.controller held_for_binding = row.get("response_code") == STEWARD_BINDING_ADOPTION_PENDING if held_for_binding: row.pop("response_code") - self.bindings.ensure_delivery_scope(selected) + selected = ensure_external_admission_scope( + self.bindings, selected, row["source"], source_preflight=source_preflight, + refresh_steward_executor=row["command"] is None and row.get("agent_target") is None) if not row.get("routing_recorded"): choices = [item for item in self.pending() if item.get("binding_id") == row["binding_id"] and item["source"]["source_ref"] == row["source"]["source_ref"] diff --git a/loopx/chat_ingress.py b/loopx/chat_ingress.py index 4c97dca82f..3fe859d953 100644 --- a/loopx/chat_ingress.py +++ b/loopx/chat_ingress.py @@ -2,22 +2,68 @@ from __future__ import annotations from collections.abc import Callable, Iterator +from copy import deepcopy from datetime import datetime, timedelta, timezone from functools import cached_property import os import json from pathlib import Path import uuid -from typing import Any +from typing import TYPE_CHECKING, Any from .chat import require_matching_message_replay, require_matching_replay from .file_lock import exclusive_file_lock from .capabilities.native_chat.file_discovery import FileChatDiscovery +if TYPE_CHECKING: + from .capabilities.native_chat.conversation_bindings import ChatConversationBindings + SESSION_QUEUE_MAX_PENDING = 20 SESSION_QUEUE_TTL_SECONDS: int | None = None +def resolve_external_admission_source( + bindings: ChatConversationBindings, binding_id: str, source: dict[str, Any], + source_preflight: Callable[[dict[str, Any]], None] | None, +) -> dict[str, Any]: + selected = bindings.resolve(binding_id=binding_id, **source, include_observation=source_preflight is not None) + if source_preflight is not None and source_preflight(deepcopy(selected)) is not None: + raise ValueError("source preflight must complete verification or raise") + return selected + + +def ensure_external_admission_scope( + bindings: ChatConversationBindings, selected: dict[str, Any], source: dict[str, Any], *, + refresh_steward_executor: bool, source_preflight: Callable[[dict[str, Any]], None] | None, +) -> dict[str, Any]: + previous = selected["binding"] + if (not refresh_steward_executor or source.get("private_human_message") is not True + or previous["context_kind"] != "steward" or previous.get("audience", "owner") != "owner" + or previous.get("goal_scope") not in {"all_registered", "selected"} or bindings.controller is None): + bindings.ensure_delivery_scope(selected) + return selected + + with exclusive_file_lock(bindings.path, operation="configure_bound_delivery_scope"): + current = next((binding for binding in bindings.read()["bindings"] + if binding["binding_id"] == previous["binding_id"]), None) + if current == previous: + bindings._ensure_delivery_scope(selected) # This caller retains the binding fence. + return selected + if (current is None + or {previous.get("executor_endpoint_id"), current.get("executor_endpoint_id")} != {"antigravity", "codex"} + or current != {**previous, "executor_endpoint_id": current["executor_endpoint_id"]}): + raise ValueError("conversation binding is no longer authorized") + + # No delivery effect has run. Reverify the same original source outside + # the binding fence, then require the exact refreshed binding at the fence. + fresh = resolve_external_admission_source(bindings, previous["binding_id"], source, source_preflight) + if (fresh["binding"] != current or fresh["context"] != selected["context"] + or fresh["channel_id"] != selected["channel_id"]): + raise ValueError("conversation binding changed during identity verification") + bindings.ensure_delivery_scope(fresh) + return fresh + + def chat_runtime_activation_lock(runtime_root: Path, *, timeout_seconds: float | None = None): """Fence new durable input against a managed-service runtime switch.""" return exclusive_file_lock( diff --git a/tests/test_lark_private_conversations.py b/tests/test_lark_private_conversations.py index 9814201a59..6d1cb47a01 100644 --- a/tests/test_lark_private_conversations.py +++ b/tests/test_lark_private_conversations.py @@ -104,6 +104,88 @@ def interrupt_admission_projection(*args, **kwargs): fixture.close() +@pytest.mark.parametrize("change", ["executor", "scope", "revoked", "second_publication"]) +def test_private_steward_admission_rechecks_same_owner_executor_publication(tmp_path, monkeypatch, change): + from test_native_steward_private import finish, steward + from loopx import chat_ingress + from loopx.capabilities.native_chat import conversation_bindings + from loopx.file_lock import exclusive_file_lock + + fixture = steward.__wrapped__(tmp_path) + store, runtime, provider, transport, binding, capture, _ = next(fixture) + try: + original_event = provider.event("steward-app", "original-before-adoption", "Read the original conversation") + assert transport.admit("steward-app", original_event)["status"] == "durably_accepted" + original_ref = identity_ref(binding["provider_ref"], original_event["message_id"]) + original = transport.core.read_request(original_ref) + assert finish(runtime, original)["status"] == "completed" + original_path = transport.root / f"{original_ref}.json" + assert transport.reconcile_request(original_path) == 1 + original_delivery = original_path.read_bytes() + sid = original["session_id"] + + bindings = transport.bindings + bindings.select_app_steward(binding["binding_id"], expected_revision=bindings.read()["revision"]) + adopted = bindings.read() + stale = {**adopted, "bindings": [ + {**row, "executor_endpoint_id": "antigravity"} if row["binding_id"] == binding["binding_id"] else row + for row in adopted["bindings"]]} + bindings.path.write_text(json.dumps(stale)) + published = {**adopted, "revision": adopted["revision"] + 1} + if change == "scope": + published["bindings"] = [ + {**row, "goal_scope": "selected"} if row["binding_id"] == binding["binding_id"] else row + for row in adopted["bindings"]] + if change == "revoked": + published = {**published, "app_steward_binding_id": None, "bindings": [ + row for row in adopted["bindings"] if row["binding_id"] != binding["binding_id"]]} + observed_endpoints = [] + + def publish_at_scope_fence(path, **kwargs): + if path == bindings.path and kwargs.get("operation") == "configure_bound_delivery_scope": + previous = next(row for row in bindings.read()["bindings"] if row["binding_id"] == binding["binding_id"]) + observed_endpoints.append(previous["executor_endpoint_id"]) + if len(observed_endpoints) == 1: + bindings.path.write_text(json.dumps(published)) + elif change == "second_publication": + bindings.path.write_text(json.dumps({**stale, "revision": adopted["revision"] + 2})) + return exclusive_file_lock(path, **kwargs) + + monkeypatch.setattr(chat_ingress, "exclusive_file_lock", publish_at_scope_fence) + monkeypatch.setattr(conversation_bindings, "exclusive_file_lock", publish_at_scope_fence) + pending_event = provider.event("steward-app", "original-during-adoption", "Continue in this conversation") + pending_ref = identity_ref(binding["provider_ref"], pending_event["message_id"]) + pending_path = transport.root / f"{pending_ref}.json" + calls_before = capture.read_bytes() + writes_before = list(provider.writes) + admission = transport.admit("steward-app", pending_event) + request = transport.core.read_request(pending_ref) + if change != "executor": + assert admission["status"] == "command_rejected" + assert request["status"] == "rejected" and request["turn_id"] is None + assert not store.turn_for_client(sid, f"external-{pending_ref}") + assert capture.read_bytes() == calls_before and provider.writes == writes_before + assert observed_endpoints == (["antigravity", "codex"] if change == "second_publication" else ["antigravity"]) + else: + assert admission["status"] == "durably_accepted" + assert observed_endpoints == ["antigravity", "codex"] + assert request["status"] == "accepted" and request["session_id"] == sid + assert request["binding_id"] == binding["binding_id"] and request["source"] == original["source"] + assert finish(runtime, request)["status"] == "completed" + assert transport.reconcile_request(pending_path) == 1 + delivered = json.loads(pending_path.read_text()) + assert delivered["status"] == "delivered" and delivered["deliveries"]["terminal"]["verified"] + assert delivered["event"]["message_id"] == pending_event["message_id"] + assert store.turn_for_client(sid, f"external-{pending_ref}")["turn_id"] == request["turn_id"] + writes_after = list(provider.writes) + assert transport.reconcile_request(pending_path) == 0 and provider.writes == writes_after + assert sum(json.loads(line)["method"] == "turn/start" for line in capture.read_text().splitlines()) == 2 + assert [row["session_id"] for row in store.list_sessions()] == [sid] + assert original_path.read_bytes() == original_delivery + finally: + fixture.close() + + def test_historical_context_is_explicitly_bounded_and_never_decodes_unread_attachments(): rows = [{"message_id": f"m-{i}", "created_at": f"2026-01-01T00:00:{i:02}+00:00", "role": "agent", "text": f"Complete original answer {i}"} for i in range(14)] From fb6ea1d603915eaf1382a58f1c1bfac3fb0dce17 Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Wed, 11 Nov 2026 23:58:21 +0800 Subject: [PATCH 2/3] fix(chat): pin the original steward Session during scope refresh Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> --- .../native_chat/external_conversations.py | 9 ++--- loopx/chat_ingress.py | 37 ++++++++++++++++--- tests/test_lark_private_conversations.py | 8 +++- 3 files changed, 41 insertions(+), 13 deletions(-) diff --git a/loopx/capabilities/native_chat/external_conversations.py b/loopx/capabilities/native_chat/external_conversations.py index 8468d7d2e8..1903219e23 100644 --- a/loopx/capabilities/native_chat/external_conversations.py +++ b/loopx/capabilities/native_chat/external_conversations.py @@ -16,7 +16,7 @@ from typing import Any from ...chat_actions import goal_workspace_ref -from ...chat_ingress import ensure_external_admission_scope, resolve_external_admission_source +from ...chat_ingress import STEWARD_BINDING_ADOPTION_PENDING, ensure_external_admission_scope, resolve_external_admission_source from ...chat_store import _atomic_write_json, _read_json, RESUMABLE_SESSION_STATES from ...chat_attachments import normalize_chat_image_attachments from ...control_plane.quota.automation_cadence_readback import ( @@ -33,7 +33,6 @@ _REQUEST_JOURNAL_STEM = re.compile(r"[a-f0-9]{24}") QUOTA_BACKLOG_BATCH_SIZE = 64 _QUOTA_BACKLOG_BATCH_BYTES = 1024 * 1024 -STEWARD_BINDING_ADOPTION_PENDING = "steward_executor_binding_pending" def _preparation_image_identity(value: Any) -> dict[str, Any] | None: @@ -1439,6 +1438,8 @@ def admit(self, *, binding_id: str, source: dict[str, Any], request_ref: str, @staticmethod def _reject_prepared(path: Path, row: dict[str, Any]) -> None: # Definitive command/grant denials must not retry indefinitely. + if row.get("response_code") == STEWARD_BINDING_ADOPTION_PENDING: + row.pop("response_code") row.update(status="rejected", response="原授权或命令格式不可用，本条未进入执行队列；请用 /agents 或 /project 查看可用入口。") _atomic_write_json(path, row) @@ -1446,12 +1447,10 @@ def _admit_prepared(self, path: Path, row: dict[str, Any], selected: dict[str, A *, recovering_request: bool = False, source_preflight: Callable[[dict[str, Any]], None] | None = None) -> dict[str, Any]: controller = self.controller + selected = ensure_external_admission_scope(self.bindings, selected, path, row, source_preflight=source_preflight) held_for_binding = row.get("response_code") == STEWARD_BINDING_ADOPTION_PENDING if held_for_binding: row.pop("response_code") - selected = ensure_external_admission_scope( - self.bindings, selected, row["source"], source_preflight=source_preflight, - refresh_steward_executor=row["command"] is None and row.get("agent_target") is None) if not row.get("routing_recorded"): choices = [item for item in self.pending() if item.get("binding_id") == row["binding_id"] and item["source"]["source_ref"] == row["source"]["source_ref"] diff --git a/loopx/chat_ingress.py b/loopx/chat_ingress.py index 3fe859d953..7052737e04 100644 --- a/loopx/chat_ingress.py +++ b/loopx/chat_ingress.py @@ -20,24 +20,33 @@ SESSION_QUEUE_MAX_PENDING = 20 SESSION_QUEUE_TTL_SECONDS: int | None = None +STEWARD_BINDING_ADOPTION_PENDING = "steward_executor_binding_pending" def resolve_external_admission_source( bindings: ChatConversationBindings, binding_id: str, source: dict[str, Any], - source_preflight: Callable[[dict[str, Any]], None] | None, + source_preflight: Callable[[dict[str, Any]], None] | None, *, + session: dict[str, Any] | None = None, ) -> dict[str, Any]: - selected = bindings.resolve(binding_id=binding_id, **source, include_observation=source_preflight is not None) + observation_args = {**source} + if source_preflight is not None: + observation_args["include_observation"] = True + if session is not None: + observation_args.update(session=session, session_context=session["steward_context"]) + selected = bindings.resolve(binding_id=binding_id, **observation_args) if source_preflight is not None and source_preflight(deepcopy(selected)) is not None: raise ValueError("source preflight must complete verification or raise") return selected def ensure_external_admission_scope( - bindings: ChatConversationBindings, selected: dict[str, Any], source: dict[str, Any], *, - refresh_steward_executor: bool, source_preflight: Callable[[dict[str, Any]], None] | None, + bindings: ChatConversationBindings, selected: dict[str, Any], path: Path, request: dict[str, Any], *, + source_preflight: Callable[[dict[str, Any]], None] | None, ) -> dict[str, Any]: + source = request["source"] previous = selected["binding"] - if (not refresh_steward_executor or source.get("private_human_message") is not True + if (request["command"] is not None or request.get("agent_target") is not None + or source.get("private_human_message") is not True or previous["context_kind"] != "steward" or previous.get("audience", "owner") != "owner" or previous.get("goal_scope") not in {"all_registered", "selected"} or bindings.controller is None): bindings.ensure_delivery_scope(selected) @@ -53,10 +62,26 @@ def ensure_external_admission_scope( or {previous.get("executor_endpoint_id"), current.get("executor_endpoint_id")} != {"antigravity", "codex"} or current != {**previous, "executor_endpoint_id": current["executor_endpoint_id"]}): raise ValueError("conversation binding is no longer authorized") + endpoint = current["executor_endpoint_id"] + candidates = bindings.controller.store.resumable_session_candidates( + goal_id="loopx-manager", agent_id=endpoint, channel_id=selected["channel_id"]) + if len(candidates) != 1: + raise ValueError("the original steward Session is unavailable for admission") + original = candidates[0] + session_id = original["session_id"] + if (original.get("executor_endpoint_id") != endpoint + or not isinstance(original.get("steward_context"), dict) + or request.get("session_id") not in (None, session_id)): + raise ValueError("the original steward Session changed during admission") + # Preserve the original Session across provider IO, service interruption + # or Close. The existing prepared-request path owns subsequent admission. + from .chat_store import _atomic_write_json + request.update(session_id=session_id, response_code=STEWARD_BINDING_ADOPTION_PENDING) + _atomic_write_json(path, request) # No delivery effect has run. Reverify the same original source outside # the binding fence, then require the exact refreshed binding at the fence. - fresh = resolve_external_admission_source(bindings, previous["binding_id"], source, source_preflight) + fresh = resolve_external_admission_source(bindings, previous["binding_id"], source, source_preflight, session=original) if (fresh["binding"] != current or fresh["context"] != selected["context"] or fresh["channel_id"] != selected["channel_id"]): raise ValueError("conversation binding changed during identity verification") diff --git a/tests/test_lark_private_conversations.py b/tests/test_lark_private_conversations.py index 6d1cb47a01..8cd5965410 100644 --- a/tests/test_lark_private_conversations.py +++ b/tests/test_lark_private_conversations.py @@ -104,7 +104,7 @@ def interrupt_admission_projection(*args, **kwargs): fixture.close() -@pytest.mark.parametrize("change", ["executor", "scope", "revoked", "second_publication"]) +@pytest.mark.parametrize("change", ["executor", "scope", "revoked", "second_publication", "closed"]) def test_private_steward_admission_rechecks_same_owner_executor_publication(tmp_path, monkeypatch, change): from test_native_steward_private import finish, steward from loopx import chat_ingress @@ -149,6 +149,8 @@ def publish_at_scope_fence(path, **kwargs): bindings.path.write_text(json.dumps(published)) elif change == "second_publication": bindings.path.write_text(json.dumps({**stale, "revision": adopted["revision"] + 2})) + elif change == "closed": + assert runtime.close_session(sid) return exclusive_file_lock(path, **kwargs) monkeypatch.setattr(chat_ingress, "exclusive_file_lock", publish_at_scope_fence) @@ -165,7 +167,9 @@ def publish_at_scope_fence(path, **kwargs): assert request["status"] == "rejected" and request["turn_id"] is None assert not store.turn_for_client(sid, f"external-{pending_ref}") assert capture.read_bytes() == calls_before and provider.writes == writes_before - assert observed_endpoints == (["antigravity", "codex"] if change == "second_publication" else ["antigravity"]) + assert observed_endpoints == (["antigravity", "codex"] if change in {"second_publication", "closed"} else ["antigravity"]) + if change == "closed": + assert request["session_id"] == sid and "response_code" not in request else: assert admission["status"] == "durably_accepted" assert observed_endpoints == ["antigravity", "codex"] From 43bd8bd8c009cfe5afe2caa00c71534390b158ab Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Thu, 12 Nov 2026 00:40:36 +0800 Subject: [PATCH 3/3] fix(lark): verify original message after steward executor adoption Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> --- .../extensions/lark/private_conversations.py | 23 ++++----- loopx/extensions/lark/private_ingress.py | 47 +++++++++++++++++++ tests/test_lark_private_conversations.py | 38 ++++++++++++--- 3 files changed, 88 insertions(+), 20 deletions(-) diff --git a/loopx/extensions/lark/private_conversations.py b/loopx/extensions/lark/private_conversations.py index 952dfccb13..a732be14e6 100644 --- a/loopx/extensions/lark/private_conversations.py +++ b/loopx/extensions/lark/private_conversations.py @@ -42,6 +42,7 @@ from .outbound import present_lark_conversation_text as _presentation_text from .private_images import private_message_caption, private_message_images from .private_delivery_discovery import PrivateDeliveryDiscovery, prepared_steward_binding_waits +from .private_ingress import PrivateSourceUnavailable, private_source_text, steward_source_preflight from .private_ingress import project_private_reply_history as project_private_reply_history from .return_files import uploaded_chat_images, verify_chat_images from .private_progress import (ACTIVITY_UPDATE_INTERVAL_SEC, UPDATE_INTERVAL_SEC, legacy_recovery_repairs, @@ -914,19 +915,9 @@ def admit(self, profile: str, event: dict[str, Any], *, canonical_type = source_message.get("msg_type", source_message.get("message_type")) if canonical_type != "text": return {"status": "source_conflict"} - content = source_message.get("content") - if isinstance(content, str): - text = content - else: - # Raw provider envelopes are supported by the existing - # readback contract too; keep parsing isolated to body. - try: - raw = json.loads(source_message["body"]["content"]) - text = raw["text"] - except (ValueError, KeyError, TypeError): - return {"status": "invalid_text"} - if not isinstance(text, str): - return {"status": "invalid_text"} + text = private_source_text(source_message) + if text is None: + return {"status": "invalid_text"} if not text.strip(): return {"status": "empty_text"} elif message_type in {"image", "post"}: @@ -1024,10 +1015,14 @@ def read(tail: list[str]) -> Mapping[str, Any] | None: _atomic_write_json(path, record) return {"status": "history_answer_held"} else: + preflight = (steward_source_preflight(selected["binding"], source_message, + lambda fresh: self._source_message(record, selected=fresh)) if command is None else None) admitted = self.core.admit(binding_id=binding["binding_id"], source=source, request_ref=request, message=text, command=command, attachments=attachments, - defer_background_stop=defer_background_stop) + defer_background_stop=defer_background_stop, source_preflight=preflight) record_event_timing(profile, event["message_id"], "core_admission_returned") + except PrivateSourceUnavailable: + return {"status": "source_verification_failed"} except ValueError: if record.get("capture_origin") == "provider_history": return {"status": "history_answer_held"} diff --git a/loopx/extensions/lark/private_ingress.py b/loopx/extensions/lark/private_ingress.py index e3c837c93c..3450e34bd9 100644 --- a/loopx/extensions/lark/private_ingress.py +++ b/loopx/extensions/lark/private_ingress.py @@ -25,6 +25,53 @@ from .private_json import write_private_json_atomic +class PrivateSourceUnavailable(OSError): + """Keep the captured request for its existing delivery worker.""" + + +def private_source_text(message: Mapping[str, Any]) -> str | None: + content = message.get("content") + if isinstance(content, str): + return content + try: + text = json.loads(message["body"]["content"])["text"] + except (KeyError, TypeError, ValueError): + return None + return text if isinstance(text, str) else None + + +def _private_source_input(message: Mapping[str, Any]) -> tuple[Any, ...]: + kind = message.get("msg_type", message.get("message_type")) + content = private_source_text(message) if kind == "text" else message.get("content") + ancestry = tuple(message.get(field) or None for field in ("parent_id", "root_id", "thread_id")) + return kind, content, message.get("create_time"), ancestry + + +def steward_source_preflight( + binding: Mapping[str, Any], original: Mapping[str, Any], + read_source: Callable[[dict[str, Any]], Mapping[str, Any] | None], +) -> Callable[[dict[str, Any]], None] | None: + """Require fresh Lark content if Core observes a different executor.""" + if binding.get("context_kind") != "steward" or binding.get("audience", "owner") != "owner": + return None + observed_executor = binding["executor_endpoint_id"] + + def verify(fresh: dict[str, Any]) -> None: + nonlocal observed_executor + executor = fresh["binding"]["executor_endpoint_id"] + if executor == observed_executor: + return + current = read_source(fresh) + if current is None: + raise PrivateSourceUnavailable("original Lark source could not be verified") + if (current.get("deleted") is True or original.get("deleted") is True + or _private_source_input(current) != _private_source_input(original)): + raise ValueError("original Lark source changed during executor adoption") + observed_executor = executor + + return verify + + class IngressObservationStatus(str, Enum): """Lark observation vocabulary; it grants no Core action or permission.""" diff --git a/tests/test_lark_private_conversations.py b/tests/test_lark_private_conversations.py index 8cd5965410..6771b93483 100644 --- a/tests/test_lark_private_conversations.py +++ b/tests/test_lark_private_conversations.py @@ -104,7 +104,8 @@ def interrupt_admission_projection(*args, **kwargs): fixture.close() -@pytest.mark.parametrize("change", ["executor", "scope", "revoked", "second_publication", "closed"]) +@pytest.mark.parametrize("change", ["executor", "scope", "revoked", "second_publication", "closed", + "source_unavailable", "source_changed"]) def test_private_steward_admission_rechecks_same_owner_executor_publication(tmp_path, monkeypatch, change): from test_native_steward_private import finish, steward from loopx import chat_ingress @@ -114,8 +115,12 @@ def test_private_steward_admission_rechecks_same_owner_executor_publication(tmp_ fixture = steward.__wrapped__(tmp_path) store, runtime, provider, transport, binding, capture, _ = next(fixture) try: + def source_reads(message_id): + return sum("+messages-mget" in args and message_id in args for args in provider.calls) + original_event = provider.event("steward-app", "original-before-adoption", "Read the original conversation") assert transport.admit("steward-app", original_event)["status"] == "durably_accepted" + assert source_reads(original_event["message_id"]) == 1 original_ref = identity_ref(binding["provider_ref"], original_event["message_id"]) original = transport.core.read_request(original_ref) assert finish(runtime, original)["status"] == "completed" @@ -140,6 +145,9 @@ def test_private_steward_admission_rechecks_same_owner_executor_publication(tmp_ published = {**published, "app_steward_binding_id": None, "bindings": [ row for row in adopted["bindings"] if row["binding_id"] != binding["binding_id"]]} observed_endpoints = [] + pending_event = provider.event("steward-app", "original-during-adoption", "Continue in this conversation") + pending_id = pending_event["message_id"] + pending_source = dict(provider.messages[pending_id]) def publish_at_scope_fence(path, **kwargs): if path == bindings.path and kwargs.get("operation") == "configure_bound_delivery_scope": @@ -147,6 +155,10 @@ def publish_at_scope_fence(path, **kwargs): observed_endpoints.append(previous["executor_endpoint_id"]) if len(observed_endpoints) == 1: bindings.path.write_text(json.dumps(published)) + if change == "source_unavailable": + provider.messages.pop(pending_id) + elif change == "source_changed": + provider.messages[pending_id]["content"] = "Edited after the first provider read" elif change == "second_publication": bindings.path.write_text(json.dumps({**stale, "revision": adopted["revision"] + 2})) elif change == "closed": @@ -155,28 +167,42 @@ def publish_at_scope_fence(path, **kwargs): monkeypatch.setattr(chat_ingress, "exclusive_file_lock", publish_at_scope_fence) monkeypatch.setattr(conversation_bindings, "exclusive_file_lock", publish_at_scope_fence) - pending_event = provider.event("steward-app", "original-during-adoption", "Continue in this conversation") - pending_ref = identity_ref(binding["provider_ref"], pending_event["message_id"]) + pending_ref = identity_ref(binding["provider_ref"], pending_id) pending_path = transport.root / f"{pending_ref}.json" calls_before = capture.read_bytes() writes_before = list(provider.writes) admission = transport.admit("steward-app", pending_event) request = transport.core.read_request(pending_ref) - if change != "executor": + if change == "source_unavailable": + assert admission["status"] == "source_verification_failed" + assert request["status"] == "prepared" and request["session_id"] == sid and request["turn_id"] is None + assert json.loads(pending_path.read_text())["status"] == "captured" + assert source_reads(pending_id) == 2 and observed_endpoints == ["antigravity"] + assert not store.turn_for_client(sid, f"external-{pending_ref}") + assert capture.read_bytes() == calls_before and provider.writes == writes_before + provider.messages[pending_id] = pending_source + transport.reconcile_request(pending_path) + request = transport.core.read_request(pending_ref) + if change not in {"executor", "source_unavailable"}: assert admission["status"] == "command_rejected" assert request["status"] == "rejected" and request["turn_id"] is None assert not store.turn_for_client(sid, f"external-{pending_ref}") assert capture.read_bytes() == calls_before and provider.writes == writes_before + if change == "source_changed": + assert source_reads(pending_id) == 2 assert observed_endpoints == (["antigravity", "codex"] if change in {"second_publication", "closed"} else ["antigravity"]) if change == "closed": assert request["session_id"] == sid and "response_code" not in request else: - assert admission["status"] == "durably_accepted" + if change == "executor": + assert admission["status"] == "durably_accepted" + assert source_reads(pending_id) == 2 assert observed_endpoints == ["antigravity", "codex"] assert request["status"] == "accepted" and request["session_id"] == sid assert request["binding_id"] == binding["binding_id"] and request["source"] == original["source"] assert finish(runtime, request)["status"] == "completed" - assert transport.reconcile_request(pending_path) == 1 + if json.loads(pending_path.read_text())["status"] != "delivered": + assert transport.reconcile_request(pending_path) == 1 delivered = json.loads(pending_path.read_text()) assert delivered["status"] == "delivered" and delivered["deliveries"]["terminal"]["verified"] assert delivered["event"]["message_id"] == pending_event["message_id"]