fix(lark): share one App listener between groups and owner Chat

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

@@ -44,6 +44,23 @@ conversation authority. One machine/App lease covers both profile aliases and
 existing group listeners. Replies recheck the original source under that App;
 they do not require group-member scopes for private messages.
 
+An existing group listener can also admit owner-private Chat through the same
+verified App profile. Settings → Lark → owner private Chat reuses that profile;
+aliases of the same App cannot start a competing listener. One consumer dispatches
+explicit group events to the existing group admission and private events to
+Core owner admission. Rejection never falls through to the other audience.
+Group grants and review callbacks remain separate, and disconnecting either
+route preserves the other. HTTP/Core fixtures cover setup, missing owner login,
+App mismatch and route removal; this does not qualify remote Agent delivery or
+an installed provider roundtrip.
+Legacy group entries without a stored App id use the freshly verified private
+profile identity while holding both the original profile-scoped lease and the
+App-scoped lease, fencing old processes and new aliases together.
+Setup checks the effective shared configuration before publishing the Core
+binding; an App identity change before publication is rejected without changing
+that binding. Real HTTP/Core/runtime-refresh fixtures cover this path and cold
+listener discovery, with no provider messages or model work.
+
 Text admission persists the canonical Core Turn without waiting for its terminal
 answer. The existing bounded Session queue keeps follow-ups in FIFO order; another
 App can continue independently. Explicit `/status`, `/stop` and `/new` use the

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

@@ -33,6 +33,18 @@ admission、安装或手机旅程已通过；grant 撤销后历史仍可读，
 Lark 页面选择一个非默认 App、当前可用工作区和宿主 executor，读回监听状态、待回复
 数量及恢复缺口；重新绑定不会把旧 Session 移到其它工作区。
 
+既有群聊监听器也可通过同一个已核验 App profile 接入本人私聊。在设置 → Lark
+中选择原 profile 和本人私聊；同 App 的别名不能另起竞争监听器。单个 consumer
+将明确的群聊事件交给既有群聊准入，将私聊交给 Core 本人准入；拒绝后不转投
+另一受众。群聊授权和审阅回调保持独立，解绑一路保留另一路。HTTP/Core fixture
+覆盖配置、本人未登录、App 不匹配和路由移除，不代表远程 Agent 投递或安装态
+provider 往返已通过。
+未保存 App id 的旧群聊配置使用刚核验的私聊 profile 身份，同时保留原 profile
+级 consumer lease 及 App 级 lease，同时隔离旧进程与新别名。配置在发布 Core
+绑定前校验实际共享组合；发布前 App 身份变化
+则拒绝，绑定保持不变。真实 HTTP/Core/runtime-refresh fixture 覆盖此路径及冷启动
+监听发现，不发送 provider 消息或执行模型。
+
 共享 typed Core 核验 App、本人、受众和当前工作区 grant，拥有稳定 Session、原生
 Turn FIFO 和确切 stop/new/status 目标。传输层保存独立的受理与最终投递意图，通过
 provider 读回确认回复；发生没有 receipt 的不确定写入时不盲目重发。两 App 复用

loopx/capabilities/manager_context/ssh_evidence.py

@@ -12,7 +12,7 @@
 import subprocess
 from typing import Any
 
-from . import POLICY_SCHEMA, _root, _read, _write
+from . import POLICY_SCHEMA, _root, _read, _write, _require_external_channel
 from ...control_plane.collaboration.source_policy_fence import source_policy_fence
 from ...control_plane.status.ssh_host_catalog import configured_ssh_host_aliases
 
@@ -55,8 +55,7 @@ def grants(root, channel):
 
 
 def configure(root, *, channel, host, goal_ids, execute=False, config_path=None):
-    if not re.fullmatch(r"manager\.external\.[a-f0-9]{24}", channel):
-        raise ValueError("exact external manager channel required")
+    _require_external_channel(channel)
     if host not in configured_ssh_host_aliases(config_path):
         raise ValueError("host must be an existing SSH alias")
     if len(goal_ids) > 128 or any(

loopx/capabilities/native_chat/conversation_bindings.py

@@ -66,14 +66,17 @@ def configure(self, *, transport_ref: str, project_ref: str,
                   executor_endpoint_id: str, context_kind: str = "project",
                   project_grant: str | None = None, goal_scope: str | None = None,
                   audience: str | None = None, group_refs: list[str] | None = None,
-                  available_group_refs: list[str] | None = None) -> dict[str, Any]:
+                  available_group_refs: list[str] | None = None,
+                  expected_provider_ref: str | None = None) -> dict[str, Any]:
         if goal_scope is not None and (context_kind != "steward" or goal_scope not in {"selected", "all_registered"}):
             raise ValueError("select a supported steward Goal scope")
         if project_grant is None:
             project_grant = self.projects.workspace_grant if context_kind == "project" and executor_endpoint_id == "codex" else "workspace_read"
         if project_grant not in {"workspace_read", "workspace_write"} or (context_kind != "project" and project_grant != "workspace_read"):
             raise ValueError("workspace write authorization is only available for project Chat")
         observation = self.observe(transport_ref, audience=audience or "owner")
+        if expected_provider_ref is not None and observation["provider_ref"] != expected_provider_ref:
+            raise ValueError("the selected App changed before binding publication; verify it again")
         candidate = {
             "schema_version": "loopx_chat_conversation_binding_v0",
             "binding_id": uuid.uuid4().hex[:24], "transport_ref": transport_ref,

loopx/chat_lark_api.py

@@ -692,10 +692,13 @@ def provider_runner(args, cwd, timeout):
                             cached_auth[key] = original_runner(args, cwd, timeout)
                         return cached_auth[key]
                 for profile in selected_profiles if configured else []:
-                    if any(row["transport_ref"] == profile for row in configured):
+                    shared_private = next((row for row in configured if row["transport_ref"] == profile), None)
+                    if shared_private is not None and shared_private.get("audience", "owner") != "owner":
                         raise ValueError("this App owns a private listener; disconnect it first")
                     identity = _app_identity_for_private_guard(profile, provider_runner, cli_bin)
-                    if identity and any(row["provider_ref"] == identity_ref(identity) for row in configured):
+                    if shared_private is not None and (not identity or shared_private["provider_ref"] != identity_ref(identity)):
+                        raise ValueError("the shared App identity changed; verify its original profile")
+                    if identity and any(row["provider_ref"] == identity_ref(identity) and row["transport_ref"] != profile for row in configured):
                         raise ValueError("this App alias owns a private listener; disconnect it first")
             executor_endpoint_id = (
                 manager_executor_endpoint_default(

loopx/extensions/lark/goal_topic_runtime.py

@@ -125,7 +125,7 @@ def __init__(self, error_code: str, effect_receipt: Mapping[str, Any]) -> None:
 _EVENT_EXIT_REASON = re.compile(r"\(reason: (limit|timeout|signal)\)$")
 
 
-def _active_profile_configs(snapshot: Mapping[str, Any]) -> dict[str, dict[str, str]]:
+def _active_group_profile_configs(snapshot: Mapping[str, object]) -> dict[str, dict[str, str]]:
     target_payload = snapshot.get("target_payload")
     target_payload = target_payload if isinstance(target_payload, Mapping) else {}
     binding_payloads = snapshot.get("binding_payloads")
@@ -155,13 +155,45 @@ def _active_profile_configs(snapshot: Mapping[str, Any]) -> dict[str, dict[str,
                     "bot_app_id": str(identity.get("bot_app_id") or ""),
                 },
             )
-    for profile, config in dict(snapshot.get("private_profiles") or {}).items():
+    return profiles
+
+
+def _active_profile_configs(snapshot: Mapping[str, object]) -> dict[str, dict[str, str]]:
+    from .conversation_identity import identity_ref
+
+    profiles = _active_group_profile_configs(snapshot)
+    private = snapshot.get("private_profiles")
+    for profile, config in (private.items() if isinstance(private, Mapping) else []):
         if profile in profiles:
-            raise ValueError("an App cannot own both private and group listeners")
-        profiles[profile] = dict(config)
+            stored_app_id = profiles[profile].get("bot_app_id")
+            app_id = stored_app_id or config.get("bot_app_id")
+            if (config.get("audience", "owner") != "owner" or not app_id
+                    or identity_ref(app_id) != config.get("provider_ref")):
+                raise ValueError("shared listener requires the same verified App and separate audiences")
+            profiles[profile] = {**profiles[profile], **dict(config)}
+            if not stored_app_id:
+                # Keep the pre-upgrade group lease until its original owner
+                # exits. Verified private discovery supplies identity, not a
+                # reason to start a second consumer under a new App key.
+                profiles[profile]["consumer_ref"] = hashlib.sha256(profile.encode("utf-8")).hexdigest()[:32]
+        else:
+            profiles[profile] = dict(config)
     return profiles
 
 
+def _native_profile_event(snapshot: Mapping[str, Any], profile: str, event: Mapping[str, object]) -> bool:
+    """Select a transport lane only; its original owner still admits the input.
+
+    An owner-private binding may share one consumer with legacy group routes.
+    A private rejection never falls through to a group authority or vice versa.
+    """
+    config = dict(snapshot.get("private_profiles") or {}).get(profile)
+    return isinstance(config, Mapping) and not (
+        config.get("audience", "owner") == "owner" and event.get("chat_type") == "group"
+        and profile in _active_group_profile_configs(snapshot)
+    )
+
+
 def _default_simple_runner(args: list[str]) -> Mapping[str, Any]:
     try:
         completed = subprocess.run(
@@ -372,7 +404,7 @@ def _profile_event_lane(
     """Only serialize transport work; actual ingress re-reads its authority."""
 
     private_config = dict(snapshot.get("private_profiles") or {}).get(profile)
-    if isinstance(private_config, Mapping):
+    if isinstance(private_config, Mapping) and _native_profile_event(snapshot, profile, event):
         from .conversation_identity import lark_private_source
 
         try:
@@ -466,7 +498,7 @@ def poll_lark_goal_topic_profile_once(
     event_statuses: list[str] = []
     event_reasons: list[str | None] = []
     for event in events:
-        if profile in dict(snapshot.get("private_profiles") or {}):
+        if _native_profile_event(snapshot, profile, event):
             private_result = private_admitter(profile, dict(event)) if private_admitter else {"status": "private_admission_unavailable"}
             event_statuses.append(str(private_result.get("status") or "unknown"))
             event_reasons.append(None)

loopx/extensions/lark/goal_topic_runtime_service.py

@@ -4,7 +4,7 @@
 
 from collections.abc import Callable, Mapping
 from concurrent.futures import Future, ThreadPoolExecutor
-from contextlib import nullcontext
+from contextlib import ExitStack, nullcontext
 from datetime import datetime, timezone
 import hashlib
 import logging
@@ -27,7 +27,7 @@
 SnapshotProvider = Callable[[], Mapping[str, Any]]
 ProfilePoller = Callable[[str, threading.Event], None]
 ManagerRouteReconciler = Callable[[Mapping[str, Any]], Mapping[str, Any]]
-ProfileWorkerFingerprint = tuple[str, str, tuple[str, ...]]
+ProfileWorkerFingerprint = tuple[str, str, str, tuple[str, ...]]
 ProfileWorker = tuple[threading.Event, threading.Thread, ProfileWorkerFingerprint]
 
 
@@ -226,6 +226,7 @@ def _profile_worker_fingerprint(
         return (
             str(profile_config.get("cli_bin") or "lark-cli"),
             str(profile_config.get("bot_app_id") or profile_config.get("binding_id") or ""),
+            str(profile_config.get("consumer_ref") or ""),
             callback_chats,
         )
 
@@ -333,16 +334,20 @@ def handle_review_callback(
                     # servers can use different runtime roots or ports. The
                     # consumer lease must therefore be machine/App scoped.
                     app_id = str(profile_config.get("bot_app_id") or "")
-                    # Both paths lease the same App-scoped digest, so aliases
-                    # or a second server cannot start another consumer.
-                    digest = (hashlib.sha256(app_id.encode("utf-8")).hexdigest()[:32]
-                              if app_id else str(profile_config.get("consumer_ref")
-                              or hashlib.sha256(profile.encode("utf-8")).hexdigest()[:32]))
-                    lease = Path.home() / ".loopx" / "lark-consumers" / digest
-                    with try_exclusive_file_lock(
-                        lease, operation="lark_event_consumer"
-                    ) as acquired:
-                        if acquired is None:
+                    # A composed legacy route explicitly retains its original
+                    # profile lease. Otherwise both paths use the App digest.
+                    digest = str(profile_config.get("consumer_ref") or (
+                        hashlib.sha256((app_id or profile).encode("utf-8")).hexdigest()[:32]))
+                    digests = {digest}
+                    if app_id:
+                        digests.add(hashlib.sha256(app_id.encode("utf-8")).hexdigest()[:32])
+                    # A legacy/shared consumer holds both identities: wait for
+                    # the old process and fence new aliases on other servers.
+                    with ExitStack() as leases:
+                        acquired = [leases.enter_context(try_exclusive_file_lock(
+                            Path.home() / ".loopx" / "lark-consumers" / key,
+                            operation="lark_event_consumer")) for key in sorted(digests)]
+                        if any(lease is None for lease in acquired):
                             self._update_health(
                                 profile,
                                 status="standby",
@@ -361,7 +366,7 @@ def handle_review_callback(
                             proposal_deliverer=deliver_proposals,
                             review_callback_handler=(
                                 handle_review_callback
-                                if self.action_service is not None and profile not in dict(self.snapshot_provider().get("private_profiles") or {})
+                                if self.action_service is not None and profile in runtime._active_group_profile_configs(self.snapshot_provider())
                                 else None
                             ),
                             health_sink=lambda update: self._update_health(

loopx/extensions/lark/private_conversation_api.py

@@ -71,8 +71,9 @@ def _private_conversation_connect(self) -> None:
             if set(body) - {"context_kind", "project_grant", "goal_scope", "audience", "group_ids"} != {"app_ref", "project_ref", "executor_endpoint_id"}:
                 raise ValueError("select an App, authorized workspace and executor")
             profile = str(body["app_ref"])
-            existing = _active_profile_configs(build_lark_goal_topic_runtime_snapshot(
-                registry_path=self.server.registry_path, runtime_root_override=self.server.runtime_root_override))
+            snapshot = build_lark_goal_topic_runtime_snapshot(
+                registry_path=self.server.registry_path, runtime_root_override=self.server.runtime_root_override)
+            existing = _active_profile_configs(snapshot)
             from .conversation_identity import identity_ref
             observed = self.server.runtime_controller.project_contexts.conversation_bindings.observe(
                 profile, audience=body.get("audience") or "owner")
@@ -94,8 +95,19 @@ def _private_conversation_connect(self) -> None:
             group_apps = [str(config.get("bot_app_id") or _app_identity_for_private_guard(
                 other_profile, self._lark_runner(), self._require_lark_cli()))
                 for other_profile, config in existing.items()]
-            if profile in existing or any(identity_ref(app) == observed["provider_ref"] for app in group_apps):
-                raise ValueError("this App already owns a group listener; disconnect it first")
+            shared_group = existing.get(profile)
+            if shared_group is not None:
+                if body.get("audience", "owner") != "owner" or identity_ref(
+                    str(shared_group.get("bot_app_id") or _app_identity_for_private_guard(
+                        profile, self._lark_runner(), self._require_lark_cli()))
+                ) != observed["provider_ref"]:
+                    raise ValueError("shared listener requires the same verified App and a private owner audience")
+            elif any(identity_ref(app) == observed["provider_ref"] for app in group_apps):
+                raise ValueError("this App already has a group listener; select its existing profile to add private Chat")
+            # Compose the same transport configuration used by refresh before
+            # publishing a binding. Failure must leave the Core grant untouched.
+            _active_profile_configs({**snapshot, "private_profiles": {
+                profile: {**observed, "audience": body.get("audience", "owner")}}})
             endpoint = str(body["executor_endpoint_id"])
             if not any(row["agent_id"] == endpoint and row["available"] for row in self.server.runtime_controller.capabilities()):
                 raise ValueError("the selected executor is unavailable")
@@ -104,7 +116,8 @@ def _private_conversation_connect(self) -> None:
                 context_kind=str(body.get("context_kind", "project")),
                 project_grant=str(body["project_grant"]) if "project_grant" in body else None,
                 goal_scope=str(body["goal_scope"]) if "goal_scope" in body else None,
-                audience=body.get("audience"), group_refs=group_refs, available_group_refs=available_group_refs)
+                audience=body.get("audience"), group_refs=group_refs, available_group_refs=available_group_refs,
+                expected_provider_ref=observed["provider_ref"])
             self.server.lark_goal_topic_runtime.refresh()
         except (ValueError, OSError, KeyError) as exc:
             self._send_error(str(exc), status=400)

loopx/extensions/lark/private_conversations.py

@@ -86,6 +86,7 @@ def profiles(self) -> dict[str, dict[str, str]]:
                     or identity_ref(app_id) != row["provider_ref"]):
                 continue
             profiles[row["transport_ref"]] = {"cli_bin": self.cli_bin, "provider_ref": row["provider_ref"],
+                "bot_app_id": app_id, "audience": row.get("audience", "owner"),
                 "binding_id": row["binding_id"], "consumer_ref": hashlib.sha256(app_id.encode("utf-8")).hexdigest()[:32]}
         return profiles