Hold private steward input during executor adoption

docs/architecture/rfcs/app-conversation-and-async-inbox-v0.md

@@ -162,6 +162,14 @@ remain unavailable. There is no fallback to ordinary steward/project history,
 Session creation, executor change or model reconfiguration on this entry.
 Group audiences retain their existing boundary.
 
+When an owner-private steward Session has adopted an executor and its
+binding still names the previous executor, a newly captured ordinary message
+stays prepared against that exact Session. The existing reply worker waits
+for the binding and Session to agree, then revalidates the original source
+before queueing the request. Closing the Session before admission ends that
+request without creating a replacement. Previously rejected requests retain
+their recorded outcome.
+
 For a private Lark reply, the adapter keeps the thread ID from the verified
 provider context when the incoming event omits it. This lets the App show the
 original body, exact quote and verified thread excerpt together. Redelivery

docs/architecture/rfcs/app-conversation-and-async-inbox-v0.zh-CN.md

@@ -115,6 +115,11 @@ Lark listener 观测。该本地 hook 只读取缓存，不发起网络请求。
 不回退到普通管家/项目历史，不新建替代 Session，也不在这里切换执行器或模型。
 群受众保留原有边界。
 
+本人私聊管家的 Session 已采用新执行器、binding 仍指向原执行器时，新捕获的
+普通消息保持待准入，关联确切的原 Session。原回复 worker 等待两者对齐，
+再次核验原消息后才入队。准入前关闭原 Session 会结束该请求，不创建替代
+会话。此前已拒绝的请求保留原结果。
+
 飞书私聊回复的事件省略话题 ID 时，适配器从已核验的原消息上下文保留该 ID。
 App 因此可以一起呈现原正文、精确引用和已核验的话题节选。重复到达时省略
 该 ID 仍可沿用原请求，明确给出不同 ID 时继续拒绝。

loopx/capabilities/native_chat/external_conversations.py

@@ -32,6 +32,7 @@
 _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:
@@ -1450,6 +1451,9 @@ def _reject_prepared(path: Path, row: dict[str, Any]) -> None:
     def _admit_prepared(self, path: Path, row: dict[str, Any], selected: dict[str, Any],
                         *, recovering_request: bool = False) -> 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)
         if not row.get("routing_recorded"):
             choices = [item for item in self.pending() if item.get("binding_id") == row["binding_id"]
@@ -1461,6 +1465,7 @@ def _admit_prepared(self, path: Path, row: dict[str, Any], selected: dict[str, A
         if row["command"] in {"agents", "select_agent", "select_project"}:
             row["agent_target"] = None
         target = row.get("agent_target")
+        steward_candidate = None
         if target:
             authority = self.bindings.resolve_agent_target(selected, target)
             row["agent_audience"] = authority["audience"]
@@ -1482,10 +1487,11 @@ def _admit_prepared(self, path: Path, row: dict[str, Any], selected: dict[str, A
                               if item.get("status") in RESUMABLE_SESSION_STATES]
                 if candidates:
                     latest = max(candidates, key=lambda item: str(item.get("created_at") or ""))
-                    candidate = controller.store.load_session(latest["session_id"])
-                    if candidate and candidate.get("manager_executor_allocation", {}).get("selection_policy") == "flexible":
-                        self.bindings.session_context(candidate["steward_context"], session=candidate)
-                        current = candidate
+                    steward_candidate = controller.store.load_session(latest["session_id"])
+                    allocation = (steward_candidate or {}).get("manager_executor_allocation")
+                    if isinstance(allocation, dict) and allocation.get("selection_policy") == "flexible":
+                        self.bindings.session_context(steward_candidate["steward_context"], session=steward_candidate)
+                        current = steward_candidate
         if row["command"] in {"status", "help"} and not target:
             # Observation must retain failed/closed originals. Admission still
             # uses the resumable selector and never resumes from this snapshot.
@@ -1510,8 +1516,14 @@ def _admit_prepared(self, path: Path, row: dict[str, Any], selected: dict[str, A
                 row.update(status="accepted", turn_id=turn["turn_id"])
                 _atomic_write_json(path, row)
                 return row
+            if held_for_binding and current.get("status") == "closed":
+                raise ValueError("the original steward Session is unavailable for admission")
+            steward_candidate = current
         from ...control_plane.effect_runtime import effect_runtime_result
         observations = {}
+        if (row["command"] is None and steward_candidate and selected["binding"]["context_kind"] == "steward"
+                and steward_candidate.get("executor_endpoint_id") != selected["binding"]["executor_endpoint_id"]):
+            observations.update(steward_adoption_session=steward_candidate, context=selected["context"])
         if (row["command"] == "stop" and current and not current.get("active_turn_id")
                 and not row.get("target_recorded") and not target):
             # Freeze canonical candidates before any fallible native inventory
@@ -1536,6 +1548,11 @@ def _admit_prepared(self, path: Path, row: dict[str, Any], selected: dict[str, A
             "current_session": current, "binding": selected["binding"],
             "agent_target": target, **observations})
         operation = plan["operation"]
+        if operation == "hold_steward_binding":
+            row.update(session_id=plan["session_id"], response_code=STEWARD_BINDING_ADOPTION_PENDING)
+            if not held_for_binding:
+                _atomic_write_json(path, row)
+            return row
         if operation == "select_recipient":
             if selected["binding"]["context_kind"] != "project":
                 raise ValueError("Agent selection requires the project assistant")

loopx/control_plane/collaboration/conversation_binding.ts

@@ -323,6 +323,11 @@ function boundProject(params: JsonObject, row: JsonObject): Record<string, strin
   return projects[0];
 }
 
+function matchesStewardContext(saved: JsonObject, context: JsonObject, row: JsonObject): boolean {
+  return isDeepStrictEqual({...saved, goal_ids: []}, {...context, goal_ids: []})
+    && (row.goal_scope === "all_registered"
+      || normalizeStewardGoalScope(saved.goal_ids).every(goal => (row.goal_ids as string[]).includes(goal)));
+}
 
 export function resolveBoundConversation(params: JsonObject): JsonObject {
   const current = state(params.current);
@@ -366,9 +371,7 @@ export function resolveBoundConversation(params: JsonObject): JsonObject {
     // JSON object member order is not identity. Preserve every value and field
     // check while accepting equivalent contexts from another JSON writer.
     const matches = row.context_kind === "steward"
-      ? isDeepStrictEqual({...saved, goal_ids: []}, {...context, goal_ids: []})
-        && (row.goal_scope === "all_registered" || normalizeStewardGoalScope(saved.goal_ids).every(g => (row.goal_ids as string[]).includes(g)))
-      : isDeepStrictEqual(saved, context);
+      ? matchesStewardContext(saved, context, row) : isDeepStrictEqual(saved, context);
     if (!matches) throw new EffectRuntimeRequestError("bound Session context changed");
   }
   const channel = row.context_kind === "steward"
@@ -658,7 +661,38 @@ export function planBoundConversationRequest(params: JsonObject): JsonObject {
       ...(background.length ? {background_turn_ids: background} : {}),
       response_code: command === "new" ? "new_session" : turn || background.length ? "stop_requested" : "no_active_turn"};
   }
-  return {operation: "admit_turn", client_turn_id: `external-${request}`, session_id: session, turn_id: null};
+  const adoptionHold = planStewardBindingAdoptionHold(params, row);
+  return adoptionHold ?? {operation: "admit_turn", client_turn_id: `external-${request}`, session_id: session, turn_id: null};
+}
+
+function planStewardBindingAdoptionHold(params: JsonObject, request: JsonObject): JsonObject | null {
+  if (params.steward_adoption_session == null || request.status !== "prepared"
+      || request.command !== null || params.agent_target != null) return null;
+  const original = requireJsonObject(params.steward_adoption_session, "original steward Session");
+  const row = binding(params.binding);
+  const allocation = original.manager_executor_allocation as JsonObject | undefined;
+  const endpoint = original.executor_endpoint_id;
+  if (row.context_kind !== "steward" || row.audience === "group"
+      || !["antigravity", "codex"].includes(String(row.executor_endpoint_id))
+      || !["antigravity", "codex"].includes(String(endpoint))
+      || original.agent_id !== endpoint || endpoint === row.executor_endpoint_id
+      || !["preferred", "pinned"].includes(String(allocation?.selection_policy))
+      || !["ready", "busy", "stale", "resuming"].includes(String(original.status))) return null;
+  const source = requireJsonObject(request.source, "original owner source");
+  const context = requireJsonObject(params.context, "verified steward context");
+  const saved = requireJsonObject(original.steward_context, "saved steward context");
+  if (request.binding_id !== row.binding_id || source.private_human_message !== true
+      || source.sender_ref !== row.operator_ref || context.operator_ref !== row.operator_ref
+      || context.binding_id !== row.binding_id || context.provider_ref !== row.provider_ref
+      || context.project_ref !== row.project_ref || context.source_ref !== source.source_ref
+      || context.audience !== "bound_owner" || context.grant !== "portfolio_read"
+      || original.goal_id !== "loopx-manager" || original.session_mode !== "managed_runtime"
+      || original.project_context != null
+      || original.channel_id !== `manager.external.native.${row.binding_id}.${source.source_ref}`
+      || !matchesStewardContext(saved, context, row)) {
+    throw new EffectRuntimeRequestError("the original steward Session belongs to another source");
+  }
+  return {operation: "hold_steward_binding", session_id: sessionIdentity(original.session_id, "original steward Session")};
 }
 
 function pureResultText(response: JsonObject | null, allowEmpty = false): boolean {

loopx/extensions/lark/private_conversations.py

@@ -17,7 +17,11 @@
 from threading import Lock
 from typing import Any
 
-from ...capabilities.native_chat.external_conversations import ChatExternalConversations, QUOTA_BACKLOG_BATCH_SIZE
+from ...capabilities.native_chat.external_conversations import (
+    ChatExternalConversations,
+    QUOTA_BACKLOG_BATCH_SIZE,
+    STEWARD_BINDING_ADOPTION_PENDING,
+)
 from ...capabilities.native_chat.conversation_bindings import ChatConversationBindings
 from ...capabilities.native_chat.file_discovery import FileChatDiscovery, file_change_fingerprint
 from ...chat_store import CHAT_MESSAGE_SCHEMA_VERSION, ChatSessionStore, TERMINAL_TURN_STATES, _atomic_write_json, _read_json
@@ -37,7 +41,7 @@
                              mark_lark_event_inbox_processing, mark_lark_event_inbox_received)
 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
+from .private_delivery_discovery import PrivateDeliveryDiscovery, prepared_steward_binding_waits
 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,
@@ -681,14 +685,15 @@ def _reject_command(path: Path, record: dict[str, Any], binding: dict[str, Any])
 
     @staticmethod
     def _record_admission(path: Path, record: dict[str, Any], admitted: dict[str, Any]) -> dict[str, Any]:
-        # Keep the existing recoverable capture until Core performs the frozen
-        # stop. A prepared command never claims execution or a delivered reply.
+        # Keep the recoverable capture while Core holds the original request.
         record.update(status="captured" if admitted["status"] == "prepared" else admitted["status"],
                       session_id=admitted["session_id"], turn_id=admitted["turn_id"],
                       response=admitted.get("response"), response_code=admitted.get("response_code"))
         if admitted.get("context_recovery"):
             record["context_recovery"] = admitted["context_recovery"]
         _atomic_write_json(path, record)
+        if admitted["status"] == "prepared" and admitted.get("response_code") == STEWARD_BINDING_ADOPTION_PENDING:
+            return {"status": "admission_deferred"}
         status = "durably_accepted" if admitted["status"] == "accepted" else (
             "queue_rejected" if admitted["status"] == "rejected" else "command_recorded")
         return {"status": status}
@@ -1947,6 +1952,13 @@ def reconcile_request(self, path: Path) -> int:
                 native_path = self.core.root / f"{row['request_ref']}.json"
                 if row["status"] == "captured" and native_path.exists():
                     native = self.core.read_request(row["request_ref"])
+                    if (row.get("response_code") == STEWARD_BINDING_ADOPTION_PENDING
+                            and native.get("status") == "prepared"
+                            and native.get("response_code") == STEWARD_BINDING_ADOPTION_PENDING
+                            and all(native.get(key) == row.get(key)
+                                    for key in ("request_ref", "binding_id", "source", "session_id", "turn_id"))
+                            and prepared_steward_binding_waits(native, self.bindings.read()["bindings"], self.core.controller.store)):
+                        return 0
                     if (native.get("command") == "stop" and native.get("target_recorded") is True
                             and all(native.get(key) == row.get(key)
                                     for key in ("request_ref", "binding_id", "source"))):
@@ -1966,6 +1978,8 @@ def reconcile_request(self, path: Path) -> int:
             if row["status"] == "captured":
                 self.admit(row["profile"], row["event"])
                 row = _read_json(path)
+                if row["status"] == "captured" and row.get("response_code") == STEWARD_BINDING_ADOPTION_PENDING:
+                    return 0
         native_path = self.core.root / f"{row['request_ref']}.json"
         if native_path.exists():
             if (self.reaction_feedback and not row.get("context_recovery") and row["status"] == "accepted"

loopx/extensions/lark/private_delivery_discovery.py

@@ -5,11 +5,25 @@
 from pathlib import Path
 from typing import Any, cast
 
-from ...chat_store import ChatSessionStore, _read_json
+from ...chat_store import RESUMABLE_SESSION_STATES, ChatSessionStore, _read_json
 from ...capabilities.native_chat.file_discovery import FileChatDiscovery, file_change_fingerprint
 from .private_progress import legacy_recovery_repairs
 
 
+def prepared_steward_binding_waits(
+    request: dict[str, Any], bindings: list[dict[str, Any]], store: ChatSessionStore
+) -> bool:
+    """Skip provider reads while a recorded executor mismatch persists."""
+    session = store.load_session(request["session_id"])
+    binding = next((row for row in bindings if row["binding_id"] == request["binding_id"]), None)
+    if not session or not binding or not binding.get("enabled"):
+        return False
+    allocation = session.get("manager_executor_allocation")
+    return (session.get("status") in RESUMABLE_SESSION_STATES
+            and session.get("executor_endpoint_id") != binding.get("executor_endpoint_id")
+            and isinstance(allocation, dict) and allocation.get("selection_policy") in ("preferred", "pinned"))
+
+
 class PrivateDeliveryDiscovery:
     def __init__(self, root: Path) -> None:
         self.root = root