fix(chat): retain steward admission across executor publication

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 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 (
@@ -32,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:
@@ -1397,17 +1397,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 +1419,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
@@ -1445,16 +1438,19 @@ 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)
 
     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
+        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")
-        self.bindings.ensure_delivery_scope(selected)
         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"]

loopx/chat_ingress.py

@@ -2,20 +2,91 @@
 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
+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, *,
+    session: dict[str, Any] | None = None,
+) -> dict[str, Any]:
+    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], path: Path, request: dict[str, Any], *,
+    source_preflight: Callable[[dict[str, Any]], None] | None,
+) -> dict[str, Any]:
+    source = request["source"]
+    previous = selected["binding"]
+    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)
+        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")
+        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, 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")
+    bindings.ensure_delivery_scope(fresh)
+    return fresh
 
 
 def chat_runtime_activation_lock(runtime_root: Path, *, timeout_seconds: float | None = None):

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"}

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."""