From d80c03b52c82c8b64960ff8549eb0bb79a5e7d8f Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Wed, 11 Nov 2026 17:46:53 +0800 Subject: [PATCH] fix(chat): hold private steward input during executor adoption Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> --- .../app-conversation-and-async-inbox-v0.md | 8 ++ ...p-conversation-and-async-inbox-v0.zh-CN.md | 5 ++ .../native_chat/external_conversations.py | 25 +++++- .../collaboration/conversation_binding.ts | 42 +++++++++- .../extensions/lark/private_conversations.py | 22 ++++- .../lark/private_delivery_discovery.py | 16 +++- tests/test_lark_private_conversations.py | 84 +++++++++++++++++++ 7 files changed, 189 insertions(+), 13 deletions(-) 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 2f171a5d88..2746a0e2ee 100644 --- a/docs/architecture/rfcs/app-conversation-and-async-inbox-v0.md +++ b/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 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 1aeabed0b2..b2c7cb0de8 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 @@ -115,6 +115,11 @@ Lark listener 观测。该本地 hook 只读取缓存，不发起网络请求。 不回退到普通管家/项目历史，不新建替代 Session，也不在这里切换执行器或模型。 群受众保留原有边界。 +本人私聊管家的 Session 已采用新执行器、binding 仍指向原执行器时，新捕获的 +普通消息保持待准入，关联确切的原 Session。原回复 worker 等待两者对齐， +再次核验原消息后才入队。准入前关闭原 Session 会结束该请求，不创建替代 +会话。此前已拒绝的请求保留原结果。 + 飞书私聊回复的事件省略话题 ID 时，适配器从已核验的原消息上下文保留该 ID。 App 因此可以一起呈现原正文、精确引用和已核验的话题节选。重复到达时省略 该 ID 仍可沿用原请求，明确给出不同 ID 时继续拒绝。 diff --git a/loopx/capabilities/native_chat/external_conversations.py b/loopx/capabilities/native_chat/external_conversations.py index 7eb8e3c87d..c49ce6c592 100644 --- a/loopx/capabilities/native_chat/external_conversations.py +++ b/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") diff --git a/loopx/control_plane/collaboration/conversation_binding.ts b/loopx/control_plane/collaboration/conversation_binding.ts index f176966b4b..b7b8658342 100644 --- a/loopx/control_plane/collaboration/conversation_binding.ts +++ b/loopx/control_plane/collaboration/conversation_binding.ts @@ -323,6 +323,11 @@ function boundProject(params: JsonObject, row: JsonObject): Record (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 { diff --git a/loopx/extensions/lark/private_conversations.py b/loopx/extensions/lark/private_conversations.py index 0466d0a5d3..952dfccb13 100644 --- a/loopx/extensions/lark/private_conversations.py +++ b/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" diff --git a/loopx/extensions/lark/private_delivery_discovery.py b/loopx/extensions/lark/private_delivery_discovery.py index 673409f5bd..68f14e5bf9 100644 --- a/loopx/extensions/lark/private_delivery_discovery.py +++ b/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 diff --git a/tests/test_lark_private_conversations.py b/tests/test_lark_private_conversations.py index d20af3d8e6..9814201a59 100644 --- a/tests/test_lark_private_conversations.py +++ b/tests/test_lark_private_conversations.py @@ -20,6 +20,90 @@ from loopx.extensions.lark.goal_topic_runtime import poll_lark_goal_topic_profile_once +@pytest.mark.parametrize("outcome", ["adopted", "closed", "interrupted"]) +def test_private_steward_holds_original_request_during_executor_adoption(tmp_path, monkeypatch, outcome): + from test_native_steward_private import finish, steward + + fixture = steward.__wrapped__(tmp_path) + store, runtime, provider, transport, binding, capture, _ = next(fixture) + try: + event = provider.event("steward-app", "steward-original", "Read the original conversation") + transport.admit("steward-app", event) + original_ref = identity_ref(binding["provider_ref"], event["message_id"]) + original = transport.core.read_request(original_ref) + sid = original["session_id"] + 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() + + session = store.load_session(sid) + session["manager_executor_allocation"] = { + "schema_version": 1, "selection_policy": "pinned" if outcome == "interrupted" else "preferred"} + store._session_path(sid).write_text(json.dumps(session)) + bindings = transport.bindings + 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)) + pending_event = provider.event("steward-app", "steward-awaiting-executor", "Continue in the same 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) + + def open_obsolete_executor(**kwargs): + raise AssertionError(f"admission tried to open a replacement Session: {kwargs['agent_id']}") + + monkeypatch.setattr(runtime, "open_session", open_obsolete_executor) + assert transport.admit("steward-app", pending_event)["status"] == "admission_deferred" + held = transport.core.read_request(pending_ref) + assert held["status"] == "prepared" and held["session_id"] == sid and held["turn_id"] is None + assert held["response_code"] == "steward_executor_binding_pending" + provider_checks = len(provider.calls) + assert transport.reconcile_request(pending_path) == 0 + assert len(provider.calls) == provider_checks + assert transport.core.read_request(pending_ref) == held + assert capture.read_bytes() == calls_before and provider.writes == writes_before + assert not store.queued_turns(sid) + + if outcome == "closed": + assert runtime.close_session(sid) + bindings.path.write_text(json.dumps({**adopted, "revision": adopted["revision"] + 1})) + if outcome == "interrupted": + def interrupt_admission_projection(*args, **kwargs): + raise OSError("interrupted after Core admission") + + with monkeypatch.context() as interruption: + interruption.setattr(transport, "_record_admission", interrupt_admission_projection) + with pytest.raises(OSError, match="after Core admission"): + transport.reconcile_request(pending_path) + assert transport.core.read_request(pending_ref)["status"] == "accepted" + assert json.loads(pending_path.read_text())["status"] == "captured" + transport.reconcile_request(pending_path) + settled = transport.core.read_request(pending_ref) + assert settled["session_id"] == sid and "response_code" not in settled + if outcome == "closed": + assert settled["status"] == "rejected" and settled["turn_id"] is None + assert not store.turn_for_client(sid, f"external-{pending_ref}") + assert [row["session_id"] for row in store.list_sessions()] == [sid] + else: + assert settled["status"] == "accepted" + assert finish(runtime, settled)["status"] == "completed" + transport.reconcile_request(pending_path) + delivered = json.loads(pending_path.read_text()) + assert delivered["status"] == "delivered" and delivered["deliveries"]["terminal"]["verified"] + assert store.turn_for_client(sid, f"external-{pending_ref}")["turn_id"] == settled["turn_id"] + writes = list(provider.writes) + transport.reconcile_request(pending_path) + assert provider.writes == writes + assert sum(json.loads(line)["method"] == "turn/start" for line in capture.read_text().splitlines()) == 2 + 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)]