From 7fd09d1f8d202556e83690d0ff00055097977990 Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Sat, 7 Nov 2026 17:06:18 +0800 Subject: [PATCH 1/4] Steer bound Lark conversations from native bot mentions Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> --- .../src/features/personal-workspace/i18n.tsx | 12 +- .../native_chat/queued_admission.py | 67 ++++ loopx/chat_runtime.py | 20 +- loopx/control_plane/chat_turn_acceptance.py | 5 + .../control_plane/effect_runtime_handlers.ts | 1 + .../turn_driver/chat_queue_steering.ts | 31 ++ loopx/extensions/lark/goal_topic_dispatch.py | 73 +++- loopx/extensions/lark/goal_topic_runtime.py | 351 ++++++++++++++++-- .../lark/goal_topic_runtime_service.py | 16 +- .../chat_queue_steering.test.ts | 26 +- .../test_lark_goal_topic_runtime.py | 248 +++++++++++++ .../extensions/test_lark_realtime_dispatch.py | 65 ++++ .../test_lark_topic_dispatch_bounds.py | 44 +++ 13 files changed, 891 insertions(+), 68 deletions(-) diff --git a/apps/presentation/dashboard/src/features/personal-workspace/i18n.tsx b/apps/presentation/dashboard/src/features/personal-workspace/i18n.tsx index 6a6eadf865..b37a14cdff 100644 --- a/apps/presentation/dashboard/src/features/personal-workspace/i18n.tsx +++ b/apps/presentation/dashboard/src/features/personal-workspace/i18n.tsx @@ -846,7 +846,7 @@ const en = { "lark.turnTrigger": "When to respond", "lark.triggerAddressed": "Only when mentioned or replied to", "lark.triggerHuman": "Respond to group members without @", - "lark.triggerDescription": "Everyone in this connected group can start a conversation. Bot messages and historical backfill stay as context. Requires the Lark app permission to receive all group messages; tool and delegation permissions stay unchanged.", + "lark.triggerDescription": "Everyone in the connected group can start a conversation. A native @ of the bound Bot can adjust an active Turn. Bot messages and history stay as context. Receiving all group messages requires the corresponding Lark permission.", "lark.readGoalScope": "Goal evidence access", "lark.readAllGoals": "All registered Goals, including new ones", "lark.readSelectedGoals": "Keep the selected Goal scope", @@ -856,9 +856,9 @@ const en = { "lark.ingressLegacy": "Upgrade available", "lark.ingressLegacyDescription": "Save this connection to upgrade to the Agent inbox. Existing messages remain in the same Topic.", "lark.ingressQueue": "Queuing", - "lark.ingressQueueDescription": "Enter the same Agent Session's bounded FIFO queue and run after the current Turn.", + "lark.ingressQueueDescription": "Queue ordinary messages. A native @ adds text to a supported active Turn.", "lark.ingressSteering": "Steering", - "lark.ingressSteeringDescription": "Deliver to the active Turn and reject safely when no exact active Turn exists.", + "lark.ingressSteeringDescription": "Steer eligible messages. A native @ while idle starts a new Turn.", "lark.loading": "Loading Lark configuration…", "lark.management": "Lark management", "lark.openEventSettings": "Open Feishu event settings", @@ -2509,7 +2509,7 @@ const zhCN: Record = { "lark.turnTrigger": "何时回应", "lark.triggerAddressed": "仅 @ 或回复管家时", "lark.triggerHuman": "群成员直接发消息,无需 @", - "lark.triggerDescription": "此群所有成员都可以发起对话;机器人消息和补拉历史只作背景。需要飞书应用具备接收群内所有消息的权限;工具权限和委派授权保持不变。", + "lark.triggerDescription": "群成员可发起对话,原生 @ 已绑定的 Bot 可补充当前回合。机器人消息和补拉历史只作背景。接收群内所有消息需要相应飞书权限。", "lark.readGoalScope": "Goal 证据读取范围", "lark.readAllGoals": "全部已注册 Goal(含后续新增)", "lark.readSelectedGoals": "保留选定 Goal 范围", @@ -2519,9 +2519,9 @@ const zhCN: Record = { "lark.ingressLegacy": "待升级", "lark.ingressLegacyDescription": "保存连接即可升级为 Agent 异步收件箱,已有消息保留在原 Topic 中。", "lark.ingressQueue": "排队", - "lark.ingressQueueDescription": "进入同一 Agent Session 的有界 FIFO 队列,当前 Turn 完成后处理。", + "lark.ingressQueueDescription": "普通消息排队。原生 @ Bot 可向当前回合追加文字。", "lark.ingressSteering": "实时引导", - "lark.ingressSteeringDescription": "投到当前活跃 Turn;没有精确活跃 Turn 时安全拒绝。", + "lark.ingressSteeringDescription": "符合条件的消息补充当前回合。空闲时原生 @ Bot 可发起新回合。", "lark.loading": "加载 Lark 配置…", "lark.management": "Lark 管理", "lark.openEventSettings": "查看飞书事件配置", diff --git a/loopx/capabilities/native_chat/queued_admission.py b/loopx/capabilities/native_chat/queued_admission.py index 3f8c3fd814..7c3c30a76c 100644 --- a/loopx/capabilities/native_chat/queued_admission.py +++ b/loopx/capabilities/native_chat/queued_admission.py @@ -65,6 +65,73 @@ def enqueue_turn(controller: Any, *, session_id: str, client_turn_id: str, messa return turn, created +def deliver_native_mention( + controller, *, session_id: str, ingress_id: str, target_turn_id: str | None, + binding_matches: bool, text_only: bool, may_deliver_new: bool, + steering_text: str, queued_text: str, origin: str, +) -> dict[str, Any]: + """Project Chat receipts through the shared mention ingress rule.""" + from ...control_plane.chat_turn_acceptance import plan_native_mention_ingress + + def plan(): + receipt = controller.store.steering_receipt(session_id, ingress_id) if binding_matches else None + queued_turn_exists = ( + binding_matches and receipt is None + and controller.store.turn_for_client(session_id, ingress_id) is not None + ) + message = queued_text if (receipt or {}).get("mode") == "session_queue" else steering_text + return plan_native_mention_ingress( + mentioned=True, binding_matches=binding_matches, text_only=text_only, may_deliver_new=may_deliver_new, + target_turn_id=target_turn_id, queued_turn_exists=queued_turn_exists, + receipt=({key: receipt.get(key) for key in ("mode", "status", "expected_turn_id", "active_turn_id", "error_code")} + if receipt is not None else None), + receipt_message_matches=receipt is not None and receipt.get("message") == message, + ) + + try: + decision = plan() + if decision["action"] == "enqueue": + # Reserve the delivery mode before creating a FIFO Turn. Replays + # use this mode even if a different Turn is now running. + controller.store.create_ingress_receipt( + session_id, client_ingress_id=ingress_id, mode="session_queue", message=queued_text, + ) + elif decision["action"] == "steer": + try: + controller.steer_active_turn( + session_id=session_id, client_ingress_id=ingress_id, message=steering_text, + expected_turn_id=decision["target_turn_id"], origin=origin, + ) + except Exception: + # The native response can be lost after sending. The existing + # receipt, rather than the exception, decides whether it landed. + pass + decision = plan() + if decision["action"] == "steer": + return {"action": "unresolved"} + return decision + except (OSError, ValueError): + return {"action": "unresolved"} + + +def record_steering_delivery( + store, *, session_id: str, turn_id: str, ingress_id: str, + message: str, queued_turn_id: str | None, origin: str | None, +) -> None: + """Persist transcript evidence before settling the native ingress receipt.""" + payload = {"client_ingress_id": ingress_id} + if origin is not None: + payload["origin"] = origin + if queued_turn_id is None: + store.append_message(session_id, role="user", text=message, turn_id=turn_id, origin=origin) + else: + payload["queued_turn_id"] = queued_turn_id + store.append_event(session_id, turn_id, kind="turn.steered", payload=payload) + store.update_ingress_receipt( + session_id, ingress_id, status="delivered", active_turn_id=turn_id, error_code=None, + ) + + def _enqueue_bound(controller: Any, session: dict[str, Any], client_turn_id: str, message: str, attachments: list[dict[str, Any]] | None, origin: str) -> tuple[dict[str, Any], bool]: contexts = controller.project_contexts diff --git a/loopx/chat_runtime.py b/loopx/chat_runtime.py index bb4724f066..0a4b6d39a0 100644 --- a/loopx/chat_runtime.py +++ b/loopx/chat_runtime.py @@ -1361,6 +1361,7 @@ def plan(phase): def steer_active_turn( self, *, session_id: str, client_ingress_id: str, message: str, expected_turn_id: str | None = None, _queued_turn_id: str | None = None, + origin: str | None = None, ) -> tuple[dict[str, Any], bool]: """Steer the exact active Codex Turn with durable ingress deduplication.""" session = self._require_steering_session(session_id) @@ -1452,21 +1453,10 @@ def steer_active_turn( error_code="live_steering_rejected", ) raise - if _queued_turn_id is None: - self.store.append_message(session_id, role="user", text=message, turn_id=active_turn_id) - self.store.append_event( - session_id, - active_turn_id, - kind="turn.steered", - payload={"client_ingress_id": client_ingress_id, - **({"queued_turn_id": _queued_turn_id} if _queued_turn_id is not None else {})}, - ) - self.store.update_ingress_receipt( - session_id, - client_ingress_id, - status="delivered", - active_turn_id=active_turn_id, - error_code=None, + from .capabilities.native_chat.queued_admission import record_steering_delivery + record_steering_delivery( + self.store, session_id=session_id, turn_id=active_turn_id, ingress_id=client_ingress_id, + message=message, queued_turn_id=_queued_turn_id, origin=origin, ) turn = self.store.load_turn(session_id, active_turn_id) if turn is None: diff --git a/loopx/control_plane/chat_turn_acceptance.py b/loopx/control_plane/chat_turn_acceptance.py index de7a9fab8f..000839ec4c 100644 --- a/loopx/control_plane/chat_turn_acceptance.py +++ b/loopx/control_plane/chat_turn_acceptance.py @@ -184,6 +184,11 @@ def queued_chat_turn_ids( })["turn_ids"] +def plan_native_mention_ingress(**facts: object) -> dict[str, Any]: + """Ask the shared queue/steering owner to select one exact ingress.""" + return effect_runtime_result("chat.queue.mention", facts) + + def plan_queued_chat_steering( session: dict[str, Any], turn: dict[str, Any], active: dict[str, Any] | None, receipt: dict[str, Any] | None, *, ingress_id: str, operation_id: str, target: str, phase: str, diff --git a/loopx/control_plane/effect_runtime_handlers.ts b/loopx/control_plane/effect_runtime_handlers.ts index d8653e76d5..693942db67 100644 --- a/loopx/control_plane/effect_runtime_handlers.ts +++ b/loopx/control_plane/effect_runtime_handlers.ts @@ -628,6 +628,7 @@ export function createEffectRuntimeHandlers( ["collaboration.conversation.reply_context", lazyHandler(() => import("./collaboration/conversation_reply_context.ts"), ({projectConversationReplyContext}) => projectConversationReplyContext)], ["chat.turn.accept", lazyHandler(() => import("./turn_driver/chat_turn_acceptance.ts"), ({planChatTurnAcceptance}) => planChatTurnAcceptance)], ["chat.queue.select", lazyHandler(() => import("./turn_driver/chat_queue_steering.ts"), ({selectQueuedChatTurns}) => selectQueuedChatTurns)], + ["chat.queue.mention", lazyHandler(() => import("./turn_driver/chat_queue_steering.ts"), ({planNativeMentionIngress}) => planNativeMentionIngress)], ["chat.queue.steer", lazyHandler(() => import("./turn_driver/chat_queue_steering.ts"), ({planQueuedChatSteering}) => planQueuedChatSteering)], ["chat.queue.mutate", lazyHandler(() => import("./turn_driver/chat_queue_steering.ts"), ({planQueuedChatMutation}) => planQueuedChatMutation)], ["chat.turn.execution_allowed", lazyHandler(() => import("./turn_driver/chat_turn_acceptance.ts"), ({mayContinueChatTurn}) => mayContinueChatTurn)], diff --git a/loopx/control_plane/turn_driver/chat_queue_steering.ts b/loopx/control_plane/turn_driver/chat_queue_steering.ts index f42fddb216..2e429cc8ae 100644 --- a/loopx/control_plane/turn_driver/chat_queue_steering.ts +++ b/loopx/control_plane/turn_driver/chat_queue_steering.ts @@ -54,6 +54,37 @@ export function selectQueuedChatTurns(input: JsonObject): JsonObject { return {turn_ids: ids}; } +/** Select delivery for a provider-verified native mention in an exact Session. + * A recorded ingress owns retries, even after the observed Turn has changed. + */ +export function planNativeMentionIngress(input: JsonObject): JsonObject { + if (input.mentioned !== true) return {action: "configured"}; + if (input.binding_matches !== true) return {action: "reject", error_code: "live_steering_session_not_attached"}; + if (input.text_only !== true) return {action: "reject", error_code: "queue_steering_attachments_unavailable"}; + + if (input.receipt != null) { + const receipt = requireJsonObject(input.receipt, "mention ingress receipt"); + if (input.receipt_message_matches !== true) return {action: "unresolved"}; + if (receipt.mode === "session_queue") return {action: "enqueue"}; + if (receipt.mode !== "live_steering") return {action: "unresolved"}; + if (receipt.status === "delivered" && typeof receipt.active_turn_id === "string" && + receipt.active_turn_id && (receipt.expected_turn_id == null || receipt.expected_turn_id === receipt.active_turn_id)) { + return {action: "delivered", target_turn_id: receipt.active_turn_id}; + } + if (receipt.status === "failed" && NOT_DELIVERED.has(String(receipt.error_code))) { + return {action: "reject", error_code: receipt.error_code}; + } + return {action: "unresolved"}; + } + + if (input.queued_turn_exists === true) return {action: "enqueue"}; + // A recovered source event without a Chat receipt cannot bind to a newer + // active Turn. Keep the original message available for explicit recovery. + if (input.may_deliver_new !== true) return {action: "unresolved"}; + if (input.target_turn_id === null) return {action: "enqueue"}; + return {action: "steer", target_turn_id: requireNonEmptyString(input.target_turn_id, "target_turn_id")}; +} + export function planQueuedChatSteering(input: JsonObject): QueueSteeringPlan { const session = requireJsonObject(input.session, "session"); const turn = requireJsonObject(input.turn, "queued turn"); diff --git a/loopx/extensions/lark/goal_topic_dispatch.py b/loopx/extensions/lark/goal_topic_dispatch.py index b0dc46f443..e3042e7f41 100644 --- a/loopx/extensions/lark/goal_topic_dispatch.py +++ b/loopx/extensions/lark/goal_topic_dispatch.py @@ -21,6 +21,7 @@ def __init__( *, workers: int = 4, capacity: int = 64, + reserve_priority: bool = False, ) -> None: if workers < 1 or capacity < workers: raise ValueError("dispatch capacity must cover its positive worker limit") @@ -28,11 +29,14 @@ def __init__( self._stop = stop self._workers = workers self._capacity = capacity + self._reserve_priority = reserve_priority and workers > 1 self._condition = threading.Condition() - self._waiting: dict[str, deque[tuple[tuple[str, ...], str]]] = {} + self._waiting: dict[str, deque[tuple[tuple[str, ...], str, bool]]] = {} self._ready: deque[str] = deque() - self._active: dict[str, tuple[str, ...]] = {} + self._active: dict[str, tuple[tuple[str, ...], bool]] = {} self._pending = 0 + self._ordinary_pending = 0 + self._priority_pending = 0 self._closing = False self._cancelled = False self._failure: Exception | None = None @@ -47,9 +51,23 @@ def failed(self) -> bool: with self._condition: return self._failure is not None - def submit(self, lane: tuple[str, ...], line: str) -> bool: + @property + def ordinary_full(self) -> bool: + """Report whether ordinary admission has reached its non-priority bound.""" + with self._condition: - while self._pending >= self._capacity and not self._closing: + return self._full(False) + + def _full(self, priority: bool) -> bool: + if not self._reserve_priority: + return self._pending >= self._capacity + if priority: + return self._priority_pending >= self._workers + return self._ordinary_pending >= self._capacity - 1 or self._pending >= self._capacity + + def submit(self, lane: tuple[str, ...], line: str, *, priority: bool = False) -> bool: + with self._condition: + while self._full(priority) and not self._closing: if self._stop.is_set(): return False self._condition.wait(0.1) @@ -59,10 +77,17 @@ def submit(self, lane: tuple[str, ...], line: str) -> bool: return False source = lane[0] queue = self._waiting.setdefault(source, deque()) - queue.append((lane, line)) + queue.append((lane, line, priority)) self._pending += 1 + if priority: + self._priority_pending += 1 + else: + self._ordinary_pending += 1 if source not in self._active and source not in self._ready: - self._ready.append(source) + if priority: + self._ready.appendleft(source) + else: + self._ready.append(source) self._start_ready() return True @@ -75,22 +100,36 @@ def _start_ready(self) -> None: if self._cancelled or len(self._active) >= self._workers: break source = self._ready.popleft() - lane, line = self._waiting[source][0] - occupied = {key for active in self._active.values() for key in active} + lane, line, priority = self._waiting[source][0] + active_ordinary = sum( + 1 for _lane, is_priority in self._active.values() if not is_priority + ) + # Keep one of the existing workers available for a verified live + # mention while ordinary answer handlers wait on Chat completion. + if self._reserve_priority and not priority and active_ordinary >= self._workers - 1: + self._ready.append(source) + continue + occupied = {key for active, _is_priority in self._active.values() for key in active} if occupied.intersection(lane): self._ready.append(source) continue self._waiting[source].popleft() - self._active[source] = lane - self._executor.submit(self._run, source, line) + self._active[source] = (lane, priority) + self._executor.submit(self._run, source, line, priority) def _discard_waiting(self) -> None: self._cancelled = True - self._pending -= sum(len(queue) for queue in self._waiting.values()) + for queue in self._waiting.values(): + for _lane, _line, priority in queue: + self._pending -= 1 + if priority: + self._priority_pending -= 1 + else: + self._ordinary_pending -= 1 self._waiting.clear() self._ready.clear() - def _run(self, source: str, line: str) -> None: + def _run(self, source: str, line: str, priority: bool) -> None: try: self._handle(line) except Exception as exc: @@ -102,10 +141,17 @@ def _run(self, source: str, line: str) -> None: finally: with self._condition: self._pending -= 1 + if priority: + self._priority_pending -= 1 + else: + self._ordinary_pending -= 1 self._active.pop(source) queue = self._waiting.get(source) if queue: - self._ready.append(source) + if queue[0][2]: + self._ready.appendleft(source) + else: + self._ready.append(source) else: self._waiting.pop(source, None) self._start_ready() @@ -116,6 +162,7 @@ def close(self, *, cancel_pending: bool = False) -> None: with self._condition: self._closing = True + self._condition.notify_all() while self._pending: if cancel_pending or self._stop.is_set(): self._discard_waiting() diff --git a/loopx/extensions/lark/goal_topic_runtime.py b/loopx/extensions/lark/goal_topic_runtime.py index 63b9e1c858..a57637ec39 100644 --- a/loopx/extensions/lark/goal_topic_runtime.py +++ b/loopx/extensions/lark/goal_topic_runtime.py @@ -9,12 +9,13 @@ import re import subprocess import threading +from collections import deque from collections.abc import Callable, Mapping from datetime import datetime, timezone from pathlib import Path from typing import Any -from ...chat_manager import MANAGER_AGENT_OBJECTIVE +from ...chat_manager import MANAGER_AGENT_OBJECTIVE, manager_channel from ...control_plane.collaboration import conversation_reply_context from ...file_lock import exclusive_file_lock from .manager_routing import ( @@ -27,6 +28,7 @@ ManagerAuthorityMode, ) from ..external_connector_runtime import ( + ExternalEffectKind, ExternalResponsePolicy, build_external_event_response_receipt, decide_external_event_ack, @@ -44,7 +46,7 @@ normalize_lark_topic_event_rejection_reason, ) from .goal_channel_targets import goal_channel_target_for_name -from .goal_channel_transport import record_event_timing +from .goal_channel_transport import lark_provider_mention_identities, record_event_timing from .goal_topic_connections import decide_lark_topic_event from .goal_topic_dispatch import ProfileEventDispatch from .inbox_reply import CommandRunner, reply_lark_event_inbox @@ -399,15 +401,20 @@ def _profile_event_lane( snapshot: Mapping[str, Any], event: Mapping[str, Any], runtime_root: str | Path, -) -> tuple[str, ...]: - """Only serialize transport work; actual ingress re-reads its authority.""" + capture_mention: Callable[[Mapping[str, object]], dict[str, object]] | None = None, + include_capture: bool = False, +) -> Any: + """Capture a live target before joining the ordinary reply lane.""" + + def result(lane: tuple[str, ...], capture: dict[str, object] | None = None) -> Any: + return (lane, capture) if include_capture else lane private_config = dict(snapshot.get("private_profiles") or {}).get(profile) if isinstance(private_config, Mapping) and _native_profile_event(snapshot, profile, event): from .conversation_identity import lark_private_source try: - source = lark_private_source( + private_source = lark_private_source( provider_ref=private_config["provider_ref"], event=event ) except (KeyError, TypeError, ValueError): @@ -417,7 +424,7 @@ def _profile_event_lane( # Use the same App/chat/topic (or private sender) source shape for # scheduling, never a grant or a persisted Session selection. Core # still verifies the provider and acquires its fresh source fence. - return ("private." + source["source_ref"],) + return result(("private." + private_source["source_ref"],)) scoped, _status = _profile_event_route( profile=profile, snapshot=snapshot, event=event @@ -440,10 +447,16 @@ def _profile_event_lane( route.get("app_ref"), route.get("target_ref"), route.get("topic_root_message_id"), ) + capture = ( + capture_mention(route) if capture_mention is not None + and _route_native_mention(routed_event, route, targets) else None + ) + if capture and capture.get("target_turn_id"): + return result(("steer." + source, "steer.session." + str(route["session_id"])), capture) if route.get("session_id"): - return (source, "session." + str(route["session_id"])) - return (source,) - return ("unmatched." + _opaque_digest(profile, event.get("chat_id")),) + return result((source, "session." + str(route["session_id"])), capture) + return result((source,), capture) + return result(("unmatched." + _opaque_digest(profile, event.get("chat_id")),)) def poll_lark_goal_topic_profile_once( @@ -457,6 +470,7 @@ def poll_lark_goal_topic_profile_once( reply_runner: CommandRunner = _default_simple_runner, proposal_deliverer: ProposalDeliverer | None = None, private_admitter: Callable[[str, dict[str, object]], Mapping[str, object]] | None = None, + mention_capture: Mapping[str, object] | None = None, ) -> dict[str, Any]: """Consume one bounded event batch for an App and reuse Inbox reply/ACK.""" @@ -527,6 +541,7 @@ def poll_lark_goal_topic_profile_once( reply_runner=reply_runner, provider_runner=provider_runner, proposal_deliverer=proposal_deliverer, + mention_capture=mention_capture, ) except Exception: event_statuses.append("processing_failed") @@ -558,6 +573,7 @@ def stream_lark_goal_topic_profile( health_sink: HealthSink | None = None, proposal_deliverer: ProposalDeliverer | None = None, private_admitter: Callable[[str, dict[str, object]], Mapping[str, object]] | None = None, + capture_mention: Callable[[Mapping[str, object]], dict[str, object]] | None = None, review_callback_handler: ReviewCallbackHandler | None = None, ) -> dict[str, Any]: """Keep one bounded long-lived CLI consumer attached between messages.""" @@ -669,7 +685,8 @@ def stop_consumer() -> None: def handle_event(line: str) -> None: nonlocal event_count, replied_count, provider_ready - event = _event_payloads(line)[0] + envelope = json.loads(line) + event = envelope["event"] message_id = str(event.get("message_id") or "") record_event_timing(profile, message_id, "dispatch_started") current_snapshot = snapshot_provider() @@ -679,15 +696,16 @@ def handle_event(line: str) -> None: snapshot=current_snapshot, runtime_root=runtime_root, answer=answer, - consume_runner=lambda _args, payload=line: { + consume_runner=lambda _args: { "returncode": 0, - "stdout": payload, + "stdout": json.dumps(event, ensure_ascii=False), "stderr": "", }, provider_runner=provider_runner, reply_runner=reply_runner, proposal_deliverer=proposal_deliverer, private_admitter=private_admitter, + mention_capture=envelope["mention_capture"], ) record_event_timing(profile, message_id, "dispatch_returned") with result_lock: @@ -733,7 +751,67 @@ def handle_event(line: str) -> None: flush=True, ) - dispatch = ProfileEventDispatch(handle_event, stop) + reserve_priority = capture_mention is not None and _profile_supports_mention_steering( + snapshot, profile, capture_mention + ) + dispatch = ProfileEventDispatch(handle_event, stop, reserve_priority=reserve_priority) + ordinary_backlog: deque[tuple[tuple[str, ...], str]] = deque() + backlog_condition = threading.Condition() + backlog_closed = False + backlog_worker: threading.Thread | None = None + + def drain_ordinary_backlog() -> None: + while True: + with backlog_condition: + while not ordinary_backlog and not backlog_closed: + if stop.is_set() or dispatch.failed: + return + backlog_condition.wait(0.05) + if backlog_closed or stop.is_set() or dispatch.failed: + ordinary_backlog.clear() + backlog_condition.notify_all() + return + lane, envelope = ordinary_backlog[0] + try: + submitted = dispatch.submit(lane, envelope, priority=False) + except Exception: + with backlog_condition: + ordinary_backlog.clear() + backlog_condition.notify_all() + return + with backlog_condition: + if ordinary_backlog: + ordinary_backlog.popleft() + if not submitted: + ordinary_backlog.clear() + backlog_condition.notify_all() + if not submitted: + return + + def submit_event(lane: tuple[str, ...], envelope: str, *, priority: bool) -> bool: + nonlocal backlog_worker + if priority or not reserve_priority: + return dispatch.submit(lane, envelope, priority=priority) + with backlog_condition: + if not ordinary_backlog and not dispatch.ordinary_full: + return dispatch.submit(lane, envelope, priority=False) + while len(ordinary_backlog) >= 64 and not backlog_closed: + if stop.is_set() or dispatch.failed: + return False + backlog_condition.wait(0.05) + if backlog_closed or stop.is_set() or dispatch.failed: + return False + ordinary_backlog.append((lane, envelope)) + if backlog_worker is None: + backlog_worker = threading.Thread( + target=drain_ordinary_backlog, + name=f"loopx-lark-ordinary-{profile}", + daemon=True, + ) + backlog_worker.start() + backlog_condition.notify_all() + return True + watcher.start() stream_failed = True try: @@ -766,20 +844,35 @@ def handle_event(line: str) -> None: for event in _event_payloads(line): message_id = str(event.get("message_id") or "") record_event_timing(profile, message_id, "received") - lane = _profile_event_lane( + lane, mention_capture = _profile_event_lane( profile=profile, snapshot=snapshot_provider(), event=event, runtime_root=runtime_root, + capture_mention=capture_mention, + include_capture=True, ) record_event_timing(profile, message_id, "lane_ready") - if not dispatch.submit(lane, json.dumps(event, ensure_ascii=False)): + envelope = json.dumps({"event": event, "mention_capture": mention_capture}, ensure_ascii=False) + if not submit_event(lane, envelope, priority=bool((mention_capture or {}).get("target_turn_id"))): break + while True: + with backlog_condition: + if not ordinary_backlog or stop.is_set() or dispatch.failed: + break + backlog_condition.wait(0.05) stream_failed = False finally: + with backlog_condition: + backlog_closed = True + if stream_failed or stop.is_set() or dispatch.failed: + ordinary_backlog.clear() + backlog_condition.notify_all() try: dispatch.close(cancel_pending=stream_failed) finally: + if backlog_worker is not None: + backlog_worker.join(timeout=1) watcher_done.set() if process.poll() is None: process.terminate() @@ -835,6 +928,145 @@ def handle_event(line: str) -> None: } +def _profile_supports_mention_steering( + snapshot: Mapping[str, Any], + profile: str, + capture_mention: Callable[[Mapping[str, object]], dict[str, object]] | None = None, +) -> bool: + """Reserve a priority worker only when a bound Session can steer on @mention.""" + + profile_config = _active_profile_configs(snapshot).get(profile) + if profile_config is None: + return False + bot_app_id = str(profile_config.get("bot_app_id") or "") + target_payload = snapshot.get("target_payload") + target_payload = target_payload if isinstance(target_payload, Mapping) else {} + binding_payloads = snapshot.get("binding_payloads") + binding_payloads = binding_payloads if isinstance(binding_payloads, Mapping) else {} + for goal_id, payload in binding_payloads.items(): + if not isinstance(payload, Mapping): + continue + for binding in bindings_for_goal(payload, str(goal_id)): + if binding.get("enabled") is not True: + continue + session_id = str(binding.get("session_id") or "").strip() + routing = binding.get("routing") + routing = routing if isinstance(routing, Mapping) else {} + ingress_mode = str(routing.get("ingress_mode") or "") + if not session_id or ingress_mode not in {"live_steering", "session_queue"}: + continue + target = goal_channel_target_for_name( + target_payload, str(binding.get("target_ref") or "") + ) + if target is None or target.get("enabled") is not True: + continue + identity = target.get("identity") + identity = identity if isinstance(identity, Mapping) else {} + sender_profile = str(identity.get("sender_profile") or "").strip() + same_app = bool(bot_app_id) and str(identity.get("bot_app_id") or "") == bot_app_id + if sender_profile != profile and not same_app: + continue + if not (identity.get("bot_app_id") or identity.get("bot_open_id")): + continue + if capture_mention is not None: + channel = target.get("channel") + channel = channel if isinstance(channel, Mapping) else {} + chat_id = str(channel.get("chat_id") or "") + kind = str(routing.get("conversation_kind") or "goal") + route = { + "goal_id": str(goal_id), + "connection_id": str(binding.get("connection_id") or ""), + "agent_id": str(binding.get("agent_id") or "codex"), + "session_id": session_id, + "conversation_kind": kind, + "manager_channel_id": ( + manager_channel( + provider="lark", + audience=f"{sender_profile or 'default'}\0{chat_id}", + ) + if kind == "manager" + else "" + ), + "ingress_mode": ingress_mode, + } + if capture_mention(route).get("binding_matches") is not True: + continue + return True + return False + + +def _bound_topic_session( + route: Mapping[str, object], runtime_controller: Any +) -> Mapping[str, Any] | None: + session_id = str(route.get("session_id") or "") + store = getattr(runtime_controller, "store", None) + if not session_id or store is None: + return None + session = store.load_session(session_id) + manager = route.get("conversation_kind") == "manager" + agent_id = manager_turn_executor(runtime_controller) if manager else str(route.get("agent_id") or "codex") + goal_id = str(route.get("goal_id") or "") + channel_id = str(route.get("manager_channel_id") or "") if manager else f"goal.{goal_id}" + if ( + not isinstance(session, Mapping) or session.get("status") == "closed" + or (not manager and session.get("goal_id") != goal_id) + or session.get("agent_id") != agent_id or session.get("channel_id") != channel_id + ): + return None + return session + + +def capture_topic_mention( + route: Mapping[str, object], runtime_controller: Any +) -> dict[str, object]: + """Read the bound Session/Turn without opening or resuming an executor.""" + session = _bound_topic_session(route, runtime_controller) + return { + "connection_id": route.get("connection_id"), + "session_id": route.get("session_id"), + "binding_matches": session is not None, + "target_turn_id": (session or {}).get("active_turn_id") or None, + } + + +def _native_topic_mention(event: Mapping[str, object], identity: Mapping[str, object]) -> bool: + if event.get("sender_type") != "user" or not event.get("sender_id") or event.get("historical_context_only"): + return False + expected = {str(identity.get(key) or "") for key in ("bot_app_id", "bot_open_id")} - {""} + mentions = event.get("mentions") + return bool(expected and isinstance(mentions, list) and any( + isinstance(item, Mapping) and lark_provider_mention_identities(item).intersection(expected) + for item in mentions + )) + + +def _route_native_mention(event: Mapping[str, object], route: Mapping[str, object], + target_payload: Mapping[str, object]) -> bool: + if route.get("ingress_mode") not in {"live_steering", "session_queue", "async_inbox"}: + return False + if (route.get("conversation_kind") == "manager" + and route.get("authority_mode") != ManagerAuthorityMode.TURN_AUTHORIZED.value): + return False + target = goal_channel_target_for_name(target_payload, str(route.get("target_ref") or "")) or {} + identity = target.get("identity") + return isinstance(identity, Mapping) and _native_topic_mention(event, identity) + + +def _mention_reply(route: Mapping[str, object], decision: Mapping[str, object]) -> dict[str, object]: + receipt = _session_turn_effect(route) + if decision["action"] == "delivered": + return {"response_text": "已送达当前回合。", "effect_receipt": receipt} + receipt["effect_kind"] = ExternalEffectKind.NO_FOLLOW_UP.value + if decision["action"] == "unresolved": + return {"response_text": "送达状态待核实。原消息保留在 Lark。", "effect_receipt": receipt} + text = { + "queue_steering_attachments_unavailable": "本轮调整仅支持文字。原消息和附件保留在 Lark。", + "live_steering_turn_mismatch": "接收时的回合已结束或切换。原消息保留在 Lark。", + "live_steering_turn_not_started": "接收时的回合尚未就绪。原消息保留在 Lark。", + }.get(str(decision.get("error_code")), "当前会话无法调整本轮。原消息保留在 Lark。") + return {"response_text": text, "effect_receipt": receipt} + + def answer_lark_goal_topic( *, route: Mapping[str, Any], @@ -878,17 +1110,52 @@ def answer_lark_goal_topic( route.get("message_id"), route.get("topic_root_message_id"), ) + native_mention = route.get("native_mention") is True + if native_mention: + from ...capabilities.native_chat.queued_admission import deliver_native_mention + + capture = route.get("mention_capture") + capture = capture if isinstance(capture, Mapping) else capture_topic_mention(route, runtime_controller) + current = _bound_topic_session(route, runtime_controller) + binding_matches = ( + current is not None and capture.get("binding_matches") is True + and capture.get("session_id") == session_id + and capture.get("connection_id") == route.get("connection_id") + and ingress_mode in {"live_steering", "session_queue"} + ) + if manager and binding_matches and current is not None and capture.get("target_turn_id"): + from ...chat_manager_context import manager_authorization_scope_is_current + + binding_matches = manager_authorization_scope_is_current( + runtime_controller.registry_path, runtime_controller.manager_scope_resolver, current, + current.get("manager_authorization_scope_id"), + runtime_root=runtime_controller.coordination_runtime_root, + ) + body = text + try: + payload = json.loads(text) + except (TypeError, ValueError): + payload = None + if isinstance(payload, Mapping) and isinstance(payload.get("text"), str): + body = payload["text"] + target_turn_id = capture.get("target_turn_id") + decision = deliver_native_mention( + runtime_controller, session_id=session_id, ingress_id=client_turn_id, + target_turn_id=str(target_turn_id) if isinstance(target_turn_id, str) and target_turn_id else None, + binding_matches=binding_matches, + text_only=route.get("native_mention_text_only") is True, + may_deliver_new=route.get("native_mention_fresh") is True, + steering_text=(reply_context + "\n\n" if reply_context else "") + body, + queued_text=message, origin="lark", + ) + if decision["action"] != "enqueue": + return _mention_reply(route, decision) + ingress_mode = "session_queue" + if ingress_mode in {"live_steering", "session_queue"}: - session = runtime_controller.store.load_session(session_id) - if ( - session is None - or (not manager and session.get("goal_id") != goal_id) - or session.get("agent_id") != agent_id - or session.get("channel_id") != expected_channel - or session.get("status") == "closed" - ): + if _bound_topic_session(route, runtime_controller) is None: if manager and manager_session_requires_executor_rebind( - session, + runtime_controller.store.load_session(session_id), expected_channel=expected_channel, agent_id=agent_id, ): @@ -896,9 +1163,7 @@ def answer_lark_goal_topic( "manager_channel_executor_rebind_required", _session_turn_effect(route), ) - raise RuntimeError( - "bound Agent session is unavailable or no longer matches" - ) + raise RuntimeError("bound Agent session is unavailable or no longer matches") if ingress_mode == "live_steering": turn, _created = runtime_controller.steer_active_turn( session_id=session_id, @@ -934,6 +1199,10 @@ def answer_lark_goal_topic( objective=str(objective or goal_id), origin="lark", ) + if native_mention: + runtime_controller.store.update_ingress_receipt( + session_id, client_turn_id, status="delivered", + ) else: # Compatibility path for bindings created before the three ingress modes. channel_id = "lark." + _opaque_digest( @@ -1055,8 +1324,9 @@ def process_lark_goal_topic_event( reply_runner: CommandRunner, provider_runner: Any | None = None, proposal_deliverer: ProposalDeliverer | None = None, + mention_capture: Mapping[str, object] | None = None, ) -> dict[str, Any]: - """Single-flight an addressed manager message through answer, reply and ACK. + """Single-flight a manager or live mention through ingress, reply and ACK. The inbox and delivery receipt make sequential retries safe, but two listeners can otherwise both observe an absent receipt and generate two @@ -1071,10 +1341,20 @@ def process_lark_goal_topic_event( runtime_root=runtime_root, ) route = decision.get("route") - if ( - isinstance(route, Mapping) - and route.get("conversation_kind") == "manager" - and route.get("authority_mode") == ManagerAuthorityMode.TURN_AUTHORIZED.value + native_mention = isinstance(route, Mapping) and _route_native_mention(event, route, target_payload) + if mention_capture is not None and not native_mention: + return {"ok": True, "status": "ignored", "reason": "mention_binding_changed"} + if native_mention: + route = { + **route, + "native_mention": True, + "native_mention_text_only": event.get("message_type") == "text" and not event.get("attachment_count"), + "mention_capture": mention_capture, + } + decision = {**decision, "route": route} + if isinstance(route, Mapping) and ( + native_mention or (route.get("conversation_kind") == "manager" + and route.get("authority_mode") == ManagerAuthorityMode.TURN_AUTHORIZED.value) ): message_id = str(route.get("message_id") or "") if not MESSAGE_ID_PATTERN.fullmatch(message_id): @@ -1252,6 +1532,7 @@ def _process_lark_goal_topic_event( "goal_id": route["goal_id"], "agent_id": route.get("agent_id"), "inbox_config_ref": config_ref, + **({"reason": "async_inbox_requires_bound_session"} if route.get("native_mention") else {}), } # Receipt ACK is visible while the synchronous manager is reasoning. The # private reaction ledger makes retries idempotent. Manager received ACKs @@ -1295,6 +1576,11 @@ def _process_lark_goal_topic_event( captured_event = next( item for item in projection["items"] if item.get("message_id") == message_id ) + if route.get("native_mention") and ( + not captured_event.get("target_mention_count") or captured_event.get("historical_context_only") + or captured_event.get("sender_id") != canonical["sender_id"] + ): + return {"ok": False, "status": "native_mention_proof_changed", "goal_id": route["goal_id"]} route = { **route, "source_sender_id": str(canonical.get("sender_id") or ""), @@ -1305,6 +1591,7 @@ def _process_lark_goal_topic_event( "root_id": captured_event.get("root_id", ""), "thread_id": captured_event.get("thread_id", ""), "thread_context": captured_event.get("thread_context"), + **({"native_mention_fresh": int(ingest.get("accepted_count") or 0) > 0} if route.get("native_mention") else {}), **({"context_materials": context_materials} if context_materials else {}), } delivery_path: Path | None = None diff --git a/loopx/extensions/lark/goal_topic_runtime_service.py b/loopx/extensions/lark/goal_topic_runtime_service.py index 0cea1658f9..17708174a4 100644 --- a/loopx/extensions/lark/goal_topic_runtime_service.py +++ b/loopx/extensions/lark/goal_topic_runtime_service.py @@ -306,6 +306,7 @@ def answer( effective_route = route if ( route.get("conversation_kind") == "manager" + and route.get("native_mention") is not True and self.manager_route_reconciler is not None ): try: @@ -343,9 +344,10 @@ def answer( else: response_text = answer_result proposal_ids = [] + effect_receipt = answer_result.get("effect_receipt") if isinstance(answer_result, Mapping) else None return { "response_text": response_text, - "effect_receipt": session_turn_effect(effective_route), + "effect_receipt": effect_receipt or session_turn_effect(effective_route), "proposal_ids": proposal_ids, } @@ -409,12 +411,24 @@ def handle_review_callback( ) stop.wait(5) continue + def capture(route: Mapping[str, object]) -> dict[str, object]: + return runtime.capture_topic_mention( + route, self.runtime_controller + ) + result = runtime.stream_lark_goal_topic_profile( profile=profile, snapshot_provider=self.snapshot_provider, stop=stop, runtime_root=self.runtime_root, answer=answer, + capture_mention=( + capture + if runtime._profile_supports_mention_steering( + self.snapshot_provider(), profile, capture + ) + else None + ), private_admitter=(self._admit_private if self.private_conversations is not None else None), proposal_deliverer=deliver_proposals, diff --git a/tests/control_plane_ts/chat_queue_steering.test.ts b/tests/control_plane_ts/chat_queue_steering.test.ts index 75d7ee327a..21e33f6393 100644 --- a/tests/control_plane_ts/chat_queue_steering.test.ts +++ b/tests/control_plane_ts/chat_queue_steering.test.ts @@ -1,6 +1,6 @@ import assert from "node:assert/strict"; import test from "node:test"; -import {planQueuedChatMutation, planQueuedChatSteering, selectQueuedChatTurns, QUEUE_STEERING_SCHEMA} from "../../loopx/control_plane/turn_driver/chat_queue_steering.ts"; +import {planNativeMentionIngress, planQueuedChatMutation, planQueuedChatSteering, selectQueuedChatTurns, QUEUE_STEERING_SCHEMA} from "../../loopx/control_plane/turn_driver/chat_queue_steering.ts"; function facts() { return {phase: "prepare", session: {status: "busy", active_turn_id: "active"}, active_status: "running", @@ -96,3 +96,27 @@ test("acknowledgement recovers after target completion while proven rejection re assert.equal(planQueuedChatSteering({...f, session: {status: "ready", active_turn_id: null}}).action, "release"); assert.equal(planQueuedChatSteering({...f, phase: "settle"}).action, "release"); }); + +test("native mention admission preserves the observed target and prior ingress", () => { + const input = {mentioned: true, binding_matches: true, text_only: true, may_deliver_new: true, + target_turn_id: "first", receipt_message_matches: true}; + assert.deepEqual(planNativeMentionIngress({...input, mentioned: false}), {action: "configured"}); + assert.deepEqual(planNativeMentionIngress(input), {action: "steer", target_turn_id: "first"}); + assert.deepEqual(planNativeMentionIngress({...input, target_turn_id: null}), {action: "enqueue"}); + assert.deepEqual(planNativeMentionIngress({...input, may_deliver_new: false, target_turn_id: "later"}), + {action: "unresolved"}); + assert.deepEqual(planNativeMentionIngress({...input, queued_turn_exists: true}), {action: "enqueue"}); + assert.equal(planNativeMentionIngress({...input, binding_matches: false}).action, "reject"); + assert.equal(planNativeMentionIngress({...input, text_only: false}).error_code, "queue_steering_attachments_unavailable"); + const receipt = {mode: "live_steering", status: "delivered", expected_turn_id: "first", active_turn_id: "first"}; + assert.deepEqual(planNativeMentionIngress({...input, receipt, may_deliver_new: false, target_turn_id: "later"}), + {action: "delivered", target_turn_id: "first"}); + for (const uncertain of [{...receipt, status: "pending"}, {...receipt, active_turn_id: "later"}, + {...receipt, status: "failed", error_code: "live_steering_rejected"}]) { + assert.deepEqual(planNativeMentionIngress({...input, receipt: uncertain}), {action: "unresolved"}); + } + assert.deepEqual(planNativeMentionIngress({...input, receipt, receipt_message_matches: false}), {action: "unresolved"}); + assert.equal(planNativeMentionIngress({...input, receipt: {...receipt, status: "failed", + error_code: "live_steering_turn_mismatch"}}).action, "reject"); + assert.equal(planNativeMentionIngress({...input, receipt: {mode: "session_queue"}}).action, "enqueue"); +}); diff --git a/tests/extensions/test_lark_goal_topic_runtime.py b/tests/extensions/test_lark_goal_topic_runtime.py index 667102e1b6..fcede209d4 100644 --- a/tests/extensions/test_lark_goal_topic_runtime.py +++ b/tests/extensions/test_lark_goal_topic_runtime.py @@ -3594,3 +3594,251 @@ def test_message_event_projection_preserves_flat_cli_and_raw_provider_provenance for key, value in expected.items(): assert row[key] == value assert row["content"] == "plain rendered text" + + +@pytest.mark.parametrize("ordinary_load", [63, 64]) +@pytest.mark.parametrize("manager", [False, True]) +def test_native_mention_reaches_the_running_topic_before_its_ordinary_reply( + tmp_path: Path, manager: bool, ordinary_load: int +) -> None: + """Use the actual listener, Inbox, Chat store, and Codex stdio protocol.""" + import runpy + import stat + import time + + from loopx.chat_manager import manager_channel + from loopx.chat_runtime import ChatRuntimeController + from loopx.chat_store import ChatSessionStore + from loopx.extensions.lark.event_inbox import ingest_lark_event_inbox + from loopx.extensions.lark.goal_topic_runtime import ( + answer_lark_goal_topic, + capture_topic_mention, + process_lark_goal_topic_event, + stream_lark_goal_topic_profile, + ) + from loopx.extensions.lark.manager_context import opaque_digest, session_turn_effect + from loopx.extensions.lark.manager_routing import authorized_manager_goal_ids + + fixture = runpy.run_path(str(Path(__file__).parents[2] / "examples/loopx-chat-runtime-smoke.py")) + release = tmp_path / "release" + executable = tmp_path / "codex" + fake = fixture["FAKE_CODEX"].replace("import json\n", "import json\nfrom pathlib import Path\n", 1) + fake = fake.replace( + " response = 'Visible steered response.", + f" until = time.monotonic() + 10\n" + f" while not Path({str(release)!r}).exists() and time.monotonic() < until:\n" + " time.sleep(0.01)\n" + " response = 'Visible steered response.", + 1, + ) + executable.write_text(fake, encoding="utf-8") + executable.chmod(executable.stat().st_mode | stat.S_IXUSR) + registry = {"goals": [{"id": "goal-alpha", "repo": str(tmp_path), + "coordination": {"registered_agents": ["codex"]}}]} + registry_path = tmp_path / "registry.json" + registry_path.write_text(json.dumps(registry), encoding="utf-8") + store = ChatSessionStore(tmp_path / "chat") + runtime_root = tmp_path / "runtime" + snapshot: dict[str, Any] = {} + controller = ChatRuntimeController( + store=store, codex_bin=str(executable), registry_path=registry_path if manager else None, + manager_scope_resolver=lambda session: authorized_manager_goal_ids( + snapshot, session, runtime_root=tmp_path, registry_path=registry_path, + executor_runtime_root=runtime_root, + ), + ) + session, _ = controller.open_session( + goal_id="goal-alpha", agent_id="codex", work_dir=tmp_path, + objective="Review delivery", mode="resume_latest", + channel_id=manager_channel(provider="lark", audience="mew\0oc_public_fixture") if manager else "goal.goal-alpha", + ) + session_id = session["session_id"] + target_path, binding_path = tmp_path / "targets.json", tmp_path / "binding.json" + state: dict[str, Any] = {} + connected = connect_lark_goal_topic( + registry=registry, + goal_id="goal-alpha", agent_id="codex", session_id=session_id, target_path=target_path, + binding_path=binding_path, app_ref="mew", chat_id="oc_public_fixture", chat_name="Product group", + capture_scope="configured_chat_all", ingress_mode="session_queue", + conversation_kind="manager" if manager else "goal", + turn_trigger="human_messages" if manager else None, + runner=_connection_runner(state), cli_bin="fake-lark", + ) + assert connected["ok"], connected + target_payload = read_goal_channel_targets(target_path) + binding_payload = read_goal_channel_binding(binding_path) + topic = binding_for_goal(binding_payload, "goal-alpha")["topic"]["root_message_id"] + snapshot.update(target_payload=target_payload, binding_payloads={"goal-alpha": binding_payload}) + ordinary = {"event_id": "evt_native_initial", "message_id": "om_native_initial", + "chat_id": "oc_public_fixture", "root_id": topic, "content": "wait for steer", + "sender_id": "ou_fixture_sender", "sender_type": "user", "message_type": "text", "mentions": []} + mention_text = "Add the error case.\nKeep both outcomes." + mention = {**ordinary, "event_id": "evt_native_mention", "message_id": "om_native_mention", + "content": json.dumps({"text": mention_text}), + "mentions": [{"name": "Current bot", "id": "cli_public_fixture"}]} + attachment = {**mention, "event_id": "evt_native_file", "message_id": "om_native_file", + "message_type": "file", "content": json.dumps({"file_key": "file_fixture"})} + wrong_bot = {**mention, "event_id": "evt_other_bot", "message_id": "om_other_bot", + "content": "ordinary input for this connection", + "mentions": [{"name": "Current bot", "id": "cli_other_bot"}]} + backlog = [ + { + **wrong_bot, + "event_id": f"evt_other_bot_{index:02d}", + "message_id": f"om_other_bot_{index:02d}", + "content": f"ordinary backlog {index:02d}", + } + for index in range(ordinary_load - 2) + ] + [wrong_bot] + assert len({ordinary["message_id"], *(item["message_id"] for item in backlog)}) == ordinary_load + idle = {**mention, "event_id": "evt_native_idle", "message_id": "om_native_idle", + "content": json.dumps({"text": "Start normal input."})} + steered_reply = threading.Event() + attachment_reply = threading.Event() + runner = _reply_runner(state) + + def reply(args): + if args[3:6] == ["im", "reactions", "create"]: + return {"returncode": 0, "stdout": json.dumps({"ok": True, "data": {"reaction_id": "reaction_fixture"}})} + if args[3:6] == ["im", "reactions", "delete"]: + return {"returncode": 0, "stdout": json.dumps({"ok": True})} + result = runner(args) + if "+messages-mget" in args: + if state.get("reply_text") == "已送达当前回合。": + steered_reply.set() + if "本轮调整仅支持文字" in state.get("reply_text", ""): + attachment_reply.set() + return result + + observed_ordinary: list[str] = [] + + def answer(route, text): + mid = str(route.get("message_id") or "") + if not route.get("native_mention"): + observed_ordinary.append(mid) + if mid.startswith("om_other_bot_") and mid != "om_other_bot_00": + return { + "response_text": f"Runtime response for {mid}.", + "effect_receipt": session_turn_effect(route), + } + result = answer_lark_goal_topic( + route=route, text=text, work_dir=tmp_path, objective="Review delivery", runtime_controller=controller, + ) + return result if isinstance(result, Mapping) else { + "response_text": result, "effect_receipt": session_turn_effect(route), + } + + initial_id = "lark." + opaque_digest(ordinary["message_id"], topic) + mention_id = "lark." + opaque_digest(mention["message_id"], topic) + wrong_id = "lark." + opaque_digest(wrong_bot["message_id"], topic) + idle_id = "lark." + opaque_digest(idle["message_id"], topic) + total_events = ordinary_load + 5 + + def lines(): + yield json.dumps(ordinary) + "\n" + deadline = time.monotonic() + 8 + while time.monotonic() < deadline: + initial = store.turn_for_client(session_id, initial_id) + if initial and initial.get("upstream_turn_id"): + break + time.sleep(0.02) + else: + raise AssertionError("the initial Topic input never reached the native executor") + try: + for item in backlog: + yield json.dumps(item) + "\n" + yield json.dumps(attachment) + "\n" + assert attachment_reply.wait(5), "the original file needs an explicit retained-message reply" + yield json.dumps(mention) + "\n" + yield json.dumps(mention) + "\n" + assert steered_reply.wait(5), "the listener waited for the ordinary answer before handling the mention" + assert store.load_turn(session_id, initial["turn_id"])["status"] == "running" + assert len([e for e in store.events_after(session_id, initial["turn_id"], None) + if e["kind"] == "turn.steered"]) == 1 + assert store.steering_receipt(session_id, wrong_id) is None + assert store.turn_for_client(session_id, wrong_id) is None + finally: + release.touch() + deadline = time.monotonic() + 15 + while time.monotonic() < deadline: + queued = store.turn_for_client(session_id, wrong_id) + if queued and queued["status"] == "completed" and not store.load_session(session_id).get("active_turn_id"): + break + time.sleep(0.02) + else: + raise AssertionError("the ordinary input did not finish after the active Turn") + yield json.dumps(idle) + "\n" + yield json.dumps(idle) + "\n" + yield f"[event] exited — received {total_events} event(s) in 1s (reason: limit)\n" + + class FinishedConsumer: + stdout = lines() + + def poll(self): + return 0 + + def wait(self, timeout=None): + return 0 + + try: + result = stream_lark_goal_topic_profile( + profile="mew", snapshot_provider=lambda: snapshot, stop=threading.Event(), runtime_root=runtime_root, + answer=answer, process_factory=lambda _: FinishedConsumer(), reply_runner=reply, + provider_runner=lambda args, **kwargs: subprocess.CompletedProcess( + args, 0, json.dumps({"ok": True, "data": {"items": [], "has_more": False}}), "", + ), + capture_mention=lambda route: capture_topic_mention(route, controller), + ) + assert result["ok"] and result["event_count"] == total_events and result["replied_count"] == total_events - 2, result + initial = store.turn_for_client(session_id, initial_id) + assert store.turn_for_client(session_id, mention_id) is None + assert store.turn_for_client(session_id, idle_id)["status"] == "completed" + assert store.steering_receipt(session_id, idle_id)["mode"] == "session_queue" + attachment_id = "lark." + opaque_digest(attachment["message_id"], topic) + assert store.turn_for_client(session_id, attachment_id) is None + receipt = store.steering_receipt(session_id, mention_id) + assert receipt["expected_turn_id"] == receipt["active_turn_id"] == initial["turn_id"] + assert receipt["status"] == "delivered" + user = [msg for msg in store.messages(session_id) if msg["role"] == "user"] + assert [msg["text"] for msg in user if msg["text"] == mention_text] == [mention_text] + steered = next(msg for msg in user if msg["text"] == mention_text) + assert steered["origin"] == "lark" and steered["turn_id"] == initial["turn_id"] + assert observed_ordinary == [ + ordinary["message_id"], + *(item["message_id"] for item in backlog), + ] + ordinary_prompts = [msg["text"] for msg in user if msg["turn_id"] != initial["turn_id"]] + expected_ordinary = [backlog[0]["content"], wrong_bot["content"], idle["content"]] + assert len(ordinary_prompts) == len(expected_ordinary) + for prompt, expected in zip(ordinary_prompts, expected_ordinary, strict=True): + assert prompt.strip().endswith(expected) + assert len([e for e in store.events_after(session_id, initial["turn_id"], None) + if e["kind"] == "turn.steered"]) == 1 + late = {**mention, "event_id": "evt_native_late", "message_id": "om_native_late"} + ended = process_lark_goal_topic_event( + **snapshot, event=late, runtime_root=runtime_root, answer=answer, reply_runner=reply, + mention_capture={ + "connection_id": binding_for_goal(binding_payload, "goal-alpha")["connection_id"], + "session_id": session_id, "binding_matches": True, "target_turn_id": initial["turn_id"], + }, + ) + assert ended["ok"] and "已结束或切换" in state["reply_text"], ended + late_id = "lark." + opaque_digest(late["message_id"], topic) + assert store.turn_for_client(session_id, late_id) is None + assert store.steering_receipt(session_id, late_id)["error_code"] == "live_steering_turn_mismatch" + interrupted = {**mention, "event_id": "evt_native_interrupted", "message_id": "om_native_interrupted"} + stored = ingest_lark_event_inbox( + project=runtime_root, config_path=ended["inbox_config_ref"], execute=True, + events=[{**interrupted, "schema_version": "lark_event_inbox_event_v0"}], + ) + assert stored["accepted_count"] == 1 + recovered = process_lark_goal_topic_event( + **snapshot, event=interrupted, runtime_root=runtime_root, answer=answer, reply_runner=reply, + ) + assert recovered["ok"] and "送达状态待核实" in state["reply_text"], recovered + interrupted_id = "lark." + opaque_digest(interrupted["message_id"], topic) + assert store.turn_for_client(session_id, interrupted_id) is None + assert store.steering_receipt(session_id, interrupted_id) is None + finally: + release.touch() + controller.close_session(session_id) diff --git a/tests/extensions/test_lark_realtime_dispatch.py b/tests/extensions/test_lark_realtime_dispatch.py index f94385b206..6aa2f412c7 100644 --- a/tests/extensions/test_lark_realtime_dispatch.py +++ b/tests/extensions/test_lark_realtime_dispatch.py @@ -586,3 +586,68 @@ def owned_stream(**kwargs): assert health[-1]["status"] == "listening" assert health[-1]["restart_count"] == 1 assert _processed(tmp_path) == 0 + + +def test_profile_without_steerable_binding_keeps_four_ordinary_workers( + tmp_path, monkeypatch +): + snapshot = { + "private_profiles": { + "mew": {"cli_bin": "fake-lark", "provider_ref": "a" * 24} + } + } + all_four = threading.Barrier(4, timeout=3) + release = threading.Event() + entered = [] + stop = threading.Event() + + def admit(_profile, event): + entered.append(event["message_id"]) + all_four.wait() + assert release.wait(5) + if len(entered) == 4: + stop.set() + return {"status": "command_recorded"} + + def lines(): + yield "[event] ready event_key=im.message.receive_v1\n" + for index in range(4): + yield json.dumps( + { + **_event(f"role_{index}", f"topic_{index}"), + "chat_type": "group", + "message_type": "text", + } + ) + assert release.wait(5) + yield "[event] exited (reason: timeout)\n" + + service = LarkGoalTopicRuntimeService( + snapshot_provider=lambda: snapshot, + runtime_root=tmp_path / "runtime", + runtime_controller=object(), + private_conversations=object(), + ) + monkeypatch.setattr(Path, "home", classmethod(lambda cls: tmp_path)) + monkeypatch.setattr(service, "_admit_private", admit) + stream = runtime.stream_lark_goal_topic_profile + + def owned_stream(**kwargs): + assert kwargs.get("capture_mention") is None + result = stream(**kwargs, process_factory=lambda _args: _Consumer(lines())) + service._closed.set() + return result + + monkeypatch.setattr(runtime, "stream_lark_goal_topic_profile", owned_stream) + with ThreadPoolExecutor(max_workers=1) as executor: + future = executor.submit(service._poll_profile, "mew", stop) + try: + deadline = threading.Event() + while len(entered) < 4 and not deadline.wait(0.02): + pass + assert len(entered) == 4 + finally: + release.set() + stop.set() + service._closed.set() + future.result(timeout=6) diff --git a/tests/extensions/test_lark_topic_dispatch_bounds.py b/tests/extensions/test_lark_topic_dispatch_bounds.py index e392bfdaac..b03e10b0d3 100644 --- a/tests/extensions/test_lark_topic_dispatch_bounds.py +++ b/tests/extensions/test_lark_topic_dispatch_bounds.py @@ -133,3 +133,47 @@ def handle(line): release.set() closing.result(timeout=3) assert observed == ["active"] + + +def test_reserved_priority_admits_steering_when_ordinary_capacity_is_full(): + active, release, steered, submitting = [threading.Event() for _ in range(4)] + observed = [] + + def handle(line): + observed.append(line) + if line == "ordinary-0": + active.set() + assert release.wait(5) + elif line == "steer": + steered.set() + + dispatch = ProfileEventDispatch( + handle, threading.Event(), workers=2, capacity=4, reserve_priority=True + ) + try: + assert dispatch.submit(("source",), "ordinary-0") + assert active.wait(5) + assert dispatch.submit(("source",), "ordinary-1") + assert dispatch.submit(("source",), "ordinary-2") + assert dispatch.ordinary_full + + def submit_overflow(): + submitting.set() + return dispatch.submit(("source",), "ordinary-3") + + with ThreadPoolExecutor(max_workers=1) as executor: + overflow = executor.submit(submit_overflow) + try: + assert submitting.wait(5) + with pytest.raises(TimeoutError): + overflow.result(timeout=0.1) + assert dispatch.submit(("steer.source",), "steer", priority=True) + assert steered.wait(3) + assert observed[:2] == ["ordinary-0", "steer"] + finally: + release.set() + assert overflow.result(timeout=3) is True + finally: + release.set() + dispatch.close() + assert observed == ["ordinary-0", "steer", "ordinary-1", "ordinary-2", "ordinary-3"] From 81d05f079c11ac7cb8b80530554dae38e50d9067 Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Mon, 9 Nov 2026 01:24:49 +0800 Subject: [PATCH 2/4] Type the Lark mention dispatch capture boundary Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> (cherry picked from commit 9b2d34d427c104f39d232f91ec0a37272426be07) --- loopx/extensions/lark/goal_topic_runtime.py | 47 ++++++++++++++++----- 1 file changed, 36 insertions(+), 11 deletions(-) diff --git a/loopx/extensions/lark/goal_topic_runtime.py b/loopx/extensions/lark/goal_topic_runtime.py index a57637ec39..3fd7d6309d 100644 --- a/loopx/extensions/lark/goal_topic_runtime.py +++ b/loopx/extensions/lark/goal_topic_runtime.py @@ -13,7 +13,7 @@ from collections.abc import Callable, Mapping from datetime import datetime, timezone from pathlib import Path -from typing import Any +from typing import TYPE_CHECKING, Any, Literal, overload from ...chat_manager import MANAGER_AGENT_OBJECTIVE, manager_channel from ...control_plane.collaboration import conversation_reply_context @@ -85,6 +85,10 @@ start_team_plan_review_callback_stream, ) +if TYPE_CHECKING: + from ...chat_runtime import ChatRuntimeController + + Answer = Callable[[Mapping[str, Any], str], str | Mapping[str, Any]] SnapshotProvider = Callable[[], Mapping[str, Any]] SimpleRunner = Callable[[list[str]], Mapping[str, Any]] @@ -94,6 +98,9 @@ [Mapping[str, Any], list[str]], Mapping[str, Any] ] ReviewCallbackHandler = Callable[[Mapping[str, Any]], Mapping[str, Any]] +MentionCapture = Callable[[Mapping[str, object]], dict[str, object]] +EventLane = tuple[str, ...] +CapturedEventLane = tuple[EventLane, dict[str, object] | None] class LarkGoalTopicTurnFailed(RuntimeError): @@ -395,18 +402,36 @@ def _profile_event_route( ), "matched" +@overload +def _profile_event_lane( + *, profile: str, snapshot: Mapping[str, object], event: Mapping[str, object], + runtime_root: str | Path, capture_mention: MentionCapture | None = None, + include_capture: Literal[False] = False, +) -> EventLane: ... + + +@overload +def _profile_event_lane( + *, profile: str, snapshot: Mapping[str, object], event: Mapping[str, object], + runtime_root: str | Path, capture_mention: MentionCapture | None = None, + include_capture: Literal[True], +) -> CapturedEventLane: ... + + def _profile_event_lane( *, profile: str, snapshot: Mapping[str, Any], - event: Mapping[str, Any], + event: Mapping[str, object], runtime_root: str | Path, - capture_mention: Callable[[Mapping[str, object]], dict[str, object]] | None = None, + capture_mention: MentionCapture | None = None, include_capture: bool = False, -) -> Any: +) -> EventLane | CapturedEventLane: """Capture a live target before joining the ordinary reply lane.""" - def result(lane: tuple[str, ...], capture: dict[str, object] | None = None) -> Any: + def result( + lane: EventLane, capture: dict[str, object] | None = None + ) -> EventLane | CapturedEventLane: return (lane, capture) if include_capture else lane private_config = dict(snapshot.get("private_profiles") or {}).get(profile) @@ -573,7 +598,7 @@ def stream_lark_goal_topic_profile( health_sink: HealthSink | None = None, proposal_deliverer: ProposalDeliverer | None = None, private_admitter: Callable[[str, dict[str, object]], Mapping[str, object]] | None = None, - capture_mention: Callable[[Mapping[str, object]], dict[str, object]] | None = None, + capture_mention: MentionCapture | None = None, review_callback_handler: ReviewCallbackHandler | None = None, ) -> dict[str, Any]: """Keep one bounded long-lived CLI consumer attached between messages.""" @@ -929,9 +954,9 @@ def submit_event(lane: tuple[str, ...], envelope: str, *, priority: bool) -> boo def _profile_supports_mention_steering( - snapshot: Mapping[str, Any], + snapshot: Mapping[str, object], profile: str, - capture_mention: Callable[[Mapping[str, object]], dict[str, object]] | None = None, + capture_mention: MentionCapture | None = None, ) -> bool: """Reserve a priority worker only when a bound Session can steer on @mention.""" @@ -996,8 +1021,8 @@ def _profile_supports_mention_steering( def _bound_topic_session( - route: Mapping[str, object], runtime_controller: Any -) -> Mapping[str, Any] | None: + route: Mapping[str, object], runtime_controller: ChatRuntimeController +) -> Mapping[str, object] | None: session_id = str(route.get("session_id") or "") store = getattr(runtime_controller, "store", None) if not session_id or store is None: @@ -1017,7 +1042,7 @@ def _bound_topic_session( def capture_topic_mention( - route: Mapping[str, object], runtime_controller: Any + route: Mapping[str, object], runtime_controller: ChatRuntimeController ) -> dict[str, object]: """Read the bound Session/Turn without opening or resuming an executor.""" session = _bound_topic_session(route, runtime_controller) From 828df11c2015f98c6355ae9018ef9a1a1c08fe7d Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Mon, 9 Nov 2026 02:41:55 +0800 Subject: [PATCH 3/4] Persist Lark Topic input before waiting on replies Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> --- loopx/extensions/lark/event_inbox.py | 143 +++++++++- loopx/extensions/lark/goal_topic_dispatch.py | 213 ++++++++++++-- loopx/extensions/lark/goal_topic_runtime.py | 262 +++++++++--------- .../test_lark_goal_topic_runtime.py | 200 +++++++++++-- 4 files changed, 635 insertions(+), 183 deletions(-) diff --git a/loopx/extensions/lark/event_inbox.py b/loopx/extensions/lark/event_inbox.py index 453d78e1ae..a0d82e865a 100644 --- a/loopx/extensions/lark/event_inbox.py +++ b/loopx/extensions/lark/event_inbox.py @@ -1,14 +1,16 @@ from __future__ import annotations import hashlib +import heapq import json import os import re -from collections.abc import Mapping, Sequence +from collections.abc import Iterator, Mapping, Sequence from contextlib import nullcontext from datetime import UTC, datetime from pathlib import Path, PurePosixPath -from typing import Any +from time import time_ns +from typing import Any, TypedDict from ...control_plane.work_items.operator_inbox import ( CAPTURE_SCOPES, @@ -597,6 +599,143 @@ def lark_event_mentions_bot( ) + +class LarkTopicInput(TypedDict): + order: int + provider_ref: str + route_ref: str + lane: tuple[str, ...] + event: dict[str, object] + mention_capture: dict[str, object] | None + started: bool + + +class LarkTopicInboxIntake: + """Private live input in the same Topic Inbox, before reply dispatch. + + This keeps the exact provider envelope separate from the compact context + projection. The existing App listener lease owns capture, and its existing + source lane owns dispatch. An ambiguous attempt stays retained for readback. + """ + + def __init__(self, *, project: Path, config_path: Path, provider_ref: str) -> None: + config = load_lark_event_inbox_config(project=project, config_path=config_path) + if not config["enabled"]: + raise ValueError("Lark Topic intake requires an enabled Inbox") + self._inbox: Path = config["inbox_path"] + self._provider_ref = provider_ref + reply = config["reply"] + configured_provider = str(reply["bot_app_id"] or reply["sender_profile"]) + configured_ref = hashlib.sha256(configured_provider.encode("utf-8")).hexdigest()[:32] + self._source_matches = configured_ref == provider_ref + self._directory = self._inbox / ".state" / "topic-intake" + self._last_order = 0 + if self._source_matches: + self._last_order = max( + (packet["order"] for _, packet in self._inputs()), default=0 + ) + + @property + def has_retained_input(self) -> bool: + return self._last_order > 0 + + @staticmethod + def _read(path: Path) -> LarkTopicInput | None: + try: + row = json.loads(path.read_text(encoding="utf-8")) + except FileNotFoundError: + return None + if not isinstance(row, dict): + raise ValueError("invalid Lark Topic intake record") + order = row.get("order") + event = row.get("event") + provider_ref = row.get("provider_ref") + route_ref = row.get("route_ref") + lane = row.get("lane") + mention = row.get("mention_capture") + started = row.get("started") + if ( + type(order) is not int or order <= 0 + or not isinstance(event, dict) or event.get("message_id") != path.stem + or not isinstance(provider_ref, str) or not re.fullmatch(r"[a-f0-9]{32}", provider_ref) + or not isinstance(route_ref, str) or not re.fullmatch(r"[a-f0-9]{32}", route_ref) + or not isinstance(lane, list) or not 1 <= len(lane) <= 2 + or not all(isinstance(key, str) and 0 < len(key) <= 256 for key in lane) + or (mention is not None and not isinstance(mention, dict)) + or type(started) is not bool + ): + raise ValueError("invalid Lark Topic intake record") + return { + "order": order, "event": event, "provider_ref": provider_ref, + "route_ref": route_ref, "lane": tuple(lane), + "mention_capture": mention, "started": started, + } + + def _inputs(self) -> Iterator[tuple[Path, LarkTopicInput]]: + for path in self._directory.glob("*.json"): + if not MESSAGE_ID_PATTERN.fullmatch(path.stem): + continue + packet = self._read(path) + if packet is not None and packet["provider_ref"] == self._provider_ref: + yield path, packet + + def retains(self, message_id: str) -> bool: + if not MESSAGE_ID_PATTERN.fullmatch(message_id): + return False + packet = self._read(self._directory / f"{message_id}.json") + return packet is not None and packet["provider_ref"] == self._provider_ref + + def capture( + self, *, event: Mapping[str, object], route_ref: str, lane: tuple[str, ...], + mention_capture: dict[str, object] | None, + ) -> bool: + if not self._source_matches: + raise ValueError("Lark Topic intake belongs to a different provider") + message_id = str(event.get("message_id") or "") + if not MESSAGE_ID_PATTERN.fullmatch(message_id): + raise ValueError("invalid Lark Topic input message id") + path = self._directory / f"{message_id}.json" + if path.is_file(): + # Provider redelivery cannot overwrite the original body, route or + # pending/ambiguous attempt. The caller still wakes the same lane. + return False + self._last_order = max(self._last_order + 1, time_ns()) + packet: LarkTopicInput = { + "order": self._last_order, "provider_ref": self._provider_ref, + "route_ref": route_ref, "lane": lane, "event": dict(event), + "mention_capture": mention_capture, "started": False, + } + self._directory.mkdir(parents=True, exist_ok=True, mode=0o700) + write_private_json_atomic(path, packet) + return True + + def pending( + self, lane: tuple[str, ...] | None = None, *, limit: int | None = None, + ) -> list[Path]: + # Keep only references in memory while a reply waits on a Turn. + pending = ( + (packet["order"], path) + for path, packet in self._inputs() + if not packet["started"] and (lane is None or packet["lane"] == lane) + ) + ordered = sorted(pending) if limit is None else heapq.nsmallest(limit, pending) + return [path for _, path in ordered] + + def read(self, path: Path) -> LarkTopicInput | None: + packet = self._read(path) + return packet if packet is not None and packet["provider_ref"] == self._provider_ref else None + + def begin(self, path: Path, packet: LarkTopicInput) -> None: + # Persist the attempt fence before any Chat/model/provider effect. + write_private_json_atomic(path, {**packet, "started": True}) + + def finish(self, path: Path) -> None: + # The existing processed receipt is authoritative even after cleanup. + # If reply/ACK was ambiguous, retain the exact source instead of replay. + if path.stem in _load_processed(self._inbox / "processed.json"): + path.unlink(missing_ok=True) + + def ingest_lark_event_inbox( *, project: str | Path, diff --git a/loopx/extensions/lark/goal_topic_dispatch.py b/loopx/extensions/lark/goal_topic_dispatch.py index e3042e7f41..f92a91be59 100644 --- a/loopx/extensions/lark/goal_topic_dispatch.py +++ b/loopx/extensions/lark/goal_topic_dispatch.py @@ -1,9 +1,26 @@ """Bounded transient dispatch for one provider stream, not Chat admission authority.""" +import hashlib +import json +import re from collections import deque -from collections.abc import Callable +from collections.abc import Callable, Mapping from concurrent.futures import ThreadPoolExecutor +from dataclasses import dataclass +from pathlib import Path import threading +from typing import NamedTuple + +from .event_inbox import LarkTopicInboxIntake, LarkTopicInput +from .goal_channel_targets import goal_channel_target_for_name +from .manager_context import opaque_digest + + +class _Work(NamedTuple): + lane: tuple[str, ...] + line: str + priority: bool + durable: bool class ProfileEventDispatch: @@ -11,7 +28,8 @@ class ProfileEventDispatch: A lane is only a transport scheduling key. The handler must read fresh binding/authority and persist through the existing Inbox/Chat owners. - Pending lines are transient, just like the consumer's pipe buffer. + Durable inputs use a coalesced wake for their Inbox source. Only transient + input bodies count against the pipe buffer bound. """ def __init__( @@ -31,7 +49,7 @@ def __init__( self._capacity = capacity self._reserve_priority = reserve_priority and workers > 1 self._condition = threading.Condition() - self._waiting: dict[str, deque[tuple[tuple[str, ...], str, bool]]] = {} + self._waiting: dict[str, deque[_Work]] = {} self._ready: deque[str] = deque() self._active: dict[str, tuple[tuple[str, ...], bool]] = {} self._pending = 0 @@ -53,21 +71,27 @@ def failed(self) -> bool: @property def ordinary_full(self) -> bool: - """Report whether ordinary admission has reached its non-priority bound.""" + """Whether transient ordinary admission has reached its bound.""" with self._condition: return self._full(False) def _full(self, priority: bool) -> bool: + transient = self._ordinary_pending + self._priority_pending if not self._reserve_priority: - return self._pending >= self._capacity + return transient >= self._capacity if priority: return self._priority_pending >= self._workers - return self._ordinary_pending >= self._capacity - 1 or self._pending >= self._capacity + return self._ordinary_pending >= self._capacity - 1 or transient >= self._capacity - def submit(self, lane: tuple[str, ...], line: str, *, priority: bool = False) -> bool: + def submit( + self, lane: tuple[str, ...], line: str, *, priority: bool = False, + durable: bool = False, + ) -> bool: + if durable and priority: + raise ValueError("Inbox source wakes use their ordinary lane") with self._condition: - while self._full(priority) and not self._closing: + while not durable and self._full(priority) and not self._closing: if self._stop.is_set(): return False self._condition.wait(0.1) @@ -77,11 +101,14 @@ def submit(self, lane: tuple[str, ...], line: str, *, priority: bool = False) -> return False source = lane[0] queue = self._waiting.setdefault(source, deque()) - queue.append((lane, line, priority)) + if durable and any(work.durable and work.line == line for work in queue): + return True + work = _Work(lane, line, priority, durable) + queue.append(work) self._pending += 1 if priority: self._priority_pending += 1 - else: + elif not durable: self._ordinary_pending += 1 if source not in self._active and source not in self._ready: if priority: @@ -100,7 +127,8 @@ def _start_ready(self) -> None: if self._cancelled or len(self._active) >= self._workers: break source = self._ready.popleft() - lane, line, priority = self._waiting[source][0] + work = self._waiting[source][0] + lane, priority = work.lane, work.priority active_ordinary = sum( 1 for _lane, is_priority in self._active.values() if not is_priority ) @@ -115,23 +143,26 @@ def _start_ready(self) -> None: continue self._waiting[source].popleft() self._active[source] = (lane, priority) - self._executor.submit(self._run, source, line, priority) + self._executor.submit(self._run, source, work) def _discard_waiting(self) -> None: self._cancelled = True for queue in self._waiting.values(): - for _lane, _line, priority in queue: - self._pending -= 1 - if priority: - self._priority_pending -= 1 - else: - self._ordinary_pending -= 1 + for work in queue: + self._retire(work) self._waiting.clear() self._ready.clear() - def _run(self, source: str, line: str, priority: bool) -> None: + def _retire(self, work: _Work) -> None: + self._pending -= 1 + if work.priority: + self._priority_pending -= 1 + elif not work.durable: + self._ordinary_pending -= 1 + + def _run(self, source: str, work: _Work) -> None: try: - self._handle(line) + self._handle(work.line) except Exception as exc: with self._condition: if self._failure is None: @@ -140,15 +171,11 @@ def _run(self, source: str, line: str, priority: bool) -> None: self._discard_waiting() finally: with self._condition: - self._pending -= 1 - if priority: - self._priority_pending -= 1 - else: - self._ordinary_pending -= 1 + self._retire(work) self._active.pop(source) queue = self._waiting.get(source) if queue: - if queue[0][2]: + if queue[0].priority: self._ready.appendleft(source) else: self._ready.append(source) @@ -158,10 +185,16 @@ def _run(self, source: str, line: str, priority: bool) -> None: self._condition.notify_all() def close(self, *, cancel_pending: bool = False) -> None: - """Retain ownership until active handlers settle; never replay a handler.""" + """Drain started handlers and transient lines, then release the listener.""" with self._condition: self._closing = True + # A new provider connection can wake these captured records. Do + # not hold the old listener open to answer its whole durable backlog. + for source, queue in self._waiting.items(): + self._waiting[source] = deque(work for work in queue if not work.durable) + self._pending -= len(queue) - len(self._waiting[source]) + self._ready = deque(source for source in self._ready if self._waiting[source]) self._condition.notify_all() while self._pending: if cancel_pending or self._stop.is_set(): @@ -170,3 +203,129 @@ def close(self, *, cancel_pending: bool = False) -> None: self._executor.shutdown(wait=True) if self._failure is not None: raise self._failure + + +@dataclass(frozen=True) +class TopicIntakeRoute: + route: Mapping[str, object] + targets: Mapping[str, object] + digest: str + inbox_key: str + + +def topic_inbox_key(profile: str, chat_id: str, topic_root: str) -> str: + return hashlib.sha256(f"{profile}\0{chat_id}\0{topic_root}".encode("utf-8")).hexdigest()[:20] + + +def topic_intake_route( + profile: str, route: Mapping[str, object], targets: Mapping[str, object], + event: Mapping[str, object], +) -> TopicIntakeRoute | None: + if route.get("ingress_mode") != "session_queue" or not route.get("session_id"): + return None + target = goal_channel_target_for_name(targets, str(route.get("target_ref") or "")) + identity = target.get("identity") if target is not None else None + if not isinstance(identity, Mapping): + return None + chat_id = str(event.get("chat_id") or "") + key = topic_inbox_key( + str(route.get("app_ref") or ""), chat_id, str(route.get("topic_root_message_id") or "") + ) + # Freeze the existing typed routing decision and the provider's bound Bot. + # Capture is not authority to execute against a changed binding. + digest = opaque_digest( + profile, chat_id, identity.get("bot_app_id"), identity.get("bot_open_id"), + json.dumps(dict(route), sort_keys=True), + ) + return TopicIntakeRoute(route=route, targets=targets, digest=digest, inbox_key=key) + + +class ProfileTopicIntake: + """Turn durable Inbox inputs into coalesced wakes on the existing lanes.""" + + def __init__(self, root: Path, provider_ref: str) -> None: + self._root = root + self._provider_ref = provider_ref + self._sources: dict[str, LarkTopicInboxIntake] = {} + self._pages: dict[tuple[str, tuple[str, ...]], deque[Path]] = {} + + def _open(self, config_path: Path) -> LarkTopicInboxIntake: + return LarkTopicInboxIntake( + project=self._root, config_path=config_path, provider_ref=self._provider_ref, + ) + + @staticmethod + def _offer(dispatch: ProfileEventDispatch, inbox_key: str, lane: tuple[str, ...]) -> None: + dispatch.submit(lane, json.dumps({"intake": inbox_key, "lane": lane}), durable=True) + + def retains(self, message_id: str) -> bool: + return any(inbox.retains(message_id) for inbox in self._sources.values()) + + def capture( + self, dispatch: ProfileEventDispatch, route: TopicIntakeRoute, lane: tuple[str, ...], + event: Mapping[str, object], mention: dict[str, object] | None, + config_path: Callable[[], Path], + ) -> bool: + key = route.inbox_key + if key not in self._sources: + self._sources[key] = self._open(config_path()) + captured = self._sources[key].capture( + event=event, route_ref=route.digest, lane=lane, mention_capture=mention, + ) + self._offer(dispatch, key, lane) + return captured + + def dispatch_next( + self, dispatch: ProfileEventDispatch, inbox_key: str, lane: tuple[str, ...], + stop: threading.Event, prepare: Callable[[str, LarkTopicInput], Callable[[], None]], + ) -> None: + if stop.is_set(): + return + inbox = self._sources[inbox_key] + key = (inbox_key, lane) + page = self._pages.get(key) + if page is None: + page = deque(inbox.pending(lane, limit=32)) + if not page: + return + self._pages[key] = page + path = page.popleft() + packet = inbox.read(path) + if packet is not None and not packet["started"] and packet["lane"] == lane: + run = prepare(inbox_key, packet) + if stop.is_set(): + return + inbox.begin(path, packet) + run() + inbox.finish(path) + if not page: + self._pages.pop(key) + # Check for the next page or a new capture after giving other sources + # sharing this Session their usual place in the dispatcher. + self._offer(dispatch, inbox_key, lane) + + def restore( + self, dispatch: ProfileEventDispatch, + matches: Callable[[str, LarkTopicInput], bool] | None, + ) -> None: + config_dir = self._root / ".loopx" / "config" / "lark-goal-topics" + for config_path in config_dir.glob("*.json"): + key = config_path.stem + if not re.fullmatch(r"[a-f0-9]{20}", key): + continue + state_dir = self._root / ".loopx" / "inbox" / "lark-goal-topics" / key / ".state" / "topic-intake" + if not state_dir.is_dir(): + continue + inbox = self._open(config_path) + if not inbox.has_retained_input: + continue + self._sources[key] = inbox + if matches is None: + continue + restored = set() + for path in inbox.pending(): + packet = inbox.read(path) + if packet is None or packet["lane"] in restored or not matches(key, packet): + continue + self._offer(dispatch, key, packet["lane"]) + restored.add(packet["lane"]) diff --git a/loopx/extensions/lark/goal_topic_runtime.py b/loopx/extensions/lark/goal_topic_runtime.py index 3fd7d6309d..bae603bdd9 100644 --- a/loopx/extensions/lark/goal_topic_runtime.py +++ b/loopx/extensions/lark/goal_topic_runtime.py @@ -9,7 +9,6 @@ import re import subprocess import threading -from collections import deque from collections.abc import Callable, Mapping from datetime import datetime, timezone from pathlib import Path @@ -35,6 +34,7 @@ ) from .event_collector_runtime import EVENT_DIAGNOSTIC_PREFIX, EVENT_READY_PREFIX from .event_inbox import ( + LarkTopicInput, MESSAGE_ID_PATTERN, acknowledge_lark_event_inbox, ingest_lark_event_inbox, @@ -48,7 +48,13 @@ from .goal_channel_targets import goal_channel_target_for_name from .goal_channel_transport import lark_provider_mention_identities, record_event_timing from .goal_topic_connections import decide_lark_topic_event -from .goal_topic_dispatch import ProfileEventDispatch +from .goal_topic_dispatch import ( + ProfileEventDispatch, + ProfileTopicIntake, + TopicIntakeRoute, + topic_inbox_key, + topic_intake_route, +) from .inbox_reply import CommandRunner, reply_lark_event_inbox from .manager_reply_delivery import ( load_delivery as _load_manager_delivery, @@ -100,7 +106,7 @@ ReviewCallbackHandler = Callable[[Mapping[str, Any]], Mapping[str, Any]] MentionCapture = Callable[[Mapping[str, object]], dict[str, object]] EventLane = tuple[str, ...] -CapturedEventLane = tuple[EventLane, dict[str, object] | None] +CapturedEventLane = tuple[EventLane, dict[str, object] | None, TopicIntakeRoute | None] class LarkGoalTopicTurnFailed(RuntimeError): @@ -406,7 +412,7 @@ def _profile_event_route( def _profile_event_lane( *, profile: str, snapshot: Mapping[str, object], event: Mapping[str, object], runtime_root: str | Path, capture_mention: MentionCapture | None = None, - include_capture: Literal[False] = False, + include_capture: Literal[False] = False, prepare_intake: bool = False, ) -> EventLane: ... @@ -414,7 +420,7 @@ def _profile_event_lane( def _profile_event_lane( *, profile: str, snapshot: Mapping[str, object], event: Mapping[str, object], runtime_root: str | Path, capture_mention: MentionCapture | None = None, - include_capture: Literal[True], + include_capture: Literal[True], prepare_intake: bool = False, ) -> CapturedEventLane: ... @@ -426,13 +432,15 @@ def _profile_event_lane( runtime_root: str | Path, capture_mention: MentionCapture | None = None, include_capture: bool = False, + prepare_intake: bool = False, ) -> EventLane | CapturedEventLane: """Capture a live target before joining the ordinary reply lane.""" def result( - lane: EventLane, capture: dict[str, object] | None = None + lane: EventLane, capture: dict[str, object] | None = None, + intake: TopicIntakeRoute | None = None, ) -> EventLane | CapturedEventLane: - return (lane, capture) if include_capture else lane + return (lane, capture, intake) if include_capture else lane private_config = dict(snapshot.get("private_profiles") or {}).get(profile) if isinstance(private_config, Mapping) and _native_profile_event(snapshot, profile, event): @@ -479,7 +487,8 @@ def result( if capture and capture.get("target_turn_id"): return result(("steer." + source, "steer.session." + str(route["session_id"])), capture) if route.get("session_id"): - return result((source, "session." + str(route["session_id"])), capture) + intake = topic_intake_route(profile, route, targets, routed_event) if prepare_intake else None + return result((source, "session." + str(route["session_id"])), capture, intake) return result((source,), capture) return result(("unmatched." + _opaque_digest(profile, event.get("chat_id")),)) @@ -707,14 +716,50 @@ def stop_consumer() -> None: exit_reason: str | None = None result_lock = threading.Lock() - def handle_event(line: str) -> None: - nonlocal event_count, replied_count, provider_ready + def note_provider_ready() -> None: + nonlocal provider_ready - envelope = json.loads(line) - event = envelope["event"] + with result_lock: + if provider_ready: + return + provider_ready = True + if health_sink is not None and not dispatch.failed: + health_sink({"status": "listening", "error_code": None}) + + def report_result(result: Mapping[str, Any]) -> None: + nonlocal event_count, replied_count + + events = int(result.get("event_count") or 0) + if not events: + return + replies = int(result.get("replied_count") or 0) + statuses = list(result.get("event_statuses") or []) + reasons = list(result.get("event_reasons") or []) + with result_lock: + event_count += events + replied_count += replies + if health_sink is not None: + failed = dispatch.failed + health_sink({ + "status": "failed" if failed else "listening" if provider_ready else "starting", + "error_code": "lark_event_listener_failed" if failed else None, + "event_count": events, + "replied_count": replies, + "last_event_status": str(statuses[-1]) if statuses else None, + "last_event_reason": str(reasons[-1]) if reasons and reasons[-1] else None, + }) + print(json.dumps({ + "schema_version": "loopx_lark_goal_topic_runtime_event_v0", + "event_count": events, "replied_count": replies, + "event_statuses": statuses, "event_reasons": reasons, + }, separators=(",", ":")), flush=True) + + def dispatch_event( + event: Mapping[str, object], mention: Mapping[str, object] | None, + current_snapshot: Mapping[str, object], + ) -> None: message_id = str(event.get("message_id") or "") record_event_timing(profile, message_id, "dispatch_started") - current_snapshot = snapshot_provider() record_event_timing(profile, message_id, "dispatch_snapshot_ready") result = poll_lark_goal_topic_profile_once( profile=profile, @@ -730,116 +775,59 @@ def handle_event(line: str) -> None: reply_runner=reply_runner, proposal_deliverer=proposal_deliverer, private_admitter=private_admitter, - mention_capture=envelope["mention_capture"], + mention_capture=mention, ) record_event_timing(profile, message_id, "dispatch_returned") - with result_lock: - failed = dispatch.failed - if int(result.get("event_count") or 0) and not provider_ready: - # A provider event is stronger readiness evidence than a - # diagnostic marker and protects compatibility with providers - # that omit the marker while still emitting the typed stream. - provider_ready = True - if health_sink is not None and not failed: - health_sink({"status": "listening", "error_code": None}) - event_count += int(result.get("event_count") or 0) - replied_count += int(result.get("replied_count") or 0) - if health_sink is not None and int(result.get("event_count") or 0): - statuses = list(result.get("event_statuses") or []) - reasons = list(result.get("event_reasons") or []) - health_sink( - { - "status": "failed" if failed else "listening", - "error_code": ( - "lark_event_listener_failed" if failed else None - ), - "event_count": int(result.get("event_count") or 0), - "replied_count": int(result.get("replied_count") or 0), - "last_event_status": str(statuses[-1]) if statuses else None, - "last_event_reason": ( - str(reasons[-1]) if reasons and reasons[-1] else None - ), - } - ) - if int(result.get("event_count") or 0): - print( - json.dumps( - { - "schema_version": "loopx_lark_goal_topic_runtime_event_v0", - "event_count": int(result.get("event_count") or 0), - "replied_count": int(result.get("replied_count") or 0), - "event_statuses": list(result.get("event_statuses") or []), - "event_reasons": list(result.get("event_reasons") or []), - }, - separators=(",", ":"), - ), - flush=True, - ) + report_result(result) reserve_priority = capture_mention is not None and _profile_supports_mention_steering( snapshot, profile, capture_mention ) + root = Path(runtime_root).expanduser().resolve() + provider_ref = _opaque_digest(profile_config.get("bot_app_id") or profile) + intake = ProfileTopicIntake(root, provider_ref) + + def report_retained_input() -> None: + report_result({ + "event_count": 1, "replied_count": 0, + "event_statuses": ["already_captured"], "event_reasons": [None], + }) + + def matches_input(inbox_key: str, packet: LarkTopicInput, snapshot: Mapping[str, object]) -> bool: + lane, _, route = _profile_event_lane( + profile=profile, snapshot=snapshot, event=packet["event"], + runtime_root=root, include_capture=True, prepare_intake=True, + ) + return ( + route is not None and route.inbox_key == inbox_key + and route.digest == packet["route_ref"] and lane == packet["lane"] + ) + + def prepare_input(inbox_key: str, packet: LarkTopicInput) -> Callable[[], None]: + fresh = snapshot_provider() + if matches_input(inbox_key, packet, fresh): + return lambda: dispatch_event(packet["event"], packet["mention_capture"], fresh) + return lambda: report_result({ + "event_count": 1, "replied_count": 0, "event_statuses": ["ignored"], + "event_reasons": [LarkTopicEventDecisionReason.BINDING_UNAVAILABLE.value], + }) + + def handle_event(line: str) -> None: + envelope = json.loads(line) + if "intake" in envelope: + intake.dispatch_next(dispatch, envelope["intake"], tuple(envelope["lane"]), stop, prepare_input) + else: + dispatch_event(envelope["event"], envelope["mention_capture"], snapshot_provider()) + dispatch = ProfileEventDispatch(handle_event, stop, reserve_priority=reserve_priority) - ordinary_backlog: deque[tuple[tuple[str, ...], str]] = deque() - backlog_condition = threading.Condition() - backlog_closed = False - backlog_worker: threading.Thread | None = None - - def drain_ordinary_backlog() -> None: - while True: - with backlog_condition: - while not ordinary_backlog and not backlog_closed: - if stop.is_set() or dispatch.failed: - return - backlog_condition.wait(0.05) - if backlog_closed or stop.is_set() or dispatch.failed: - ordinary_backlog.clear() - backlog_condition.notify_all() - return - lane, envelope = ordinary_backlog[0] - try: - submitted = dispatch.submit(lane, envelope, priority=False) - except Exception: - with backlog_condition: - ordinary_backlog.clear() - backlog_condition.notify_all() - return - with backlog_condition: - if ordinary_backlog: - ordinary_backlog.popleft() - if not submitted: - ordinary_backlog.clear() - backlog_condition.notify_all() - if not submitted: - return - - def submit_event(lane: tuple[str, ...], envelope: str, *, priority: bool) -> bool: - nonlocal backlog_worker - if priority or not reserve_priority: - return dispatch.submit(lane, envelope, priority=priority) - with backlog_condition: - if not ordinary_backlog and not dispatch.ordinary_full: - return dispatch.submit(lane, envelope, priority=False) - while len(ordinary_backlog) >= 64 and not backlog_closed: - if stop.is_set() or dispatch.failed: - return False - backlog_condition.wait(0.05) - if backlog_closed or stop.is_set() or dispatch.failed: - return False - ordinary_backlog.append((lane, envelope)) - if backlog_worker is None: - backlog_worker = threading.Thread( - target=drain_ordinary_backlog, - name=f"loopx-lark-ordinary-{profile}", - daemon=True, - ) - backlog_worker.start() - backlog_condition.notify_all() - return True watcher.start() stream_failed = True try: + intake.restore( + dispatch, + (lambda key, packet: matches_input(key, packet, snapshot)) if reserve_priority else None, + ) stdout = process.stdout if stdout is None: return { @@ -853,10 +841,7 @@ def submit_event(lane: tuple[str, ...], envelope: str, *, priority: bool) -> boo break stripped = line.strip() if stripped.startswith(EVENT_READY_PREFIX): - with result_lock: - provider_ready = True - if health_sink is not None and not dispatch.failed: - health_sink({"status": "listening", "error_code": None}) + note_provider_ready() continue if stripped.startswith("[event] exited "): match = _EVENT_EXIT_REASON.search(stripped) @@ -867,37 +852,46 @@ def submit_event(lane: tuple[str, ...], envelope: str, *, priority: bool) -> boo # Batch-shaped provider records are split before scheduling, so an # unrelated conversation in a batch can progress independently too. for event in _event_payloads(line): + # Recovered local input alone does not attest provider readiness. + note_provider_ready() message_id = str(event.get("message_id") or "") record_event_timing(profile, message_id, "received") - lane, mention_capture = _profile_event_lane( + if intake.retains(message_id): + # A provider retry is the same source input even if the + # App or Session is now routed to a different Goal Topic. + report_retained_input() + continue + lane, mention_capture, intake_route = _profile_event_lane( profile=profile, snapshot=snapshot_provider(), event=event, runtime_root=runtime_root, capture_mention=capture_mention, include_capture=True, + prepare_intake=reserve_priority, ) record_event_timing(profile, message_id, "lane_ready") - envelope = json.dumps({"event": event, "mention_capture": mention_capture}, ensure_ascii=False) - if not submit_event(lane, envelope, priority=bool((mention_capture or {}).get("target_turn_id"))): - break - while True: - with backlog_condition: - if not ordinary_backlog or stop.is_set() or dispatch.failed: - break - backlog_condition.wait(0.05) + priority = bool((mention_capture or {}).get("target_turn_id")) + if reserve_priority and intake_route is not None and not priority: + if intake.capture( + dispatch, intake_route, lane, event, mention_capture, + lambda: _inbox_config( + runtime_root=root, route=intake_route.route, + target_payload=intake_route.targets, + )[0], + ): + record_event_timing(profile, message_id, "inbox_captured") + else: + report_retained_input() + else: + envelope = json.dumps({"event": event, "mention_capture": mention_capture}, ensure_ascii=False) + if not dispatch.submit(lane, envelope, priority=priority): + break stream_failed = False finally: - with backlog_condition: - backlog_closed = True - if stream_failed or stop.is_set() or dispatch.failed: - ordinary_backlog.clear() - backlog_condition.notify_all() try: dispatch.close(cancel_pending=stream_failed) finally: - if backlog_worker is not None: - backlog_worker.join(timeout=1) watcher_done.set() if process.poll() is None: process.terminate() @@ -1303,9 +1297,7 @@ def _inbox_config( chat_id = str(channel.get("chat_id") or "") profile = str(route.get("app_ref") or "") bot_display_name = str(identity.get("bot_display_name") or profile) - digest = hashlib.sha256( - f"{profile}\0{chat_id}\0{route.get('topic_root_message_id')}".encode("utf-8") - ).hexdigest()[:20] + digest = topic_inbox_key(profile, chat_id, str(route.get("topic_root_message_id") or "")) config_ref = f".loopx/config/lark-goal-topics/{digest}.json" config_path = runtime_root / config_ref payload = { diff --git a/tests/extensions/test_lark_goal_topic_runtime.py b/tests/extensions/test_lark_goal_topic_runtime.py index fcede209d4..516e11d786 100644 --- a/tests/extensions/test_lark_goal_topic_runtime.py +++ b/tests/extensions/test_lark_goal_topic_runtime.py @@ -131,6 +131,7 @@ def run(args: list[str], _cwd: object, _timeout: object) -> dict[str, Any]: "available": True, "verified": True, "appName": "linkmacbot", + **({"openId": state["bot_open_id"]} if "bot_open_id" in state else {}), } }, } @@ -3596,7 +3597,7 @@ def test_message_event_projection_preserves_flat_cli_and_raw_provider_provenance assert row["content"] == "plain rendered text" -@pytest.mark.parametrize("ordinary_load", [63, 64]) +@pytest.mark.parametrize("ordinary_load", [63, 64, 128]) @pytest.mark.parametrize("manager", [False, True]) def test_native_mention_reaches_the_running_topic_before_its_ordinary_reply( tmp_path: Path, manager: bool, ordinary_load: int @@ -3654,7 +3655,7 @@ def test_native_mention_reaches_the_running_topic_before_its_ordinary_reply( ) session_id = session["session_id"] target_path, binding_path = tmp_path / "targets.json", tmp_path / "binding.json" - state: dict[str, Any] = {} + state: dict[str, Any] = {"bot_open_id": "ou_fixture_bot"} connected = connect_lark_goal_topic( registry=registry, goal_id="goal-alpha", agent_id="codex", session_id=session_id, target_path=target_path, @@ -3671,30 +3672,32 @@ def test_native_mention_reaches_the_running_topic_before_its_ordinary_reply( snapshot.update(target_payload=target_payload, binding_payloads={"goal-alpha": binding_payload}) ordinary = {"event_id": "evt_native_initial", "message_id": "om_native_initial", "chat_id": "oc_public_fixture", "root_id": topic, "content": "wait for steer", - "sender_id": "ou_fixture_sender", "sender_type": "user", "message_type": "text", "mentions": []} - mention_text = "Add the error case.\nKeep both outcomes." + "sender_id": "ou_fixture_sender", "sender_type": "user", "message_type": "text", "mentions": [], + "create_time": "1794140000000"} + mention_text = "@Current bot Add the error case.\nKeep both outcomes." mention = {**ordinary, "event_id": "evt_native_mention", "message_id": "om_native_mention", - "content": json.dumps({"text": mention_text}), - "mentions": [{"name": "Current bot", "id": "cli_public_fixture"}]} + "content": mention_text, + "mentions": [{"key": "@_user_1", "name": "Current bot", "id": "ou_fixture_bot"}]} attachment = {**mention, "event_id": "evt_native_file", "message_id": "om_native_file", "message_type": "file", "content": json.dumps({"file_key": "file_fixture"})} wrong_bot = {**mention, "event_id": "evt_other_bot", "message_id": "om_other_bot", "content": "ordinary input for this connection", - "mentions": [{"name": "Current bot", "id": "cli_other_bot"}]} + "mentions": [{"key": "@_user_1", "name": "Current bot", "id": "ou_other_bot"}]} backlog = [ { **wrong_bot, - "event_id": f"evt_other_bot_{index:02d}", - "message_id": f"om_other_bot_{index:02d}", - "content": f"ordinary backlog {index:02d}", + "event_id": f"evt_other_bot_{index:03d}", + "message_id": f"om_other_bot_{index:03d}", + "content": f"ordinary backlog {index:03d}\n" + "retained text " * 100 + "\nlast line", } - for index in range(ordinary_load - 2) + for index in reversed(range(ordinary_load - 2)) ] + [wrong_bot] assert len({ordinary["message_id"], *(item["message_id"] for item in backlog)}) == ordinary_load idle = {**mention, "event_id": "evt_native_idle", "message_id": "om_native_idle", - "content": json.dumps({"text": "Start normal input."})} + "content": "@Current bot Start normal input."} steered_reply = threading.Event() attachment_reply = threading.Event() + last_ordinary_answer = threading.Event() runner = _reply_runner(state) def reply(args): @@ -3716,7 +3719,7 @@ def answer(route, text): mid = str(route.get("message_id") or "") if not route.get("native_mention"): observed_ordinary.append(mid) - if mid.startswith("om_other_bot_") and mid != "om_other_bot_00": + if mid.startswith("om_other_bot_") and mid != backlog[0]["message_id"]: return { "response_text": f"Runtime response for {mid}.", "effect_receipt": session_turn_effect(route), @@ -3724,6 +3727,8 @@ def answer(route, text): result = answer_lark_goal_topic( route=route, text=text, work_dir=tmp_path, objective="Review delivery", runtime_controller=controller, ) + if mid == wrong_bot["message_id"]: + last_ordinary_answer.set() return result if isinstance(result, Mapping) else { "response_text": result, "effect_receipt": session_turn_effect(route), } @@ -3744,9 +3749,17 @@ def lines(): time.sleep(0.02) else: raise AssertionError("the initial Topic input never reached the native executor") + def interrupt_fixture_if_blocked(): + if not steered_reply.is_set(): + controller.interrupt_turn(session_id=session_id, turn_id=initial["turn_id"]) + + watchdog = threading.Timer(8, interrupt_fixture_if_blocked) + watchdog.daemon = True + watchdog.start() try: for item in backlog: yield json.dumps(item) + "\n" + assert store.load_turn(session_id, initial["turn_id"])["status"] == "running" yield json.dumps(attachment) + "\n" assert attachment_reply.wait(5), "the original file needs an explicit retained-message reply" yield json.dumps(mention) + "\n" @@ -3758,17 +3771,25 @@ def lines(): assert store.steering_receipt(session_id, wrong_id) is None assert store.turn_for_client(session_id, wrong_id) is None finally: + watchdog.cancel() release.touch() - deadline = time.monotonic() + 15 + assert last_ordinary_answer.wait(40), "the FIFO backlog did not reach its last ordinary input" + queued = store.turn_for_client(session_id, wrong_id) + assert queued and queued["status"] == "completed" + deadline = time.monotonic() + 3 + while store.load_session(session_id).get("active_turn_id"): + assert time.monotonic() < deadline, "the completed Turn did not release the Session" + time.sleep(0.02) + yield json.dumps(idle) + "\n" + yield json.dumps(idle) + "\n" + deadline = time.monotonic() + 6 while time.monotonic() < deadline: - queued = store.turn_for_client(session_id, wrong_id) - if queued and queued["status"] == "completed" and not store.load_session(session_id).get("active_turn_id"): + receipt = store.steering_receipt(session_id, idle_id) + if receipt and receipt["status"] == "delivered": break time.sleep(0.02) else: - raise AssertionError("the ordinary input did not finish after the active Turn") - yield json.dumps(idle) + "\n" - yield json.dumps(idle) + "\n" + raise AssertionError("the idle mention did not enter the same Session queue") yield f"[event] exited — received {total_events} event(s) in 1s (reason: limit)\n" class FinishedConsumer: @@ -3842,3 +3863,144 @@ def wait(self, timeout=None): finally: release.touch() controller.close_session(session_id) + + +@pytest.mark.parametrize("binding_change", ["unchanged", "session", "topic"]) +def test_topic_stream_recovers_only_unstarted_input_with_its_original_binding( + tmp_path: Path, binding_change: str, +) -> None: + from loopx.extensions.lark.goal_topic_runtime import stream_lark_goal_topic_profile + from loopx.extensions.lark.manager_context import session_turn_effect + + runtime_root = tmp_path / "runtime" + registry = {"goals": [{"id": "goal-alpha", "repo": str(tmp_path), + "coordination": {"registered_agents": ["codex"]}}]} + targets_path = tmp_path / "targets.json" + binding_path = tmp_path / "binding.json" + state: dict[str, Any] = {} + assert connect_lark_goal_topic( + registry=registry, goal_id="goal-alpha", agent_id="codex", + session_id="session-original", target_path=targets_path, binding_path=binding_path, + app_ref="mew", chat_id="oc_public_fixture", chat_name="Product group", + capture_scope="configured_chat_all", ingress_mode="session_queue", + runner=_connection_runner(state), cli_bin="fake-lark", + )["ok"] + targets = read_goal_channel_targets(targets_path) + bindings = read_goal_channel_binding(binding_path) + topic = binding_for_goal(bindings, "goal-alpha")["topic"]["root_message_id"] + snapshot = {"target_payload": targets, "binding_payloads": {"goal-alpha": bindings}} + original = { + "event_id": "evt_started", "message_id": "om_started", + "chat_id": "oc_public_fixture", "root_id": topic, "content": "Original request.", + "create_time": "1794140000000", "sender_type": "user", + "sender_id": "ou_fixture_sender", "message_type": "text", "mentions": [], + } + later = [ + {**original, "event_id": f"evt_{suffix}", "message_id": f"om_{suffix}", + "content": f"{suffix}\n" + "Complete source text. " * 100 + "\nLast line."} + for suffix in ("second", "first_by_id") + ] + entered = threading.Event() + release = threading.Event() + stop = threading.Event() + last_reply = threading.Event() + answered = [] + + def answer(route, text): + answered.append((route["session_id"], route["message_id"], text)) + if route["message_id"] == original["message_id"]: + entered.set() + assert release.wait(5) + return {"response_text": "Verified reply.", "effect_receipt": session_turn_effect(route)} + + def first_consumer(): + yield "[event] ready event_key=im.message.receive_v1\n" + yield json.dumps(original) + "\n" + try: + assert entered.wait(5) + for row in later: + yield json.dumps(row) + "\n" + finally: + stop.set() + release.set() + + runner = _reply_runner(state) + first_writes = [] + + def unverified_reply(args): + if "+messages-reply" in args and "--dry-run" not in args: + first_writes.append(args[args.index("--message-id") + 1]) + return {"returncode": 0, "stdout": '{"data": {}}', "stderr": ""} + return runner(args) + + class FinishedConsumer: + def __init__(self, stdout): + self.stdout = stdout + + def poll(self): + return 0 + + def wait(self, timeout=None): + return 0 + + def consume(stdout, stop, reply): + return stream_lark_goal_topic_profile( + profile="mew", snapshot_provider=lambda: snapshot, runtime_root=runtime_root, + stop=stop, answer=answer, process_factory=lambda _: FinishedConsumer(stdout), + reply_runner=reply, capture_mention=lambda route: { + "connection_id": route["connection_id"], "session_id": route["session_id"], + "binding_matches": True, "target_turn_id": None, + }, + ) + + stopped = consume(first_consumer(), stop, unverified_reply) + assert stopped["status"] == "stopped" + assert answered == [("session-original", original["message_id"], original["content"])] + assert first_writes == [original["message_id"]] + + captures = { + path.stem: json.loads(path.read_text(encoding="utf-8")) + for path in runtime_root.rglob(".state/topic-intake/om_*.json") + } + assert captures[original["message_id"]]["started"] is True + for row in later: + assert captures[row["message_id"]]["started"] is False + assert captures[row["message_id"]]["event"] == row + + rebind = binding_change != "unchanged" + connection, = bindings["bindings"]["goal-alpha"]["connections"].values() + if binding_change == "session": + connection["session_id"] = "session-replacement" + elif binding_change == "topic": + connection["topic"]["root_message_id"] = "om_topic_replacement" + + def recovered_reply(args): + result = runner(args) + if "+messages-reply" in args and "--dry-run" not in args: + if args[args.index("--message-id") + 1] == later[-1]["message_id"]: + last_reply.set() + return result + + def replacement_consumer(): + yield "[event] ready event_key=im.message.receive_v1\n" + yield json.dumps({**original, "content": "Redelivered data must not replace the original."}) + "\n" + yield json.dumps(later[0]) + "\n" + if not rebind: + assert last_reply.wait(5), "the original Inbox should recover without provider redelivery" + yield "[event] exited (reason: timeout)\n" + + resumed = consume(replacement_consumer(), threading.Event(), recovered_reply) + assert resumed["status"] == "stream_ended" + expected = [] if rebind else [ + ("session-original", row["message_id"], row["content"]) for row in later + ] + assert answered[1:] == expected + replies = [args[args.index("--message-id") + 1] for args in state["calls"] + if "+messages-reply" in args and "--dry-run" not in args] + assert replies == ([] if rebind else [row["message_id"] for row in later]) + retained = list(runtime_root.rglob(".state/topic-intake/om_*.json")) + assert {path.stem for path in retained} == ( + set(captures) if rebind else {original["message_id"]} + ) + original_file = next(path for path in retained if path.stem == original["message_id"]) + assert json.loads(original_file.read_text(encoding="utf-8")) == captures[original["message_id"]] From a20cd9e484617628c96a720b2bd4dabc0420c3a7 Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Mon, 9 Nov 2026 05:50:06 +0800 Subject: [PATCH 4/4] Keep the Lark reader live under steering backlog Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> --- loopx/extensions/lark/event_inbox.py | 13 +- loopx/extensions/lark/goal_topic_dispatch.py | 74 +++++++++- loopx/extensions/lark/goal_topic_runtime.py | 136 +++++++++--------- .../test_lark_goal_topic_runtime.py | 116 +++++++++++++-- 4 files changed, 251 insertions(+), 88 deletions(-) diff --git a/loopx/extensions/lark/event_inbox.py b/loopx/extensions/lark/event_inbox.py index a0d82e865a..41f459ea76 100644 --- a/loopx/extensions/lark/event_inbox.py +++ b/loopx/extensions/lark/event_inbox.py @@ -10,7 +10,7 @@ from datetime import UTC, datetime from pathlib import Path, PurePosixPath from time import time_ns -from typing import Any, TypedDict +from typing import Any, NotRequired, TypedDict from ...control_plane.work_items.operator_inbox import ( CAPTURE_SCOPES, @@ -607,6 +607,7 @@ class LarkTopicInput(TypedDict): lane: tuple[str, ...] event: dict[str, object] mention_capture: dict[str, object] | None + live_turn_capture: NotRequired[dict[str, object]] started: bool @@ -653,6 +654,7 @@ def _read(path: Path) -> LarkTopicInput | None: route_ref = row.get("route_ref") lane = row.get("lane") mention = row.get("mention_capture") + live_turn = row.get("live_turn_capture") started = row.get("started") if ( type(order) is not int or order <= 0 @@ -662,14 +664,18 @@ def _read(path: Path) -> LarkTopicInput | None: or not isinstance(lane, list) or not 1 <= len(lane) <= 2 or not all(isinstance(key, str) and 0 < len(key) <= 256 for key in lane) or (mention is not None and not isinstance(mention, dict)) + or (live_turn is not None and not isinstance(live_turn, dict)) or type(started) is not bool ): raise ValueError("invalid Lark Topic intake record") - return { + packet: LarkTopicInput = { "order": order, "event": event, "provider_ref": provider_ref, "route_ref": route_ref, "lane": tuple(lane), "mention_capture": mention, "started": started, } + if live_turn is not None: + packet["live_turn_capture"] = live_turn + return packet def _inputs(self) -> Iterator[tuple[Path, LarkTopicInput]]: for path in self._directory.glob("*.json"): @@ -688,6 +694,7 @@ def retains(self, message_id: str) -> bool: def capture( self, *, event: Mapping[str, object], route_ref: str, lane: tuple[str, ...], mention_capture: dict[str, object] | None, + live_turn_capture: dict[str, object] | None = None, ) -> bool: if not self._source_matches: raise ValueError("Lark Topic intake belongs to a different provider") @@ -705,6 +712,8 @@ def capture( "route_ref": route_ref, "lane": lane, "event": dict(event), "mention_capture": mention_capture, "started": False, } + if live_turn_capture is not None: + packet["live_turn_capture"] = live_turn_capture self._directory.mkdir(parents=True, exist_ok=True, mode=0o700) write_private_json_atomic(path, packet) return True diff --git a/loopx/extensions/lark/goal_topic_dispatch.py b/loopx/extensions/lark/goal_topic_dispatch.py index f92a91be59..217d4a55a4 100644 --- a/loopx/extensions/lark/goal_topic_dispatch.py +++ b/loopx/extensions/lark/goal_topic_dispatch.py @@ -11,11 +11,16 @@ import threading from typing import NamedTuple +from ...chat_manager import manager_channel from .event_inbox import LarkTopicInboxIntake, LarkTopicInput +from .goal_channel_contracts import bindings_for_goal from .goal_channel_targets import goal_channel_target_for_name from .manager_context import opaque_digest +MentionCapture = Callable[[Mapping[str, object]], dict[str, object]] + + class _Work(NamedTuple): lane: tuple[str, ...] line: str @@ -221,7 +226,7 @@ def topic_intake_route( profile: str, route: Mapping[str, object], targets: Mapping[str, object], event: Mapping[str, object], ) -> TopicIntakeRoute | None: - if route.get("ingress_mode") != "session_queue" or not route.get("session_id"): + if route.get("ingress_mode") not in {"session_queue", "live_steering"} or not route.get("session_id"): return None target = goal_channel_target_for_name(targets, str(route.get("target_ref") or "")) identity = target.get("identity") if target is not None else None @@ -240,6 +245,71 @@ def topic_intake_route( return TopicIntakeRoute(route=route, targets=targets, digest=digest, inbox_key=key) +def profile_can_reserve_mention_worker( + snapshot: Mapping[str, object], + profile: str, + *, + bot_app_id: str, + capture_mention: MentionCapture | None, +) -> bool: + """Reserve a priority worker only when a bound Session can steer on @mention.""" + + target_payload = snapshot.get("target_payload") + target_payload = target_payload if isinstance(target_payload, Mapping) else {} + binding_payloads = snapshot.get("binding_payloads") + binding_payloads = binding_payloads if isinstance(binding_payloads, Mapping) else {} + for goal_id, payload in binding_payloads.items(): + if not isinstance(payload, Mapping): + continue + for binding in bindings_for_goal(payload, str(goal_id)): + if binding.get("enabled") is not True: + continue + session_id = str(binding.get("session_id") or "").strip() + routing = binding.get("routing") + routing = routing if isinstance(routing, Mapping) else {} + ingress_mode = str(routing.get("ingress_mode") or "") + if not session_id or ingress_mode not in {"live_steering", "session_queue"}: + continue + target = goal_channel_target_for_name( + target_payload, str(binding.get("target_ref") or "") + ) + if target is None or target.get("enabled") is not True: + continue + identity = target.get("identity") + identity = identity if isinstance(identity, Mapping) else {} + sender_profile = str(identity.get("sender_profile") or "").strip() + same_app = bool(bot_app_id) and str(identity.get("bot_app_id") or "") == bot_app_id + if sender_profile != profile and not same_app: + continue + if not (identity.get("bot_app_id") or identity.get("bot_open_id")): + continue + if capture_mention is not None: + channel = target.get("channel") + channel = channel if isinstance(channel, Mapping) else {} + chat_id = str(channel.get("chat_id") or "") + kind = str(routing.get("conversation_kind") or "goal") + route = { + "goal_id": str(goal_id), + "connection_id": str(binding.get("connection_id") or ""), + "agent_id": str(binding.get("agent_id") or "codex"), + "session_id": session_id, + "conversation_kind": kind, + "manager_channel_id": ( + manager_channel( + provider="lark", + audience=f"{sender_profile or 'default'}\0{chat_id}", + ) + if kind == "manager" + else "" + ), + "ingress_mode": ingress_mode, + } + if capture_mention(route).get("binding_matches") is not True: + continue + return True + return False + + class ProfileTopicIntake: """Turn durable Inbox inputs into coalesced wakes on the existing lanes.""" @@ -265,12 +335,14 @@ def capture( self, dispatch: ProfileEventDispatch, route: TopicIntakeRoute, lane: tuple[str, ...], event: Mapping[str, object], mention: dict[str, object] | None, config_path: Callable[[], Path], + *, live_turn_capture: dict[str, object] | None = None, ) -> bool: key = route.inbox_key if key not in self._sources: self._sources[key] = self._open(config_path()) captured = self._sources[key].capture( event=event, route_ref=route.digest, lane=lane, mention_capture=mention, + live_turn_capture=live_turn_capture, ) self._offer(dispatch, key, lane) return captured diff --git a/loopx/extensions/lark/goal_topic_runtime.py b/loopx/extensions/lark/goal_topic_runtime.py index bae603bdd9..8408b3329c 100644 --- a/loopx/extensions/lark/goal_topic_runtime.py +++ b/loopx/extensions/lark/goal_topic_runtime.py @@ -14,7 +14,7 @@ from pathlib import Path from typing import TYPE_CHECKING, Any, Literal, overload -from ...chat_manager import MANAGER_AGENT_OBJECTIVE, manager_channel +from ...chat_manager import MANAGER_AGENT_OBJECTIVE from ...control_plane.collaboration import conversation_reply_context from ...file_lock import exclusive_file_lock from .manager_routing import ( @@ -49,9 +49,11 @@ from .goal_channel_transport import lark_provider_mention_identities, record_event_timing from .goal_topic_connections import decide_lark_topic_event from .goal_topic_dispatch import ( + MentionCapture, ProfileEventDispatch, ProfileTopicIntake, TopicIntakeRoute, + profile_can_reserve_mention_worker, topic_inbox_key, topic_intake_route, ) @@ -104,7 +106,6 @@ [Mapping[str, Any], list[str]], Mapping[str, Any] ] ReviewCallbackHandler = Callable[[Mapping[str, Any]], Mapping[str, Any]] -MentionCapture = Callable[[Mapping[str, object]], dict[str, object]] EventLane = tuple[str, ...] CapturedEventLane = tuple[EventLane, dict[str, object] | None, TopicIntakeRoute | None] @@ -505,6 +506,7 @@ def poll_lark_goal_topic_profile_once( proposal_deliverer: ProposalDeliverer | None = None, private_admitter: Callable[[str, dict[str, object]], Mapping[str, object]] | None = None, mention_capture: Mapping[str, object] | None = None, + live_turn_capture: Mapping[str, object] | None = None, ) -> dict[str, Any]: """Consume one bounded event batch for an App and reuse Inbox reply/ACK.""" @@ -576,6 +578,7 @@ def poll_lark_goal_topic_profile_once( provider_runner=provider_runner, proposal_deliverer=proposal_deliverer, mention_capture=mention_capture, + live_turn_capture=live_turn_capture, ) except Exception: event_statuses.append("processing_failed") @@ -757,6 +760,7 @@ def report_result(result: Mapping[str, Any]) -> None: def dispatch_event( event: Mapping[str, object], mention: Mapping[str, object] | None, current_snapshot: Mapping[str, object], + *, live_turn_capture: Mapping[str, object] | None = None, ) -> None: message_id = str(event.get("message_id") or "") record_event_timing(profile, message_id, "dispatch_started") @@ -776,6 +780,7 @@ def dispatch_event( proposal_deliverer=proposal_deliverer, private_admitter=private_admitter, mention_capture=mention, + live_turn_capture=live_turn_capture, ) record_event_timing(profile, message_id, "dispatch_returned") report_result(result) @@ -801,12 +806,20 @@ def matches_input(inbox_key: str, packet: LarkTopicInput, snapshot: Mapping[str, return ( route is not None and route.inbox_key == inbox_key and route.digest == packet["route_ref"] and lane == packet["lane"] + and ( + route.route.get("ingress_mode") != "live_steering" + or packet.get("live_turn_capture") is not None + or packet["mention_capture"] is not None + ) ) def prepare_input(inbox_key: str, packet: LarkTopicInput) -> Callable[[], None]: fresh = snapshot_provider() if matches_input(inbox_key, packet, fresh): - return lambda: dispatch_event(packet["event"], packet["mention_capture"], fresh) + return lambda: dispatch_event( + packet["event"], packet["mention_capture"], fresh, + live_turn_capture=packet.get("live_turn_capture"), + ) return lambda: report_result({ "event_count": 1, "replied_count": 0, "event_statuses": ["ignored"], "event_reasons": [LarkTopicEventDecisionReason.BINDING_UNAVAILABLE.value], @@ -873,12 +886,18 @@ def handle_event(line: str) -> None: record_event_timing(profile, message_id, "lane_ready") priority = bool((mention_capture or {}).get("target_turn_id")) if reserve_priority and intake_route is not None and not priority: + live_turn_capture = None + if (intake_route.route.get("ingress_mode") == "live_steering" + and mention_capture is None and capture_mention is not None): + # A delayed ordinary live input keeps its original Turn. + live_turn_capture = capture_mention(intake_route.route) if intake.capture( dispatch, intake_route, lane, event, mention_capture, lambda: _inbox_config( runtime_root=root, route=intake_route.route, target_payload=intake_route.targets, )[0], + live_turn_capture=live_turn_capture, ): record_event_timing(profile, message_id, "inbox_captured") else: @@ -952,66 +971,10 @@ def _profile_supports_mention_steering( profile: str, capture_mention: MentionCapture | None = None, ) -> bool: - """Reserve a priority worker only when a bound Session can steer on @mention.""" - - profile_config = _active_profile_configs(snapshot).get(profile) - if profile_config is None: - return False - bot_app_id = str(profile_config.get("bot_app_id") or "") - target_payload = snapshot.get("target_payload") - target_payload = target_payload if isinstance(target_payload, Mapping) else {} - binding_payloads = snapshot.get("binding_payloads") - binding_payloads = binding_payloads if isinstance(binding_payloads, Mapping) else {} - for goal_id, payload in binding_payloads.items(): - if not isinstance(payload, Mapping): - continue - for binding in bindings_for_goal(payload, str(goal_id)): - if binding.get("enabled") is not True: - continue - session_id = str(binding.get("session_id") or "").strip() - routing = binding.get("routing") - routing = routing if isinstance(routing, Mapping) else {} - ingress_mode = str(routing.get("ingress_mode") or "") - if not session_id or ingress_mode not in {"live_steering", "session_queue"}: - continue - target = goal_channel_target_for_name( - target_payload, str(binding.get("target_ref") or "") - ) - if target is None or target.get("enabled") is not True: - continue - identity = target.get("identity") - identity = identity if isinstance(identity, Mapping) else {} - sender_profile = str(identity.get("sender_profile") or "").strip() - same_app = bool(bot_app_id) and str(identity.get("bot_app_id") or "") == bot_app_id - if sender_profile != profile and not same_app: - continue - if not (identity.get("bot_app_id") or identity.get("bot_open_id")): - continue - if capture_mention is not None: - channel = target.get("channel") - channel = channel if isinstance(channel, Mapping) else {} - chat_id = str(channel.get("chat_id") or "") - kind = str(routing.get("conversation_kind") or "goal") - route = { - "goal_id": str(goal_id), - "connection_id": str(binding.get("connection_id") or ""), - "agent_id": str(binding.get("agent_id") or "codex"), - "session_id": session_id, - "conversation_kind": kind, - "manager_channel_id": ( - manager_channel( - provider="lark", - audience=f"{sender_profile or 'default'}\0{chat_id}", - ) - if kind == "manager" - else "" - ), - "ingress_mode": ingress_mode, - } - if capture_mention(route).get("binding_matches") is not True: - continue - return True - return False + config = _active_profile_configs(snapshot).get(profile) + return config is not None and profile_can_reserve_mention_worker( + snapshot, profile, bot_app_id=str(config.get("bot_app_id") or ""), capture_mention=capture_mention, + ) def _bound_topic_session( @@ -1071,7 +1034,7 @@ def _route_native_mention(event: Mapping[str, object], route: Mapping[str, objec return isinstance(identity, Mapping) and _native_topic_mention(event, identity) -def _mention_reply(route: Mapping[str, object], decision: Mapping[str, object]) -> dict[str, object]: +def _steering_reply(route: Mapping[str, object], decision: Mapping[str, object]) -> dict[str, object]: receipt = _session_turn_effect(route) if decision["action"] == "delivered": return {"response_text": "已送达当前回合。", "effect_receipt": receipt} @@ -1082,6 +1045,7 @@ def _mention_reply(route: Mapping[str, object], decision: Mapping[str, object]) "queue_steering_attachments_unavailable": "本轮调整仅支持文字。原消息和附件保留在 Lark。", "live_steering_turn_mismatch": "接收时的回合已结束或切换。原消息保留在 Lark。", "live_steering_turn_not_started": "接收时的回合尚未就绪。原消息保留在 Lark。", + "live_steering_requires_active_turn": "接收消息时没有活动回合。原消息保留在 Lark。", }.get(str(decision.get("error_code")), "当前会话无法调整本轮。原消息保留在 Lark。") return {"response_text": text, "effect_receipt": receipt} @@ -1168,7 +1132,7 @@ def answer_lark_goal_topic( queued_text=message, origin="lark", ) if decision["action"] != "enqueue": - return _mention_reply(route, decision) + return _steering_reply(route, decision) ingress_mode = "session_queue" if ingress_mode in {"live_steering", "session_queue"}: @@ -1184,11 +1148,37 @@ def answer_lark_goal_topic( ) raise RuntimeError("bound Agent session is unavailable or no longer matches") if ingress_mode == "live_steering": - turn, _created = runtime_controller.steer_active_turn( - session_id=session_id, - client_ingress_id=client_turn_id, - message=message, - ) + capture = route.get("live_turn_capture") + if isinstance(capture, Mapping): + if (capture.get("binding_matches") is not True + or capture.get("session_id") != session_id + or capture.get("connection_id") != route.get("connection_id")): + return _steering_reply(route, {"action": "reject", "error_code": "live_steering_session_not_attached"}) + target_turn_id = capture.get("target_turn_id") + if not isinstance(target_turn_id, str) or not target_turn_id: + return _steering_reply(route, {"action": "reject", "error_code": "live_steering_requires_active_turn"}) + try: + turn, _created = runtime_controller.steer_active_turn( + session_id=session_id, client_ingress_id=client_turn_id, + message=message, expected_turn_id=target_turn_id, origin="lark", + ) + except Exception: + # A lost provider response may follow a successful send. + # Read the Chat owner's receipt without sending again. + from ...chat_runtime import STEERING_NOT_DELIVERED_CODES + + receipt = runtime_controller.store.steering_receipt(session_id, client_turn_id) or {} + code = receipt.get("error_code") + if ( + receipt.get("expected_turn_id") == target_turn_id and receipt.get("message") == message + and receipt.get("status") == "failed" and code in STEERING_NOT_DELIVERED_CODES + ): + return _steering_reply(route, {"action": "reject", "error_code": code}) + return _steering_reply(route, {"action": "unresolved"}) + else: + turn, _created = runtime_controller.steer_active_turn( + session_id=session_id, client_ingress_id=client_turn_id, message=message, + ) else: if ( manager @@ -1342,6 +1332,7 @@ def process_lark_goal_topic_event( provider_runner: Any | None = None, proposal_deliverer: ProposalDeliverer | None = None, mention_capture: Mapping[str, object] | None = None, + live_turn_capture: Mapping[str, object] | None = None, ) -> dict[str, Any]: """Single-flight a manager or live mention through ingress, reply and ACK. @@ -1361,6 +1352,11 @@ def process_lark_goal_topic_event( native_mention = isinstance(route, Mapping) and _route_native_mention(event, route, target_payload) if mention_capture is not None and not native_mention: return {"ok": True, "status": "ignored", "reason": "mention_binding_changed"} + if live_turn_capture is not None: + if not isinstance(route, Mapping) or route.get("ingress_mode") != "live_steering" or native_mention: + return {"ok": True, "status": "ignored", "reason": LarkTopicEventDecisionReason.BINDING_UNAVAILABLE.value} + route = {**route, "live_turn_capture": live_turn_capture} + decision = {**decision, "route": route} if native_mention: route = { **route, diff --git a/tests/extensions/test_lark_goal_topic_runtime.py b/tests/extensions/test_lark_goal_topic_runtime.py index 516e11d786..a87a0f6782 100644 --- a/tests/extensions/test_lark_goal_topic_runtime.py +++ b/tests/extensions/test_lark_goal_topic_runtime.py @@ -3597,10 +3597,12 @@ def test_message_event_projection_preserves_flat_cli_and_raw_provider_provenance assert row["content"] == "plain rendered text" -@pytest.mark.parametrize("ordinary_load", [63, 64, 128]) -@pytest.mark.parametrize("manager", [False, True]) +@pytest.mark.parametrize(("manager", "ingress_mode", "ordinary_load"), [ + *((manager, "session_queue", load) for manager in (False, True) for load in (63, 64, 128)), + (False, "live_steering", 63), (False, "live_steering", 64), +]) def test_native_mention_reaches_the_running_topic_before_its_ordinary_reply( - tmp_path: Path, manager: bool, ordinary_load: int + tmp_path: Path, manager: bool, ingress_mode: str, ordinary_load: int, ) -> None: """Use the actual listener, Inbox, Chat store, and Codex stdio protocol.""" import runpy @@ -3660,7 +3662,7 @@ def test_native_mention_reaches_the_running_topic_before_its_ordinary_reply( registry=registry, goal_id="goal-alpha", agent_id="codex", session_id=session_id, target_path=target_path, binding_path=binding_path, app_ref="mew", chat_id="oc_public_fixture", chat_name="Product group", - capture_scope="configured_chat_all", ingress_mode="session_queue", + capture_scope="configured_chat_all", ingress_mode=ingress_mode, conversation_kind="manager" if manager else "goal", turn_trigger="human_messages" if manager else None, runner=_connection_runner(state), cli_bin="fake-lark", @@ -3692,11 +3694,15 @@ def test_native_mention_reaches_the_running_topic_before_its_ordinary_reply( } for index in reversed(range(ordinary_load - 2)) ] + [wrong_bot] - assert len({ordinary["message_id"], *(item["message_id"] for item in backlog)}) == ordinary_load + ordinary_ids = {ordinary["message_id"], *(item["message_id"] for item in backlog)} + assert len(ordinary_ids) == ordinary_load idle = {**mention, "event_id": "evt_native_idle", "message_id": "om_native_idle", "content": "@Current bot Start normal input."} steered_reply = threading.Event() attachment_reply = threading.Event() + ordinary_reply_started = threading.Event() + allow_ordinary_reply = threading.Event() + ordinary_reader_blocked = threading.Event() last_ordinary_answer = threading.Event() runner = _reply_runner(state) @@ -3719,6 +3725,18 @@ def answer(route, text): mid = str(route.get("message_id") or "") if not route.get("native_mention"): observed_ordinary.append(mid) + if ingress_mode == "live_steering" and mid in ordinary_ids: + assert route["live_turn_capture"]["target_turn_id"] == store.turn_for_client( + session_id, initial_id + )["turn_id"] + # The active Turn already came from the App. Hold one ordinary + # reply while the listener reads and delivers the native @. + if mid == ordinary["message_id"]: + ordinary_reply_started.set() + assert allow_ordinary_reply.wait(10) + if mid == wrong_bot["message_id"]: + last_ordinary_answer.set() + return {"response_text": "Ordinary reply.", "effect_receipt": session_turn_effect(route)} if mid.startswith("om_other_bot_") and mid != backlog[0]["message_id"]: return { "response_text": f"Runtime response for {mid}.", @@ -3733,7 +3751,10 @@ def answer(route, text): "response_text": result, "effect_receipt": session_turn_effect(route), } - initial_id = "lark." + opaque_digest(ordinary["message_id"], topic) + initial_id = ( + "app-original" if ingress_mode == "live_steering" + else "lark." + opaque_digest(ordinary["message_id"], topic) + ) mention_id = "lark." + opaque_digest(mention["message_id"], topic) wrong_id = "lark." + opaque_digest(wrong_bot["message_id"], topic) idle_id = "lark." + opaque_digest(idle["message_id"], topic) @@ -3741,6 +3762,8 @@ def answer(route, text): def lines(): yield json.dumps(ordinary) + "\n" + if ingress_mode == "live_steering": + assert ordinary_reply_started.wait(5) deadline = time.monotonic() + 8 while time.monotonic() < deadline: initial = store.turn_for_client(session_id, initial_id) @@ -3751,7 +3774,9 @@ def lines(): raise AssertionError("the initial Topic input never reached the native executor") def interrupt_fixture_if_blocked(): if not steered_reply.is_set(): + ordinary_reader_blocked.set() controller.interrupt_turn(session_id=session_id, turn_id=initial["turn_id"]) + allow_ordinary_reply.set() watchdog = threading.Timer(8, interrupt_fixture_if_blocked) watchdog.daemon = True @@ -3759,6 +3784,7 @@ def interrupt_fixture_if_blocked(): try: for item in backlog: yield json.dumps(item) + "\n" + assert not ordinary_reader_blocked.is_set(), "ordinary backpressure blocked the stream before the mention" assert store.load_turn(session_id, initial["turn_id"])["status"] == "running" yield json.dumps(attachment) + "\n" assert attachment_reply.wait(5), "the original file needs an explicit retained-message reply" @@ -3773,9 +3799,11 @@ def interrupt_fixture_if_blocked(): finally: watchdog.cancel() release.touch() + allow_ordinary_reply.set() assert last_ordinary_answer.wait(40), "the FIFO backlog did not reach its last ordinary input" - queued = store.turn_for_client(session_id, wrong_id) - assert queued and queued["status"] == "completed" + if ingress_mode == "session_queue": + queued = store.turn_for_client(session_id, wrong_id) + assert queued and queued["status"] == "completed" deadline = time.monotonic() + 3 while store.load_session(session_id).get("active_turn_id"): assert time.monotonic() < deadline, "the completed Turn did not release the Session" @@ -3802,6 +3830,14 @@ def wait(self, timeout=None): return 0 try: + if ingress_mode == "live_steering": + started, _ = controller.enqueue_turn( + session_id=session_id, client_turn_id=initial_id, message="wait for steer", + work_dir=tmp_path, objective="Review delivery", origin="web", + ) + fixture["wait_for_active_turn"]( + store, session_id=session_id, turn_id=started["turn_id"], require_upstream=True, + ) result = stream_lark_goal_topic_profile( profile="mew", snapshot_provider=lambda: snapshot, stop=threading.Event(), runtime_root=runtime_root, answer=answer, process_factory=lambda _: FinishedConsumer(), reply_runner=reply, @@ -3829,7 +3865,9 @@ def wait(self, timeout=None): *(item["message_id"] for item in backlog), ] ordinary_prompts = [msg["text"] for msg in user if msg["turn_id"] != initial["turn_id"]] - expected_ordinary = [backlog[0]["content"], wrong_bot["content"], idle["content"]] + expected_ordinary = [idle["content"]] + if ingress_mode == "session_queue": + expected_ordinary[:0] = [backlog[0]["content"], wrong_bot["content"]] assert len(ordinary_prompts) == len(expected_ordinary) for prompt, expected in zip(ordinary_prompts, expected_ordinary, strict=True): assert prompt.strip().endswith(expected) @@ -3860,14 +3898,60 @@ def wait(self, timeout=None): interrupted_id = "lark." + opaque_digest(interrupted["message_id"], topic) assert store.turn_for_client(session_id, interrupted_id) is None assert store.steering_receipt(session_id, interrupted_id) is None + + if ingress_mode == "live_steering" and ordinary_load == 64: + replacement, _ = controller.enqueue_turn( + session_id=session_id, client_turn_id="app-replacement", message="wait for steer", + work_dir=tmp_path, objective="Review delivery", origin="web", + ) + replacement_id = replacement["turn_id"] + fixture["wait_for_active_turn"]( + store, session_id=session_id, turn_id=replacement_id, require_upstream=True, + ) + for suffix, target, feedback in ( + ("old", initial["turn_id"], "已结束或切换"), + ("idle", None, "没有活动回合"), + ("current", replacement_id, "Steered response."), + ): + message_id = f"om_live_capture_{suffix}" + input_event = {**wrong_bot, "message_id": message_id, "event_id": f"evt_live_capture_{suffix}"} + reply_result = process_lark_goal_topic_event( + **snapshot, event=input_event, runtime_root=runtime_root, answer=answer, reply_runner=reply, + live_turn_capture={ + "connection_id": binding_for_goal(binding_payload, "goal-alpha")["connection_id"], + "session_id": session_id, "binding_matches": True, "target_turn_id": target, + }, + ) + assert reply_result["ok"] and feedback in state["reply_text"], reply_result + ingress = "lark." + opaque_digest(message_id, topic) + receipt = store.steering_receipt(session_id, ingress) + assert store.turn_for_client(session_id, ingress) is None + if suffix == "old": + assert receipt["error_code"] == "live_steering_turn_mismatch" + if suffix == "idle": + assert receipt is None + if suffix == "current": + assert receipt["expected_turn_id"] == receipt["active_turn_id"] == replacement_id + assert receipt["status"] == "delivered" + steers = [ + event["payload"]["client_ingress_id"] for event in store.events_after(session_id, replacement_id, None) + if event["kind"] == "turn.steered" + ] + assert steers == ([ingress] if suffix == "current" else []) finally: release.touch() + allow_ordinary_reply.set() + active_id = store.load_session(session_id).get("active_turn_id") + if active_id and store.load_turn(session_id, active_id)["status"] in {"starting", "running"}: + controller.interrupt_turn(session_id=session_id, turn_id=active_id) + controller.wait_for_turn(session_id=session_id, turn_id=active_id, timeout_sec=4) controller.close_session(session_id) @pytest.mark.parametrize("binding_change", ["unchanged", "session", "topic"]) +@pytest.mark.parametrize("ingress_mode", ["session_queue", "live_steering"]) def test_topic_stream_recovers_only_unstarted_input_with_its_original_binding( - tmp_path: Path, binding_change: str, + tmp_path: Path, binding_change: str, ingress_mode: str, ) -> None: from loopx.extensions.lark.goal_topic_runtime import stream_lark_goal_topic_profile from loopx.extensions.lark.manager_context import session_turn_effect @@ -3882,7 +3966,7 @@ def test_topic_stream_recovers_only_unstarted_input_with_its_original_binding( registry=registry, goal_id="goal-alpha", agent_id="codex", session_id="session-original", target_path=targets_path, binding_path=binding_path, app_ref="mew", chat_id="oc_public_fixture", chat_name="Product group", - capture_scope="configured_chat_all", ingress_mode="session_queue", + capture_scope="configured_chat_all", ingress_mode=ingress_mode, runner=_connection_runner(state), cli_bin="fake-lark", )["ok"] targets = read_goal_channel_targets(targets_path) @@ -3907,6 +3991,8 @@ def test_topic_stream_recovers_only_unstarted_input_with_its_original_binding( answered = [] def answer(route, text): + if ingress_mode == "live_steering": + assert route["live_turn_capture"]["target_turn_id"] == "turn-original" answered.append((route["session_id"], route["message_id"], text)) if route["message_id"] == original["message_id"]: entered.set() @@ -3943,17 +4029,17 @@ def poll(self): def wait(self, timeout=None): return 0 - def consume(stdout, stop, reply): + def consume(stdout, stop, reply, *, active_turn): return stream_lark_goal_topic_profile( profile="mew", snapshot_provider=lambda: snapshot, runtime_root=runtime_root, stop=stop, answer=answer, process_factory=lambda _: FinishedConsumer(stdout), reply_runner=reply, capture_mention=lambda route: { "connection_id": route["connection_id"], "session_id": route["session_id"], - "binding_matches": True, "target_turn_id": None, + "binding_matches": True, "target_turn_id": active_turn, }, ) - stopped = consume(first_consumer(), stop, unverified_reply) + stopped = consume(first_consumer(), stop, unverified_reply, active_turn="turn-original") assert stopped["status"] == "stopped" assert answered == [("session-original", original["message_id"], original["content"])] assert first_writes == [original["message_id"]] @@ -3989,7 +4075,7 @@ def replacement_consumer(): assert last_reply.wait(5), "the original Inbox should recover without provider redelivery" yield "[event] exited (reason: timeout)\n" - resumed = consume(replacement_consumer(), threading.Event(), recovered_reply) + resumed = consume(replacement_consumer(), threading.Event(), recovered_reply, active_turn="turn-replacement") assert resumed["status"] == "stream_ended" expected = [] if rebind else [ ("session-original", row["message_id"], row["content"]) for row in later