From 42cc3dba9059ff7fcf68ea68fed401fe0c68cf6b Mon Sep 17 00:00:00 2001
From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com>
Date: Fri, 13 Nov 2026 11:24:10 +0800
Subject: [PATCH 1/8] fix(lark): drain private intake before reconnecting

Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com>
---
 .../lark/goal_topic_runtime_service.py        | 133 +++++++++-----
 .../extensions/lark/private_conversations.py  |   5 +-
 tests/test_lark_private_ingress.py            | 172 +++++++++++++++---
 3 files changed, 238 insertions(+), 72 deletions(-)

diff --git a/loopx/extensions/lark/goal_topic_runtime_service.py b/loopx/extensions/lark/goal_topic_runtime_service.py
index a6a260fd30..ab3f437ef5 100644
--- a/loopx/extensions/lark/goal_topic_runtime_service.py
+++ b/loopx/extensions/lark/goal_topic_runtime_service.py
@@ -9,6 +9,7 @@
 import hashlib
 import logging
 from pathlib import Path
+from subprocess import Popen
 import threading
 import time
 from typing import Any
@@ -30,6 +31,7 @@
 ManagerRouteReconciler = Callable[[Mapping[str, Any]], Mapping[str, Any]]
 ProfileWorkerFingerprint = tuple[str, str, str, tuple[str, ...]]
 ProfileWorker = tuple[threading.Event, threading.Thread, ProfileWorkerFingerprint]
+MessageConsumer = tuple[threading.Event, Popen[str], threading.Event]
 
 
 class LarkGoalTopicRuntimeService:
@@ -55,6 +57,7 @@ def __init__(
         self.manager_route_reconciler = manager_route_reconciler
         self._lock = threading.Lock()
         self._workers: dict[str, ProfileWorker] = {}
+        self._message_consumers: dict[str, MessageConsumer] = {}
         self._health: dict[str, dict[str, Any]] = {}
         self._closed = threading.Event()
         self._delivery_wakeup = threading.Event()
@@ -128,14 +131,23 @@ def step() -> None:
                 if ingress is not None and ingress.done():
                     try:
                         for profile in ingress.result():
-                            # Stop only this service's existing live worker.
-                            # Its finally/refresh path releases the App lease
-                            # before starting its replacement; Core work and
-                            # historical messages are never replayed.
                             with self._lock:
                                 worker = self._workers.get(profile)
-                                if worker is not None:
-                                    worker[0].set()
+                                current = self._message_consumers.get(profile)
+                            if worker is None or current is None or current[0] is not worker[0]:
+                                continue
+                            stop, process, requested = current
+                            if stop.is_set() or requested.is_set() or process.poll() is not None:
+                                continue
+                            # End this message pipe and let it drain under the
+                            # existing App lease. Stopping the worker discards
+                            # transient events already read by the stream.
+                            requested.set()
+                            try:
+                                process.terminate()
+                            except OSError:
+                                requested.clear()
+                                logging.getLogger(__name__).warning("Lark message consumer could not reconnect")
                     except Exception:
                         logging.getLogger(__name__).warning("Private Chat source observation is unavailable")
                     ingress = None
@@ -241,10 +253,12 @@ def step() -> None:
                 on_error=lambda: logging.getLogger(__name__).warning(
                     "Private Chat delivery discovery is pending"))
 
-    def _admit_private(self, profile: str, event: dict[str, Any]) -> dict[str, Any]:
+    def _admit_private(self, profile: str, event: dict[str, Any], *,
+                       expected_binding_id: str | None = None) -> dict[str, Any]:
         if self._closed.is_set():
             raise RuntimeError("Lark conversation service is closed")
-        result = self.private_conversations.admit(profile, event, defer_background_stop=True)
+        options = {"expected_binding_id": expected_binding_id} if expected_binding_id is not None else {}
+        result = self.private_conversations.admit(profile, event, defer_background_stop=True, **options)
         # Admission owns persistence and authorization. Only after it returns
         # may the existing reply worker look for its durable work; duplicates
         # and denied inputs still cannot manufacture a pending journal entry.
@@ -308,9 +322,14 @@ def _profile_worker_fingerprint(
             if self.action_service is not None
             else ()
         )
+        app_ref = str(profile_config.get("bot_app_id") or "")
+        binding_id = str(profile_config.get("binding_id") or "")
+        # A same-App replacement changes which workspace receives live input.
+        # Keep the App lease identity, but retire the old worker on refresh.
+        worker_ref = f"{app_ref}:{binding_id}" if app_ref and binding_id else app_ref or binding_id
         return (
             str(profile_config.get("cli_bin") or "lark-cli"),
-            str(profile_config.get("bot_app_id") or profile_config.get("binding_id") or ""),
+            worker_ref,
             str(profile_config.get("consumer_ref") or ""),
             callback_chats,
         )
@@ -411,12 +430,36 @@ def handle_review_callback(
                             profile_config=profile_config,
                         )
 
-                    profile_config = (
-                        runtime._active_profile_configs(self.snapshot_provider()).get(
-                            profile
-                        )
-                        or {}
+                    snapshot = self.snapshot_provider()
+                    profile_config = runtime._active_profile_configs(snapshot).get(profile) or {}
+                    private_profiles = snapshot.get("private_profiles")
+                    private_binding = private_profiles.get(profile) if isinstance(private_profiles, Mapping) else None
+                    binding_id = private_binding.get("binding_id") if isinstance(private_binding, Mapping) else None
+                    reconnect_requested = threading.Event()
+                    private_admitter = self._admit_private if self.private_conversations is not None else None
+                    bound_owner = (
+                        private_admitter is not None
+                        and isinstance(binding_id, str) and bool(binding_id)
+                        and isinstance(private_binding, Mapping)
+                        and private_binding.get("audience") == "owner"
                     )
+                    if bound_owner:
+                        def admit_bound_private(profile_name: str, event: dict[str, Any]) -> dict[str, Any]:
+                            # An event read by the old stream cannot follow a
+                            # same-App rebind while it waits for admission.
+                            return self._admit_private(profile_name, event, expected_binding_id=binding_id)
+
+                        private_admitter = admit_bound_private
+
+                    def start_consumer(args: list[str]) -> Popen[str]:
+                        process = runtime._default_process_factory(args)
+                        if args[3:6] == ["event", "consume", "im.message.receive_v1"]:
+                            with self._lock:
+                                worker = self._workers.get(profile)
+                                if worker is not None and worker[0] is stop:
+                                    self._message_consumers[profile] = (stop, process, reconnect_requested)
+                        return process
+
                     # A bot App can have several local profile aliases and Chat
                     # servers can use different runtime roots or ports. The
                     # consumer lease must therefore be machine/App scoped.
@@ -447,28 +490,34 @@ def capture(route: Mapping[str, object]) -> dict[str, object]:
                                 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,
-                            # The listener checks current binding eligibility
-                            # per event; a later saved Session must not require
-                            # replacing a consumer with an active reply.
-                            capture_mention=capture,
-                            private_admitter=(self._admit_private
-                                if self.private_conversations is not None else None),
-                            proposal_deliverer=deliver_proposals,
-                            review_callback_handler=(
-                                handle_review_callback
-                                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(
-                                profile, **dict(update)
-                            ),
-                        )
+                        process_options = {"process_factory": start_consumer} if bound_owner else {}
+                        try:
+                            result = runtime.stream_lark_goal_topic_profile(
+                                profile=profile,
+                                snapshot_provider=self.snapshot_provider,
+                                stop=stop,
+                                runtime_root=self.runtime_root,
+                                answer=answer,
+                                # Recheck binding eligibility per event, including
+                                # events drained from an old provider connection.
+                                capture_mention=capture,
+                                private_admitter=private_admitter,
+                                proposal_deliverer=deliver_proposals,
+                                review_callback_handler=(
+                                    handle_review_callback
+                                    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(
+                                    profile, **dict(update)
+                                ),
+                                **process_options,
+                            )
+                        finally:
+                            with self._lock:
+                                current = self._message_consumers.get(profile)
+                                if current is not None and current[0] is stop:
+                                    del self._message_consumers[profile]
                     if result.get("status") == "configuration_removed":
                         self._update_health(
                             profile,
@@ -480,16 +529,18 @@ def capture(route: Mapping[str, object]) -> dict[str, object]:
                     if stop.is_set():
                         break
                     restart_count += 1
+                    error_code = result.get("error_code")
+                    requested_disconnect = (
+                        reconnect_requested.is_set()
+                        and error_code == "lark_event_source_disconnected"
+                    )
                     self._update_health(
                         profile,
                         status="retrying",
                         error_code=(
                             None
-                            if result.get("ok") is True
-                            else str(
-                                result.get("error_code")
-                                or "lark_event_listener_failed"
-                            )
+                            if result.get("ok") is True or requested_disconnect
+                            else str(error_code or "lark_event_listener_failed")
                         ),
                         restart_count=restart_count,
                     )
diff --git a/loopx/extensions/lark/private_conversations.py b/loopx/extensions/lark/private_conversations.py
index a84c0687f8..7183fb2bde 100644
--- a/loopx/extensions/lark/private_conversations.py
+++ b/loopx/extensions/lark/private_conversations.py
@@ -822,10 +822,13 @@ def read(args: Sequence[str], **_kwargs: Any) -> subprocess.CompletedProcess[str
 
     @effect_runtime_request_scope()
     def admit(self, profile: str, event: dict[str, Any], *,
-              defer_background_stop: bool = False, history_review: bool = False) -> dict[str, Any]:
+              defer_background_stop: bool = False, history_review: bool = False,
+              expected_binding_id: str | None = None) -> dict[str, Any]:
         record_event_timing(profile, str(event.get("message_id") or ""), "admission_started")
         try:
             binding = self._binding(profile)
+            if expected_binding_id is not None and binding["binding_id"] != expected_binding_id:
+                return {"status": "audience_rejected"}
             source = lark_private_source(provider_ref=binding["provider_ref"], event=event)
             if history_review and (binding.get("context_kind") != "steward"
                     or binding.get("audience", "owner") != "owner" or event.get("chat_type") != "p2p"):
diff --git a/tests/test_lark_private_ingress.py b/tests/test_lark_private_ingress.py
index d2d656ae3e..aaf21b2901 100644
--- a/tests/test_lark_private_ingress.py
+++ b/tests/test_lark_private_ingress.py
@@ -192,37 +192,149 @@ def test_stale_or_unreadable_health_is_not_verified(source_monitor):
     assert f.monitor.health(f.binding["binding_id"])["status"] == "unavailable"
 
 
-def test_default_scanner_reconnects_only_its_owned_live_worker(tmp_path, monkeypatch):
-    from loopx.extensions.lark.goal_topic_runtime_service import LarkGoalTopicRuntimeService
-
+@pytest.mark.parametrize("transition", ["reconnect", "shutdown", "revoke", "rebind_during_reconnect"])
+def test_live_private_ingress_across_listener_transition(ordinary, tmp_path, monkeypatch, transition):  # noqa: F811
+    from concurrent.futures import ThreadPoolExecutor
+    from pathlib import Path
+
+    from loopx.extensions.lark import goal_topic_runtime as runtime
+    from loopx.extensions.lark import goal_topic_runtime_service as lifecycle
+    from loopx.extensions.lark.conversation_identity import identity_ref
+    from test_lark_private_conversations import connect
+
+    store, controller, provider, transport = connect(ordinary)
+    profile = "notes-app"
+    binding = next(row for row in transport.bindings.read()["bindings"] if row["transport_ref"] == profile)
+    config = transport.profiles()[profile]
+    active = True
+    first, second, third = [provider.event(profile, key, f"Live request {key}")
+                           for key in ("first", "second", "third")]
+    in_admission, release, waiting, received_third = (threading.Event() for _ in range(4))
     clock = [0.0]
-    monkeypatch.setattr("loopx.extensions.lark.goal_topic_runtime_service.time.monotonic", lambda: clock[0])
-    observed = []
-    source = SimpleNamespace(observe=lambda **_kwargs: observed.append(True) or ["owner-app"])
-
-    def pending():
-        clock[0] = 61
-        return []
-
-    service = LarkGoalTopicRuntimeService(snapshot_provider=lambda: {}, runtime_root=tmp_path,
-        runtime_controller=object(), private_conversations=SimpleNamespace(pending_delivery_paths=pending, ingress=source))
-    own, other = threading.Event(), threading.Event()
-    service._workers = {"owner-app": (own, threading.current_thread(), ("", "", ())),
-                        "other-app": (other, threading.current_thread(), ("", "", ()))}
-    waited = threading.Event()
-    waits = 0
-
-    def wait(_seconds):
-        nonlocal waits
-        waits += 1
-        if own.is_set() or waits >= 100:
-            service._closed.set()
-        else:
-            waited.wait(0.01)
-
-    monkeypatch.setattr(service._delivery_wakeup, "wait", wait)
-    service._reconcile_private()
-    assert observed == [True] and own.is_set() and not other.is_set()
+    admitted = []
+    consumers = []
+
+    def admit(profile_name, event, **kwargs):
+        if event["message_id"] == first["message_id"]:
+            in_admission.set()
+            assert release.wait(8)
+        result = transport.admit(profile_name, event, **kwargs)
+        admitted.append((event["message_id"], result["status"]))
+        if event["message_id"] == third["message_id"]:
+            received_third.set()
+        return result
+
+    owner = SimpleNamespace(admit=admit, pending_delivery_paths=lambda: [],
+                            ingress=SimpleNamespace(observe=lambda **_: [profile, "unowned-app"]))
+    service = lifecycle.LarkGoalTopicRuntimeService(
+        snapshot_provider=lambda: {"private_profiles": {profile: config}} if active else {},
+        runtime_root=store.root.parent, runtime_controller=controller, private_conversations=owner)
+    monkeypatch.setattr(lifecycle, "time", SimpleNamespace(monotonic=lambda: clock[0]))
+    monkeypatch.setattr(Path, "home", lambda: tmp_path)
+
+    class Consumer:
+        def __init__(self, initial):
+            self.initial = initial
+            self.exited, self.waited = threading.Event(), threading.Event()
+            self.stdout = self.lines()
+
+        def lines(self):
+            yield "[event] ready event_key=im.message.receive_v1\n"
+            if self.initial:
+                yield json.dumps(first)
+                assert in_admission.wait(8)
+                yield json.dumps(second)
+                waiting.set()  # The reader advanced past submission of the second line.
+            else:
+                if transition == "reconnect":
+                    yield json.dumps({**second, "event_id": "synthetic_provider_redelivery"})
+                yield json.dumps(third)
+            assert self.exited.wait(8)
+            yield "[event] exited (reason: signal)\n"
+
+        def poll(self):
+            return 0 if self.exited.is_set() else None
+
+        def terminate(self):
+            self.exited.set()
+
+        def wait(self, timeout=None):
+            assert self.exited.is_set()
+            self.waited.set()
+            return 0
+
+    def spawn(args):
+        assert args[3:6] == ["event", "consume", "im.message.receive_v1"]
+        if consumers:
+            assert len(consumers) == 1 and consumers[0].waited.is_set()
+        consumer = Consumer(not consumers)
+        consumers.append(consumer)
+        return consumer
+
+    stream = runtime.stream_lark_goal_topic_profile
+
+    def isolated_stream(**kwargs):
+        kwargs.setdefault("process_factory", spawn)
+        return stream(**kwargs)
+
+    monkeypatch.setattr(runtime, "_default_process_factory", spawn)
+    monkeypatch.setattr(runtime, "stream_lark_goal_topic_profile", isolated_stream)
+    service.start()
+    with ThreadPoolExecutor(max_workers=1) as pool:
+        try:
+            assert waiting.wait(8)
+            if transition in ("reconnect", "rebind_during_reconnect"):
+                clock[0] = 61
+                service._delivery_wakeup.set()
+            elif transition == "shutdown":
+                pool.submit(service.close)
+            else:
+                active = False
+                transport.bindings.disconnect(binding["binding_id"], expected_revision=transport.bindings.read()["revision"])
+                service.refresh()
+            assert consumers[0].exited.wait(8)
+            assert len(consumers) == 1
+            if transition == "rebind_during_reconnect":
+                projects = transport.bindings.projects
+                other = tmp_path / "other-workspace"
+                other.mkdir()
+                projects.roots.append((other, other))
+                target = next(x for x in projects.available() if x["workspace_path"] == str(other))
+                replacement = transport.bindings.configure(transport_ref=profile, project_ref=target["project_ref"],
+                                                           executor_endpoint_id="codex")
+                assert replacement["binding_id"] != binding["binding_id"]
+                config = transport.profiles()[profile]
+            release.set()
+            if transition in ("reconnect", "rebind_during_reconnect"):
+                assert received_third.wait(8)
+            else:
+                assert consumers[0].waited.wait(8)
+        finally:
+            release.set()
+            service.close()
+            controller.close()
+
+    second_ref = identity_ref(binding["provider_ref"], second["message_id"])
+    if transition == "reconnect":
+        assert [event_id for event_id, _ in admitted[:2]] == [first["message_id"], second["message_id"]]
+        assert consumers[0].waited.is_set() and len(consumers) == 2
+        assert [event_id for event_id, _ in admitted].count(second["message_id"]) == 2
+        original = transport.core.read_request(second_ref)
+        assert original["source"]["source_ref"] == transport.core.read_request(
+            identity_ref(binding["provider_ref"], first["message_id"]))["source"]["source_ref"]
+        messages = store.messages(original["session_id"])
+        assert sum(m.get("text") == second["content"] for m in messages) == 1
+    elif transition == "rebind_during_reconnect":
+        assert admitted[:2] == [(first["message_id"], "audience_rejected"),
+                               (second["message_id"], "audience_rejected")]
+        assert not (transport.core.root / f"{second_ref}.json").exists()
+        assert not (transport.root / f"{second_ref}.json").exists()
+        third_ref = identity_ref(binding["provider_ref"], third["message_id"])
+        assert transport.core.read_request(third_ref)["binding_id"] == replacement["binding_id"]
+        assert consumers[0].waited.is_set() and len(consumers) == 2
+    else:
+        assert not any(event_id == second["message_id"] for event_id, _ in admitted)
+        assert len(consumers) == 1
 
 
 def test_unknown_failed_capture_is_not_replayed(source_monitor):

From 37b6add48b8bd09cbe82b273a1e286960bd21520 Mon Sep 17 00:00:00 2001
From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com>
Date: Fri, 13 Nov 2026 12:38:04 +0800
Subject: [PATCH 2/8] fix(lark): preserve group intake across private binding
 changes

Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com>
---
 .../lark/goal_topic_runtime_service.py        | 108 +++++++---
 tests/test_lark_private_conversations.py      |   2 +-
 tests/test_lark_private_ingress.py            | 200 +++++++++++++++---
 3 files changed, 249 insertions(+), 61 deletions(-)

diff --git a/loopx/extensions/lark/goal_topic_runtime_service.py b/loopx/extensions/lark/goal_topic_runtime_service.py
index ab3f437ef5..1933ea8410 100644
--- a/loopx/extensions/lark/goal_topic_runtime_service.py
+++ b/loopx/extensions/lark/goal_topic_runtime_service.py
@@ -5,6 +5,7 @@
 from collections.abc import Callable, Mapping
 from concurrent.futures import Future, ThreadPoolExecutor
 from contextlib import ExitStack, nullcontext
+from dataclasses import dataclass
 from datetime import datetime, timezone
 import hashlib
 import logging
@@ -29,7 +30,23 @@
 SnapshotProvider = Callable[[], Mapping[str, Any]]
 ProfilePoller = Callable[[str, threading.Event], None]
 ManagerRouteReconciler = Callable[[Mapping[str, Any]], Mapping[str, Any]]
-ProfileWorkerFingerprint = tuple[str, str, str, tuple[str, ...]]
+
+
+@dataclass(frozen=True)
+class ProfileTransport:
+    cli_bin: str
+    app_id: str
+    consumer_lease_key: str
+    callback_chats: tuple[str, ...]
+    other_binding_id: str
+
+
+@dataclass(frozen=True)
+class ProfileWorkerFingerprint:
+    transport: ProfileTransport
+    owner_binding_id: str = ""
+
+
 ProfileWorker = tuple[threading.Event, threading.Thread, ProfileWorkerFingerprint]
 MessageConsumer = tuple[threading.Event, Popen[str], threading.Event]
 
@@ -134,20 +151,11 @@ def step() -> None:
                             with self._lock:
                                 worker = self._workers.get(profile)
                                 current = self._message_consumers.get(profile)
-                            if worker is None or current is None or current[0] is not worker[0]:
-                                continue
-                            stop, process, requested = current
-                            if stop.is_set() or requested.is_set() or process.poll() is not None:
-                                continue
-                            # End this message pipe and let it drain under the
-                            # existing App lease. Stopping the worker discards
-                            # transient events already read by the stream.
-                            requested.set()
-                            try:
-                                process.terminate()
-                            except OSError:
-                                requested.clear()
-                                logging.getLogger(__name__).warning("Lark message consumer could not reconnect")
+                                if (
+                                    worker is not None and worker[2].owner_binding_id
+                                    and current is not None and current[0] is worker[0]
+                                ):
+                                    self._disconnect_message_consumer(current)
                     except Exception:
                         logging.getLogger(__name__).warning("Private Chat source observation is unavailable")
                     ingress = None
@@ -311,6 +319,28 @@ def health_snapshot(self) -> dict[str, dict[str, Any]]:
         with self._lock:
             return {profile: dict(health) for profile, health in self._health.items()}
 
+    @staticmethod
+    def _disconnect_message_consumer(consumer: MessageConsumer) -> bool:
+        """End the current pipe under the service lock and let the worker drain."""
+
+        stop, process, requested = consumer
+        if stop.is_set():
+            return False
+        if not requested.is_set() and process.poll() is None:
+            requested.set()
+            try:
+                process.terminate()
+            except OSError:
+                requested.clear()
+                logging.getLogger(__name__).warning("Lark message consumer could not reconnect")
+                return False
+        return True
+
+    @staticmethod
+    def _consumer_lease_key(profile: str, config: Mapping[str, str]) -> str:
+        identity = config.get("bot_app_id") or profile
+        return str(config.get("consumer_ref") or hashlib.sha256(identity.encode("utf-8")).hexdigest()[:32])
+
     def _profile_worker_fingerprint(
         self,
         snapshot: Mapping[str, Any],
@@ -322,16 +352,18 @@ def _profile_worker_fingerprint(
             if self.action_service is not None
             else ()
         )
-        app_ref = str(profile_config.get("bot_app_id") or "")
+        app_id = str(profile_config.get("bot_app_id") or "")
         binding_id = str(profile_config.get("binding_id") or "")
-        # A same-App replacement changes which workspace receives live input.
-        # Keep the App lease identity, but retire the old worker on refresh.
-        worker_ref = f"{app_ref}:{binding_id}" if app_ref and binding_id else app_ref or binding_id
-        return (
-            str(profile_config.get("cli_bin") or "lark-cli"),
-            worker_ref,
-            str(profile_config.get("consumer_ref") or ""),
-            callback_chats,
+        owner_binding_id = binding_id if app_id and profile_config.get("audience") == "owner" else ""
+        return ProfileWorkerFingerprint(
+            transport=ProfileTransport(
+                cli_bin=str(profile_config.get("cli_bin") or "lark-cli"),
+                app_id=app_id,
+                consumer_lease_key=self._consumer_lease_key(profile, profile_config),
+                callback_chats=callback_chats,
+                other_binding_id="" if owner_binding_id else binding_id,
+            ),
+            owner_binding_id=owner_binding_id,
         )
 
     def _poll_profile(self, profile: str, stop: threading.Event) -> None:
@@ -436,7 +468,10 @@ def handle_review_callback(
                     private_binding = private_profiles.get(profile) if isinstance(private_profiles, Mapping) else None
                     binding_id = private_binding.get("binding_id") if isinstance(private_binding, Mapping) else None
                     reconnect_requested = threading.Event()
-                    private_admitter = self._admit_private if self.private_conversations is not None else None
+                    # A group-only connection must not pick up private work
+                    # read before a newly added owner binding.
+                    private_admitter = (self._admit_private if self.private_conversations is not None
+                                        and isinstance(private_binding, Mapping) else None)
                     bound_owner = (
                         private_admitter is not None
                         and isinstance(binding_id, str) and bool(binding_id)
@@ -466,9 +501,7 @@ def start_consumer(args: list[str]) -> Popen[str]:
                     app_id = str(profile_config.get("bot_app_id") or "")
                     # 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}
+                    digests = {self._consumer_lease_key(profile, profile_config)}
                     if app_id:
                         digests.add(hashlib.sha256(app_id.encode("utf-8")).hexdigest()[:32])
                     # A legacy/shared consumer holds both identities: wait for
@@ -490,7 +523,7 @@ def capture(route: Mapping[str, object]) -> dict[str, object]:
                                 route, self.runtime_controller
                             )
 
-                        process_options = {"process_factory": start_consumer} if bound_owner else {}
+                        process_options = {"process_factory": start_consumer} if self.private_conversations is not None else {}
                         try:
                             result = runtime.stream_lark_goal_topic_profile(
                                 profile=profile,
@@ -611,7 +644,22 @@ def refresh(self) -> None:
                 if self._workers[profile][2] != desired[profile]
             }
             for profile in stale:
-                stop, _thread, _fingerprint = self._workers.pop(profile)
+                stop, thread, previous = self._workers[profile]
+                replacement = desired.get(profile)
+                consumer = self._message_consumers.get(profile)
+                if (
+                    replacement is not None
+                    and previous.transport == replacement.transport
+                    and consumer is not None
+                    and consumer[0] is stop
+                ):
+                    # Group routes still own their queued messages. The old
+                    # private admission closure fences its revoked binding
+                    # while the same worker drains under the existing App lease.
+                    if self._disconnect_message_consumer(consumer):
+                        self._workers[profile] = (stop, thread, replacement)
+                    continue
+                self._workers.pop(profile)
                 stop.set()
             missing = set(desired) - set(self._workers)
             for profile in sorted(missing):
diff --git a/tests/test_lark_private_conversations.py b/tests/test_lark_private_conversations.py
index b3e1d8cc59..0e35a7c3b1 100644
--- a/tests/test_lark_private_conversations.py
+++ b/tests/test_lark_private_conversations.py
@@ -2813,7 +2813,7 @@ def test_app_messages_and_answers_reach_the_original_chat_once_without_new_execu
         runtime.close()
 
 
-def test_app_projection_discovers_a_restored_transcript_with_the_same_size_and_mtime(ordinary):
+def test_app_projection_discovers_a_restored_transcript_with_the_same_size_and_mtime(ordinary):  # noqa: F811
     import os
 
     store, runtime, provider, transport, original, turns, before = _app_projection_journey(ordinary)
diff --git a/tests/test_lark_private_ingress.py b/tests/test_lark_private_ingress.py
index aaf21b2901..384a676609 100644
--- a/tests/test_lark_private_ingress.py
+++ b/tests/test_lark_private_ingress.py
@@ -192,6 +192,33 @@ def test_stale_or_unreadable_health_is_not_verified(source_monitor):
     assert f.monitor.health(f.binding["binding_id"])["status"] == "unavailable"
 
 
+class _ListenerProcess:
+    """Yield fixture events and let the real service end the stream."""
+
+    def __init__(self, events):
+        self.exited = threading.Event()
+        self.waited = threading.Event()
+        self.stdout = self.lines(events)
+
+    def lines(self, events):
+        yield "[event] ready event_key=im.message.receive_v1\n"
+        for event in events:
+            yield json.dumps(event)
+        assert self.exited.wait(8)
+        yield "[event] exited (reason: signal)\n"
+
+    def poll(self):
+        return 0 if self.exited.is_set() else None
+
+    def terminate(self):
+        self.exited.set()
+
+    def wait(self, timeout=None):
+        assert self.exited.wait(timeout or 8)
+        self.waited.set()
+        return 0
+
+
 @pytest.mark.parametrize("transition", ["reconnect", "shutdown", "revoke", "rebind_during_reconnect"])
 def test_live_private_ingress_across_listener_transition(ordinary, tmp_path, monkeypatch, transition):  # noqa: F811
     from concurrent.futures import ThreadPoolExecutor
@@ -232,42 +259,22 @@ def admit(profile_name, event, **kwargs):
     monkeypatch.setattr(lifecycle, "time", SimpleNamespace(monotonic=lambda: clock[0]))
     monkeypatch.setattr(Path, "home", lambda: tmp_path)
 
-    class Consumer:
-        def __init__(self, initial):
-            self.initial = initial
-            self.exited, self.waited = threading.Event(), threading.Event()
-            self.stdout = self.lines()
-
-        def lines(self):
-            yield "[event] ready event_key=im.message.receive_v1\n"
-            if self.initial:
-                yield json.dumps(first)
-                assert in_admission.wait(8)
-                yield json.dumps(second)
-                waiting.set()  # The reader advanced past submission of the second line.
-            else:
-                if transition == "reconnect":
-                    yield json.dumps({**second, "event_id": "synthetic_provider_redelivery"})
-                yield json.dumps(third)
-            assert self.exited.wait(8)
-            yield "[event] exited (reason: signal)\n"
-
-        def poll(self):
-            return 0 if self.exited.is_set() else None
+    def first_connection():
+        yield first
+        assert in_admission.wait(8)
+        yield second
+        waiting.set()  # The reader advanced past submission of the second line.
 
-        def terminate(self):
-            self.exited.set()
-
-        def wait(self, timeout=None):
-            assert self.exited.is_set()
-            self.waited.set()
-            return 0
+    def replacement_connection():
+        if transition == "reconnect":
+            yield {**second, "event_id": "synthetic_provider_redelivery"}
+        yield third
 
     def spawn(args):
         assert args[3:6] == ["event", "consume", "im.message.receive_v1"]
         if consumers:
             assert len(consumers) == 1 and consumers[0].waited.is_set()
-        consumer = Consumer(not consumers)
+        consumer = _ListenerProcess(replacement_connection() if consumers else first_connection())
         consumers.append(consumer)
         return consumer
 
@@ -337,6 +344,139 @@ def isolated_stream(**kwargs):
         assert len(consumers) == 1
 
 
+@pytest.mark.parametrize("binding_change", ["replace", "add", "remove"])
+def test_shared_listener_drains_group_inbox_when_private_workspace_changes(
+    ordinary,  # noqa: F811
+    tmp_path, monkeypatch, binding_change,
+):
+    from pathlib import Path
+
+    from extensions.test_lark_goal_topic_runtime import _reply_runner, _seed_legacy_topic
+    from loopx.extensions.lark import goal_topic_runtime as runtime
+    from loopx.extensions.lark.event_inbox import inspect_lark_event_inbox
+    from loopx.extensions.lark.goal_channel_contracts import read_goal_channel_binding
+    from loopx.extensions.lark.goal_channel_targets import read_goal_channel_targets
+    from test_lark_private_conversations import connect
+
+    store, controller, provider, transport = connect(ordinary)
+    profile = "notes-app"
+    target_path, binding_path = tmp_path / "targets.json", tmp_path / "group-binding.json"
+    _seed_legacy_topic(target_path, binding_path)
+    targets = read_goal_channel_targets(target_path)
+    targets["targets"]["fixture"]["identity"].update(
+        sender_profile=profile, bot_app_id=provider.profile_apps[profile],
+    )
+    bindings = {"goal-alpha": read_goal_channel_binding(binding_path)}
+    before = transport.profiles()[profile]["binding_id"]
+    if binding_change == "add":
+        transport.bindings.disconnect(before, expected_revision=transport.bindings.read()["revision"])
+    config = transport.profiles().get(profile)
+
+    def snapshot():
+        return {"target_payload": targets, "binding_payloads": bindings,
+                "private_profiles": {profile: config} if config is not None else {}}
+
+    group_events = [
+        dict(provider.topic(name, f"@linkmacbot group {name}", root="om_topic_alpha",
+                            chat="oc_public_fixture"), mentioned=True)
+        for name in ("first", "second", "third")
+    ]
+    new_private = provider.event(profile, "after_rebind", "/status")
+    in_group_answer, release, queued, next_stream_ready = (threading.Event() for _ in range(4))
+    answered, consumer_processes = [], []
+    replies = {}
+
+    def answer(*, route, **_kwargs):
+        message_id = route["message_id"]
+        if message_id == group_events[0]["message_id"]:
+            in_group_answer.set()
+            assert release.wait(8)
+        answered.append(message_id)
+        return "Group answer"
+
+    def admit(profile_name, event, **kwargs):
+        result = transport.admit(profile_name, event, **kwargs)
+        if event["message_id"] == new_private["message_id"]:
+            assert result["status"] == "command_recorded"
+            next_stream_ready.set()
+        return result
+
+    def first_connection():
+        yield group_events[0]
+        assert in_group_answer.wait(8)
+        yield group_events[1]
+        queued.set()
+
+    def replacement_connection():
+        yield group_events[2] if binding_change == "remove" else new_private
+
+    def spawn(args):
+        assert args[3:6] == ["event", "consume", "im.message.receive_v1"]
+        if consumer_processes:
+            assert len(consumer_processes) == 1 and consumer_processes[0].waited.is_set()
+        consumer = _ListenerProcess(replacement_connection() if consumer_processes else first_connection())
+        consumer_processes.append(consumer)
+        return consumer
+
+    stream = runtime.stream_lark_goal_topic_profile
+
+    def isolated_stream(**kwargs):
+        kwargs.setdefault("process_factory", spawn)
+        health_sink = kwargs["health_sink"]
+
+        def observe_health(update):
+            health_sink(update)
+            if binding_change == "remove" and len(consumer_processes) == 2 and update.get("replied_count") == 1:
+                next_stream_ready.set()
+
+        return stream(**{**kwargs, "health_sink": observe_health}, reply_runner=_reply_runner(replies))
+
+    monkeypatch.setattr(Path, "home", lambda: tmp_path)
+    monkeypatch.setattr(runtime, "_default_process_factory", spawn)
+    monkeypatch.setattr(runtime, "stream_lark_goal_topic_profile", isolated_stream)
+    monkeypatch.setattr(runtime, "answer_lark_goal_topic", answer)
+    service = runtime.LarkGoalTopicRuntimeService(
+        snapshot_provider=snapshot, runtime_root=store.root.parent,
+        runtime_controller=controller, private_conversations=SimpleNamespace(admit=admit),
+    )
+
+    try:
+        service.refresh()
+        assert queued.wait(8), service.health_snapshot()
+        if binding_change == "remove":
+            transport.bindings.disconnect(before, expected_revision=transport.bindings.read()["revision"])
+            config = None
+        else:
+            projects = transport.bindings.projects
+            other = tmp_path / "other-workspace"
+            other.mkdir()
+            projects.roots.append((other, other))
+            project = next(row for row in projects.available() if row["workspace_path"] == str(other))
+            replacement = transport.bindings.configure(
+                transport_ref=profile, project_ref=project["project_ref"], executor_endpoint_id="codex",
+            )
+            config = transport.profiles()[profile]
+            assert config["binding_id"] == replacement["binding_id"] != before
+        service.refresh()
+        assert consumer_processes[0].exited.wait(8)
+        assert len(consumer_processes) == 1
+        release.set()
+        assert next_stream_ready.wait(8), service.health_snapshot()
+        expected = group_events if binding_change == "remove" else group_events[:2]
+        assert answered == [event["message_id"] for event in expected]
+        reply_targets = [call[call.index("--message-id") + 1] for call in replies["calls"]
+                         if "+messages-reply" in call and "--dry-run" not in call]
+        assert reply_targets == answered
+        config_path, = (store.root.parent / ".loopx/config/lark-goal-topics").glob("*.json")
+        inbox = inspect_lark_event_inbox(project=store.root.parent, config_path=config_path)
+        assert inbox["processed_count"] == len(expected)
+        assert inbox["pending_count"] == 0
+    finally:
+        release.set()
+        service.close()
+        controller.close()
+
+
 def test_unknown_failed_capture_is_not_replayed(source_monitor):
     f = source_monitor
     native = {**f.carrier, "status": "failed", "event": {**f.carrier["event"], "message_id": "om_missed"}}

From fec483a290c1a394fe05b9ce471770c2cf127b68 Mon Sep 17 00:00:00 2001
From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com>
Date: Fri, 13 Nov 2026 14:16:53 +0800
Subject: [PATCH 3/8] fix(lark): retain profile-only group intake during
 private setup

Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com>
---
 .../lark/goal_topic_runtime_service.py        | 41 ++++++--
 .../extensions/lark/private_conversations.py  | 25 +++--
 tests/test_lark_private_ingress.py            | 94 ++++++++++++++-----
 3 files changed, 121 insertions(+), 39 deletions(-)

diff --git a/loopx/extensions/lark/goal_topic_runtime_service.py b/loopx/extensions/lark/goal_topic_runtime_service.py
index 1933ea8410..b696f3790d 100644
--- a/loopx/extensions/lark/goal_topic_runtime_service.py
+++ b/loopx/extensions/lark/goal_topic_runtime_service.py
@@ -341,6 +341,37 @@ def _consumer_lease_key(profile: str, config: Mapping[str, str]) -> str:
         identity = config.get("bot_app_id") or profile
         return str(config.get("consumer_ref") or hashlib.sha256(identity.encode("utf-8")).hexdigest()[:32])
 
+    def _listener_configs(self, snapshot: Mapping[str, Any]) -> dict[str, dict[str, str]]:
+        from . import goal_topic_runtime as runtime
+
+        profiles = runtime._active_profile_configs(snapshot)
+        legacy = {
+            profile for profile, config in runtime._active_group_profile_configs(snapshot).items()
+            if not config.get("bot_app_id") and not profiles[profile].get("bot_app_id")
+        }
+        inventory = getattr(self.private_conversations, "configured_profile_apps", None)
+        if not legacy or not callable(inventory):
+            return profiles
+        try:
+            apps = inventory()
+        except OSError:
+            # Unknown inventory does not establish continuity across a binding
+            # change. The ordinary stop path still handles unknown identities.
+            return profiles
+        for profile in legacy:
+            app_id = apps.get(profile)
+            if app_id:
+                # Fence the original profile and the resolved App on both
+                # sides of a private binding change. Routing keeps the original
+                # snapshot and its separate group and owner grants.
+                config = profiles[profile]
+                profiles[profile] = {
+                    **config,
+                    "bot_app_id": app_id,
+                    "consumer_ref": self._consumer_lease_key(profile, config),
+                }
+        return profiles
+
     def _profile_worker_fingerprint(
         self,
         snapshot: Mapping[str, Any],
@@ -463,7 +494,7 @@ def handle_review_callback(
                         )
 
                     snapshot = self.snapshot_provider()
-                    profile_config = runtime._active_profile_configs(snapshot).get(profile) or {}
+                    profile_config = self._listener_configs(snapshot).get(profile) or {}
                     private_profiles = snapshot.get("private_profiles")
                     private_binding = private_profiles.get(profile) if isinstance(private_profiles, Mapping) else None
                     binding_id = private_binding.get("binding_id") if isinstance(private_binding, Mapping) else None
@@ -499,8 +530,8 @@ def start_consumer(args: list[str]) -> Popen[str]:
                     # 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 "")
-                    # A composed legacy route explicitly retains its original
-                    # profile lease. Otherwise both paths use the App digest.
+                    # Legacy group listeners retain their original profile
+                    # lease alongside the resolved App lease.
                     digests = {self._consumer_lease_key(profile, profile_config)}
                     if app_id:
                         digests.add(hashlib.sha256(app_id.encode("utf-8")).hexdigest()[:32])
@@ -616,12 +647,10 @@ def capture(route: Mapping[str, object]) -> dict[str, object]:
                     self.refresh()
 
     def refresh(self) -> None:
-        from . import goal_topic_runtime as runtime
-
         if self._closed.is_set():
             return
         snapshot = self.snapshot_provider()
-        profile_configs = runtime._active_profile_configs(snapshot)
+        profile_configs = self._listener_configs(snapshot)
         desired = {
             profile: self._profile_worker_fingerprint(
                 snapshot,
diff --git a/loopx/extensions/lark/private_conversations.py b/loopx/extensions/lark/private_conversations.py
index 7183fb2bde..e9a942eef8 100644
--- a/loopx/extensions/lark/private_conversations.py
+++ b/loopx/extensions/lark/private_conversations.py
@@ -254,6 +254,20 @@ def profiles(self) -> dict[str, dict[str, str]]:
         bindings = self.bindings.read()["bindings"]
         if not bindings:
             return {}
+        apps = self.configured_profile_apps()
+        profiles = {}
+        for row in bindings:
+            app_id = apps.get(row["transport_ref"], "")
+            if (row.get("enabled") is not True or not APP_ID_PATTERN.fullmatch(app_id)
+                    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
+
+    def configured_profile_apps(self) -> dict[str, str]:
+        """Read local App identities for listener leases, without granting access."""
         result = call(self.runner, [self.cli_bin, "profile", "list"])
         try:
             configured = json.loads(str(result.get("stdout") or ""))
@@ -271,16 +285,7 @@ def profiles(self) -> dict[str, dict[str, str]]:
         # profile names take precedence over the App-ID fallback.
         for app_id in list(apps.values()):
             apps.setdefault(app_id, app_id)
-        profiles = {}
-        for row in bindings:
-            app_id = apps.get(row["transport_ref"], "")
-            if (row.get("enabled") is not True or not APP_ID_PATTERN.fullmatch(app_id)
-                    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
+        return {profile: app_id for profile, app_id in apps.items() if APP_ID_PATTERN.fullmatch(app_id)}
 
     @effect_runtime_request_scope()
     def health(self) -> dict[str, dict[str, int]]:
diff --git a/tests/test_lark_private_ingress.py b/tests/test_lark_private_ingress.py
index 384a676609..b6024f15ef 100644
--- a/tests/test_lark_private_ingress.py
+++ b/tests/test_lark_private_ingress.py
@@ -344,15 +344,23 @@ def isolated_stream(**kwargs):
         assert len(consumers) == 1
 
 
-@pytest.mark.parametrize("binding_change", ["replace", "add", "remove"])
+@pytest.mark.parametrize(("stored_app", "binding_change"), [
+    (True, "replace"), (True, "add"), (True, "remove"),
+    (False, "replace"), (False, "add"), (False, "remove"),
+    (False, "add_after_retarget"), (False, "remove_after_retarget"),
+])
 def test_shared_listener_drains_group_inbox_when_private_workspace_changes(
     ordinary,  # noqa: F811
-    tmp_path, monkeypatch, binding_change,
+    tmp_path, monkeypatch, stored_app, binding_change,
 ):
+    import http.client
     from pathlib import Path
 
     from extensions.test_lark_goal_topic_runtime import _reply_runner, _seed_legacy_topic
+    from loopx.chat_server import ChatHTTPServer, ChatRequestHandler
     from loopx.extensions.lark import goal_topic_runtime as runtime
+    from loopx.extensions.lark.cli_resolution import LarkCliResolution
+    from loopx.extensions.lark.conversation_identity import identity_ref
     from loopx.extensions.lark.event_inbox import inspect_lark_event_inbox
     from loopx.extensions.lark.goal_channel_contracts import read_goal_channel_binding
     from loopx.extensions.lark.goal_channel_targets import read_goal_channel_targets
@@ -363,18 +371,26 @@ def test_shared_listener_drains_group_inbox_when_private_workspace_changes(
     target_path, binding_path = tmp_path / "targets.json", tmp_path / "group-binding.json"
     _seed_legacy_topic(target_path, binding_path)
     targets = read_goal_channel_targets(target_path)
-    targets["targets"]["fixture"]["identity"].update(
-        sender_profile=profile, bot_app_id=provider.profile_apps[profile],
-    )
+    identity = targets["targets"]["fixture"]["identity"]
+    original_app = provider.profile_apps[profile]
+    identity.update(sender_profile=profile, bot_open_id="ou_notes_app_bot")
+    if stored_app:
+        identity["bot_app_id"] = original_app
+    else:
+        identity.pop("bot_app_id")
     bindings = {"goal-alpha": read_goal_channel_binding(binding_path)}
     before = transport.profiles()[profile]["binding_id"]
-    if binding_change == "add":
+    adding = binding_change.startswith("add")
+    removing = binding_change.startswith("remove")
+    retargeted = binding_change.endswith("_retarget")
+    if adding:
         transport.bindings.disconnect(before, expected_revision=transport.bindings.read()["revision"])
-    config = transport.profiles().get(profile)
+
+    group_config = {"target_payload": targets, "binding_payloads": bindings}
 
     def snapshot():
-        return {"target_payload": targets, "binding_payloads": bindings,
-                "private_profiles": {profile: config} if config is not None else {}}
+        private = transport.profiles().get(profile)
+        return {**group_config, "private_profiles": {profile: private} if private is not None else {}}
 
     group_events = [
         dict(provider.topic(name, f"@linkmacbot group {name}", root="om_topic_alpha",
@@ -394,8 +410,10 @@ def answer(*, route, **_kwargs):
         answered.append(message_id)
         return "Group answer"
 
+    original_admit = transport.admit
+
     def admit(profile_name, event, **kwargs):
-        result = transport.admit(profile_name, event, **kwargs)
+        result = original_admit(profile_name, event, **kwargs)
         if event["message_id"] == new_private["message_id"]:
             assert result["status"] == "command_recorded"
             next_stream_ready.set()
@@ -408,7 +426,7 @@ def first_connection():
         queued.set()
 
     def replacement_connection():
-        yield group_events[2] if binding_change == "remove" else new_private
+        yield group_events[2] if removing else new_private
 
     def spawn(args):
         assert args[3:6] == ["event", "consume", "im.message.receive_v1"]
@@ -426,7 +444,7 @@ def isolated_stream(**kwargs):
 
         def observe_health(update):
             health_sink(update)
-            if binding_change == "remove" and len(consumer_processes) == 2 and update.get("replied_count") == 1:
+            if removing and len(consumer_processes) == 2 and update.get("replied_count") == 1:
                 next_stream_ready.set()
 
         return stream(**{**kwargs, "health_sink": observe_health}, reply_runner=_reply_runner(replies))
@@ -435,34 +453,61 @@ def observe_health(update):
     monkeypatch.setattr(runtime, "_default_process_factory", spawn)
     monkeypatch.setattr(runtime, "stream_lark_goal_topic_profile", isolated_stream)
     monkeypatch.setattr(runtime, "answer_lark_goal_topic", answer)
+    monkeypatch.setattr(transport, "admit", admit)
+    monkeypatch.setattr("loopx.chat_lark_api.build_lark_goal_topic_runtime_snapshot", lambda **_: group_config)
+    monkeypatch.setattr(ChatRequestHandler, "_lark_runner", lambda _: provider)
     service = runtime.LarkGoalTopicRuntimeService(
         snapshot_provider=snapshot, runtime_root=store.root.parent,
-        runtime_controller=controller, private_conversations=SimpleNamespace(admit=admit),
+        runtime_controller=controller, private_conversations=transport,
     )
+    server = ChatHTTPServer(("127.0.0.1", 0), ChatRequestHandler)
+    server.chat_store, server.runtime_controller = store, controller
+    server.registry_path, server.runtime_root_override = controller.registry_path, store.root.parent
+    server.verbose = False
+    server.lark_cli_resolution = LarkCliResolution(
+        command="lark-cli", available=True, source="explicit", version="fixture", error_code=None,
+    )
+    server.lark_private_conversations = transport
+    server.lark_goal_topic_runtime = service
+    server_thread = threading.Thread(target=server.serve_forever, daemon=True)
+    server_thread.start()
+
+    def change_private_binding(method, body):
+        connection = http.client.HTTPConnection(*server.server_address, timeout=15)
+        try:
+            connection.request(method, "/api/chat/lark/private-conversations", json.dumps(body),
+                               headers={"Content-Type": "application/json"})
+            response = connection.getresponse()
+            data = json.loads(response.read())
+            assert response.status == 200, data
+        finally:
+            connection.close()
 
     try:
         service.refresh()
         assert queued.wait(8), service.health_snapshot()
-        if binding_change == "remove":
-            transport.bindings.disconnect(before, expected_revision=transport.bindings.read()["revision"])
-            config = None
+        if retargeted:
+            provider.profile_apps[profile] = "cli_retargeted_app"
+        if removing:
+            change_private_binding("DELETE", {"binding_id": before, "revision": transport.bindings.read()["revision"]})
         else:
             projects = transport.bindings.projects
             other = tmp_path / "other-workspace"
             other.mkdir()
             projects.roots.append((other, other))
             project = next(row for row in projects.available() if row["workspace_path"] == str(other))
-            replacement = transport.bindings.configure(
-                transport_ref=profile, project_ref=project["project_ref"], executor_endpoint_id="codex",
-            )
-            config = transport.profiles()[profile]
-            assert config["binding_id"] == replacement["binding_id"] != before
-        service.refresh()
+            change_private_binding("POST", {"app_ref": profile, "project_ref": project["project_ref"],
+                                            "executor_endpoint_id": "codex"})
+            replacement = next(row for row in transport.bindings.read()["bindings"] if row["transport_ref"] == profile)
+            assert transport.profiles()[profile]["binding_id"] == replacement["binding_id"] != before
+            assert replacement["provider_ref"] == identity_ref(provider.profile_apps[profile])
         assert consumer_processes[0].exited.wait(8)
         assert len(consumer_processes) == 1
         release.set()
         assert next_stream_ready.wait(8), service.health_snapshot()
-        expected = group_events if binding_change == "remove" else group_events[:2]
+        expected = group_events[:1] if retargeted else group_events[:2]
+        if removing:
+            expected += group_events[2:]
         assert answered == [event["message_id"] for event in expected]
         reply_targets = [call[call.index("--message-id") + 1] for call in replies["calls"]
                          if "+messages-reply" in call and "--dry-run" not in call]
@@ -474,6 +519,9 @@ def observe_health(update):
     finally:
         release.set()
         service.close()
+        server.shutdown()
+        server.server_close()
+        server_thread.join(3)
         controller.close()
 
 

From 970f631c5a167c83079438e7c8c5dc02c02b6064 Mon Sep 17 00:00:00 2001
From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com>
Date: Fri, 13 Nov 2026 14:31:54 +0800
Subject: [PATCH 4/8] fix(lark): recognize the same CLI during listener handoff

Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com>
---
 .../lark/goal_topic_runtime_service.py        | 29 ++++++++++++-------
 .../extensions/lark/private_conversations.py  |  4 +--
 tests/test_lark_private_ingress.py            | 19 ++++++++++--
 3 files changed, 38 insertions(+), 14 deletions(-)

diff --git a/loopx/extensions/lark/goal_topic_runtime_service.py b/loopx/extensions/lark/goal_topic_runtime_service.py
index b696f3790d..e7fbe0c913 100644
--- a/loopx/extensions/lark/goal_topic_runtime_service.py
+++ b/loopx/extensions/lark/goal_topic_runtime_service.py
@@ -10,6 +10,7 @@
 import hashlib
 import logging
 from pathlib import Path
+import shutil
 from subprocess import Popen
 import threading
 import time
@@ -18,6 +19,7 @@
 from ...chat_manager import MANAGER_AGENT_OBJECTIVE
 from ...file_lock import try_exclusive_file_lock
 from ..background_runtime import HostedWorker
+from .cli_resolution import LarkCliUnavailableError
 from .goal_channel_contracts import bindings_for_goal
 from .manager_context import session_turn_effect
 from .team_plan_confirmation import (
@@ -350,21 +352,24 @@ def _listener_configs(self, snapshot: Mapping[str, Any]) -> dict[str, dict[str,
             if not config.get("bot_app_id") and not profiles[profile].get("bot_app_id")
         }
         inventory = getattr(self.private_conversations, "configured_profile_apps", None)
-        if not legacy or not callable(inventory):
-            return profiles
-        try:
-            apps = inventory()
-        except OSError:
-            # Unknown inventory does not establish continuity across a binding
-            # change. The ordinary stop path still handles unknown identities.
+        if not callable(inventory):
             return profiles
+        apps_by_cli: dict[str, dict[str, str]] = {}
         for profile in legacy:
-            app_id = apps.get(profile)
+            config = profiles[profile]
+            cli_bin = config["cli_bin"]
+            if cli_bin not in apps_by_cli:
+                try:
+                    apps_by_cli[cli_bin] = inventory(cli_bin=cli_bin)
+                except (OSError, LarkCliUnavailableError):
+                    # Keep the ordinary stop path if App continuity cannot
+                    # be established from this group's CLI configuration.
+                    apps_by_cli[cli_bin] = {}
+            app_id = apps_by_cli[cli_bin].get(profile)
             if app_id:
                 # Fence the original profile and the resolved App on both
                 # sides of a private binding change. Routing keeps the original
                 # snapshot and its separate group and owner grants.
-                config = profiles[profile]
                 profiles[profile] = {
                     **config,
                     "bot_app_id": app_id,
@@ -386,9 +391,13 @@ def _profile_worker_fingerprint(
         app_id = str(profile_config.get("bot_app_id") or "")
         binding_id = str(profile_config.get("binding_id") or "")
         owner_binding_id = binding_id if app_id and profile_config.get("audience") == "owner" else ""
+        cli_bin = str(profile_config.get("cli_bin") or "lark-cli")
+        executable = shutil.which(cli_bin)
         return ProfileWorkerFingerprint(
             transport=ProfileTransport(
-                cli_bin=str(profile_config.get("cli_bin") or "lark-cli"),
+                # Legacy settings can name the same CLI that Chat resolved
+                # to a full path when configuring the private binding.
+                cli_bin=str(Path(executable).absolute()) if executable else cli_bin,
                 app_id=app_id,
                 consumer_lease_key=self._consumer_lease_key(profile, profile_config),
                 callback_chats=callback_chats,
diff --git a/loopx/extensions/lark/private_conversations.py b/loopx/extensions/lark/private_conversations.py
index e9a942eef8..2419fbc1c6 100644
--- a/loopx/extensions/lark/private_conversations.py
+++ b/loopx/extensions/lark/private_conversations.py
@@ -266,9 +266,9 @@ def profiles(self) -> dict[str, dict[str, str]]:
                 "binding_id": row["binding_id"], "consumer_ref": hashlib.sha256(app_id.encode("utf-8")).hexdigest()[:32]}
         return profiles
 
-    def configured_profile_apps(self) -> dict[str, str]:
+    def configured_profile_apps(self, *, cli_bin: str | None = None) -> dict[str, str]:
         """Read local App identities for listener leases, without granting access."""
-        result = call(self.runner, [self.cli_bin, "profile", "list"])
+        result = call(self.runner, [cli_bin or self.cli_bin, "profile", "list"])
         try:
             configured = json.loads(str(result.get("stdout") or ""))
         except json.JSONDecodeError as exc:
diff --git a/tests/test_lark_private_ingress.py b/tests/test_lark_private_ingress.py
index b6024f15ef..b02bf57d93 100644
--- a/tests/test_lark_private_ingress.py
+++ b/tests/test_lark_private_ingress.py
@@ -348,12 +348,16 @@ def isolated_stream(**kwargs):
     (True, "replace"), (True, "add"), (True, "remove"),
     (False, "replace"), (False, "add"), (False, "remove"),
     (False, "add_after_retarget"), (False, "remove_after_retarget"),
+    (False, "add_cli_alias"), (False, "remove_cli_alias"),
+    (False, "add_different_cli"), (False, "remove_different_cli"),
 ])
 def test_shared_listener_drains_group_inbox_when_private_workspace_changes(
     ordinary,  # noqa: F811
     tmp_path, monkeypatch, stored_app, binding_change,
 ):
     import http.client
+    import os
+    import sys
     from pathlib import Path
 
     from extensions.test_lark_goal_topic_runtime import _reply_runner, _seed_legacy_topic
@@ -378,11 +382,20 @@ def test_shared_listener_drains_group_inbox_when_private_workspace_changes(
         identity["bot_app_id"] = original_app
     else:
         identity.pop("bot_app_id")
+    if binding_change.endswith(("_cli_alias", "_different_cli")):
+        transport.cli_bin = sys.executable
+        if binding_change.endswith("_cli_alias"):
+            # Legacy targets may save a PATH command while Chat resolves the
+            # same executable to a full path for its private transport.
+            executable = Path(sys.executable)
+            monkeypatch.setenv("PATH", str(executable.parent) + os.pathsep + os.environ.get("PATH", ""))
+            identity["cli_bin"] = executable.name
     bindings = {"goal-alpha": read_goal_channel_binding(binding_path)}
     before = transport.profiles()[profile]["binding_id"]
     adding = binding_change.startswith("add")
     removing = binding_change.startswith("remove")
     retargeted = binding_change.endswith("_retarget")
+    different_transport = retargeted or binding_change.endswith("_different_cli")
     if adding:
         transport.bindings.disconnect(before, expected_revision=transport.bindings.read()["revision"])
 
@@ -465,7 +478,7 @@ def observe_health(update):
     server.registry_path, server.runtime_root_override = controller.registry_path, store.root.parent
     server.verbose = False
     server.lark_cli_resolution = LarkCliResolution(
-        command="lark-cli", available=True, source="explicit", version="fixture", error_code=None,
+        command=transport.cli_bin, available=True, source="explicit", version="fixture", error_code=None,
     )
     server.lark_private_conversations = transport
     server.lark_goal_topic_runtime = service
@@ -505,7 +518,7 @@ def change_private_binding(method, body):
         assert len(consumer_processes) == 1
         release.set()
         assert next_stream_ready.wait(8), service.health_snapshot()
-        expected = group_events[:1] if retargeted else group_events[:2]
+        expected = group_events[:1] if different_transport else group_events[:2]
         if removing:
             expected += group_events[2:]
         assert answered == [event["message_id"] for event in expected]
@@ -516,6 +529,8 @@ def change_private_binding(method, body):
         inbox = inspect_lark_event_inbox(project=store.root.parent, config_path=config_path)
         assert inbox["processed_count"] == len(expected)
         assert inbox["pending_count"] == 0
+        if not stored_app and (adding or removing):
+            assert [identity["cli_bin"], "profile", "list"] in provider.calls
     finally:
         release.set()
         service.close()

From 1d05f09273bd40ffe8ee8bc0dc6740b724ffe177 Mon Sep 17 00:00:00 2001
From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com>
Date: Fri, 13 Nov 2026 15:27:29 +0800
Subject: [PATCH 5/8] fix(lark): retain live listener on unreadable profile
 inventory

Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com>
---
 .../lark/goal_topic_runtime_service.py        | 99 +++++++++++++++----
 tests/test_lark_private_ingress.py            | 63 +++++++++---
 2 files changed, 132 insertions(+), 30 deletions(-)

diff --git a/loopx/extensions/lark/goal_topic_runtime_service.py b/loopx/extensions/lark/goal_topic_runtime_service.py
index e7fbe0c913..6964ea360b 100644
--- a/loopx/extensions/lark/goal_topic_runtime_service.py
+++ b/loopx/extensions/lark/goal_topic_runtime_service.py
@@ -81,6 +81,7 @@ def __init__(
         self._closed = threading.Event()
         self._delivery_wakeup = threading.Event()
         self._startup_thread: threading.Thread | None = None
+        self._refresh_retry: threading.Timer | None = None
         self._delivery_thread: threading.Thread | None = None
         self._delivery_worker = HostedWorker(target=self._reconcile_private, stop=self._closed,
             name="loopx-lark-private-delivery", retry_seconds=5)
@@ -286,6 +287,27 @@ def _refresh_on_start(self) -> None:
                 )
             self._closed.wait(5)
 
+    def _retry_profile_refresh(self) -> None:
+        try:
+            unavailable = self._refresh_profile_workers(resume_queues=False)
+        except (OSError, LarkCliUnavailableError):
+            unavailable = True
+        except Exception:
+            logging.getLogger(__name__).warning("Lark profile refresh failed")
+            unavailable = False
+        with self._lock:
+            self._refresh_retry = None
+            if unavailable:
+                self._schedule_profile_refresh()
+
+    def _schedule_profile_refresh(self) -> None:
+        # Called under the service lock. Only the configuration read is retried.
+        if self._closed.is_set() or self._refresh_retry is not None:
+            return
+        self._refresh_retry = threading.Timer(5, self._retry_profile_refresh)
+        self._refresh_retry.daemon = True
+        self._refresh_retry.start()
+
     @staticmethod
     def _now() -> str:
         return datetime.now(timezone.utc).isoformat()
@@ -343,7 +365,7 @@ def _consumer_lease_key(profile: str, config: Mapping[str, str]) -> str:
         identity = config.get("bot_app_id") or profile
         return str(config.get("consumer_ref") or hashlib.sha256(identity.encode("utf-8")).hexdigest()[:32])
 
-    def _listener_configs(self, snapshot: Mapping[str, Any]) -> dict[str, dict[str, str]]:
+    def _listener_configs(self, snapshot: Mapping[str, Any]) -> tuple[dict[str, dict[str, str]], set[str]]:
         from . import goal_topic_runtime as runtime
 
         profiles = runtime._active_profile_configs(snapshot)
@@ -352,20 +374,23 @@ def _listener_configs(self, snapshot: Mapping[str, Any]) -> dict[str, dict[str,
             if not config.get("bot_app_id") and not profiles[profile].get("bot_app_id")
         }
         inventory = getattr(self.private_conversations, "configured_profile_apps", None)
+        unavailable: set[str] = set()
         if not callable(inventory):
-            return profiles
-        apps_by_cli: dict[str, dict[str, str]] = {}
-        for profile in legacy:
+            return profiles, unavailable
+        apps_by_cli: dict[str, dict[str, str] | None] = {}
+        for profile in sorted(legacy):
             config = profiles[profile]
             cli_bin = config["cli_bin"]
             if cli_bin not in apps_by_cli:
                 try:
                     apps_by_cli[cli_bin] = inventory(cli_bin=cli_bin)
                 except (OSError, LarkCliUnavailableError):
-                    # Keep the ordinary stop path if App continuity cannot
-                    # be established from this group's CLI configuration.
-                    apps_by_cli[cli_bin] = {}
-            app_id = apps_by_cli[cli_bin].get(profile)
+                    apps_by_cli[cli_bin] = None
+            apps = apps_by_cli[cli_bin]
+            if apps is None:
+                unavailable.add(profile)
+                continue
+            app_id = apps.get(profile)
             if app_id:
                 # Fence the original profile and the resolved App on both
                 # sides of a private binding change. Routing keeps the original
@@ -375,7 +400,7 @@ def _listener_configs(self, snapshot: Mapping[str, Any]) -> dict[str, dict[str,
                     "bot_app_id": app_id,
                     "consumer_ref": self._consumer_lease_key(profile, config),
                 }
-        return profiles
+        return profiles, unavailable
 
     def _profile_worker_fingerprint(
         self,
@@ -503,15 +528,27 @@ def handle_review_callback(
                         )
 
                     snapshot = self.snapshot_provider()
-                    profile_config = self._listener_configs(snapshot).get(profile) or {}
+                    profile_configs, unavailable = self._listener_configs(snapshot)
+                    if profile in unavailable:
+                        # A new subprocess must acquire locks for a fresh App
+                        # observation. Only an already running pipe may stay.
+                        self._update_health(
+                            profile, status="retrying", error_code="lark_profile_inventory_unavailable"
+                        )
+                        stop.wait(5)
+                        continue
+                    profile_config = profile_configs.get(profile) or {}
                     private_profiles = snapshot.get("private_profiles")
                     private_binding = private_profiles.get(profile) if isinstance(private_profiles, Mapping) else None
                     binding_id = private_binding.get("binding_id") if isinstance(private_binding, Mapping) else None
                     reconnect_requested = threading.Event()
                     # A group-only connection must not pick up private work
-                    # read before a newly added owner binding.
-                    private_admitter = (self._admit_private if self.private_conversations is not None
-                                        and isinstance(private_binding, Mapping) else None)
+                    # read before a newly added owner binding. A provider with
+                    # no separate group route still owns its private admission.
+                    private_admitter = self._admit_private if self.private_conversations is not None and (
+                        isinstance(private_binding, Mapping)
+                        or profile not in runtime._active_group_profile_configs(snapshot)
+                    ) else None
                     bound_owner = (
                         private_admitter is not None
                         and isinstance(binding_id, str) and bool(binding_id)
@@ -656,10 +693,16 @@ def capture(route: Mapping[str, object]) -> dict[str, object]:
                     self.refresh()
 
     def refresh(self) -> None:
+        if self._refresh_profile_workers():
+            with self._lock:
+                self._schedule_profile_refresh()
+
+    def _refresh_profile_workers(self, *, resume_queues: bool = True) -> bool:
+        """Apply known listener changes and report whether discovery must retry."""
         if self._closed.is_set():
-            return
+            return False
         snapshot = self.snapshot_provider()
-        profile_configs = self._listener_configs(snapshot)
+        profile_configs, unavailable = self._listener_configs(snapshot)
         desired = {
             profile: self._profile_worker_fingerprint(
                 snapshot,
@@ -669,13 +712,29 @@ def refresh(self) -> None:
             for profile, profile_config in profile_configs.items()
         }
         if self._closed.is_set():
-            return
-        self._resume_session_queues(snapshot)
+            return False
+        if resume_queues:
+            self._resume_session_queues(snapshot)
         # A filesystem read may outlive server shutdown (for example, while
         # waiting for OS directory consent). Never start effects after close.
         with self._lock:
             if self._closed.is_set():
-                return
+                return False
+            for profile in unavailable:
+                replacement = desired.pop(profile)
+                worker = self._workers.get(profile)
+                if worker is None:
+                    continue
+                previous = worker[2]
+                # Keep the original pipe and locks while the same group's
+                # local App inventory is unavailable. Routing still checks
+                # current grants; an observed removal uses the stop path.
+                if (
+                    previous.transport.cli_bin == replacement.transport.cli_bin
+                    and previous.transport.callback_chats == replacement.transport.callback_chats
+                    and not previous.transport.other_binding_id
+                ):
+                    desired[profile] = previous
             stale = (set(self._workers) - set(desired)) | {
                 profile
                 for profile in set(self._workers) & set(desired)
@@ -719,6 +778,7 @@ def refresh(self) -> None:
                 )
                 self._workers[profile] = (stop, thread, desired[profile])
                 thread.start()
+        return bool(unavailable)
 
     def _resume_session_queues(self, snapshot: Mapping[str, Any]) -> None:
         binding_payloads = snapshot.get("binding_payloads")
@@ -770,6 +830,9 @@ def close(self) -> None:
         self._closed.set()
         self._delivery_wakeup.set()
         with self._lock:
+            if self._refresh_retry is not None:
+                self._refresh_retry.cancel()
+                self._refresh_retry = None
             workers = list(self._workers.values())
             self._workers.clear()
         for stop, _thread, _fingerprint in workers:
diff --git a/tests/test_lark_private_ingress.py b/tests/test_lark_private_ingress.py
index b02bf57d93..104d4b79b7 100644
--- a/tests/test_lark_private_ingress.py
+++ b/tests/test_lark_private_ingress.py
@@ -350,6 +350,7 @@ def isolated_stream(**kwargs):
     (False, "add_after_retarget"), (False, "remove_after_retarget"),
     (False, "add_cli_alias"), (False, "remove_cli_alias"),
     (False, "add_different_cli"), (False, "remove_different_cli"),
+    (False, "profile_read_fails"), (False, "profile_read_fails_then_retargets"),
 ])
 def test_shared_listener_drains_group_inbox_when_private_workspace_changes(
     ordinary,  # noqa: F811
@@ -396,8 +397,13 @@ def test_shared_listener_drains_group_inbox_when_private_workspace_changes(
     removing = binding_change.startswith("remove")
     retargeted = binding_change.endswith("_retarget")
     different_transport = retargeted or binding_change.endswith("_different_cli")
+    unreadable = binding_change.startswith("profile_read_fails")
+    retarget_on_retry = binding_change == "profile_read_fails_then_retargets"
     if adding:
         transport.bindings.disconnect(before, expected_revision=transport.bindings.read()["revision"])
+    if unreadable:
+        for row in transport.bindings.read()["bindings"]:
+            transport.bindings.disconnect(row["binding_id"], expected_revision=transport.bindings.read()["revision"])
 
     group_config = {"target_payload": targets, "binding_payloads": bindings}
 
@@ -412,6 +418,7 @@ def snapshot():
     ]
     new_private = provider.event(profile, "after_rebind", "/status")
     in_group_answer, release, queued, next_stream_ready = (threading.Event() for _ in range(4))
+    second_replied, inventory_retried = threading.Event(), threading.Event()
     answered, consumer_processes = [], []
     replies = {}
 
@@ -439,7 +446,7 @@ def first_connection():
         queued.set()
 
     def replacement_connection():
-        yield group_events[2] if removing else new_private
+        yield group_events[2] if removing or unreadable else new_private
 
     def spawn(args):
         assert args[3:6] == ["event", "consume", "im.message.receive_v1"]
@@ -457,8 +464,11 @@ def isolated_stream(**kwargs):
 
         def observe_health(update):
             health_sink(update)
-            if removing and len(consumer_processes) == 2 and update.get("replied_count") == 1:
-                next_stream_ready.set()
+            if update.get("replied_count") == 1:
+                if (removing or unreadable) and len(consumer_processes) == 2:
+                    next_stream_ready.set()
+                if answered == [event["message_id"] for event in group_events[:2]]:
+                    second_replied.set()
 
         return stream(**{**kwargs, "health_sink": observe_health}, reply_runner=_reply_runner(replies))
 
@@ -499,9 +509,28 @@ def change_private_binding(method, body):
     try:
         service.refresh()
         assert queued.wait(8), service.health_snapshot()
-        if retargeted:
+        if retargeted or retarget_on_retry:
             provider.profile_apps[profile] = "cli_retargeted_app"
-        if removing:
+        if unreadable:
+            original_runner = transport.runner
+            inventory_failed = threading.Event()
+
+            def read_profile(args, cwd=None, timeout=None):
+                if args[1:] == ["profile", "list"]:
+                    if not inventory_failed.is_set():
+                        inventory_failed.set()
+                        return {"returncode": 1, "stdout": "", "stderr": "temporary profile read failure"}
+                    inventory_retried.set()
+                return original_runner(args, cwd, timeout)
+
+            monkeypatch.setattr(transport, "runner", read_profile)
+            original_worker = service._workers[profile]
+            service.refresh()
+            assert inventory_failed.is_set()
+            assert service._workers[profile] == original_worker
+            assert not original_worker[0].is_set()
+            assert not consumer_processes[0].exited.is_set()
+        elif removing:
             change_private_binding("DELETE", {"binding_id": before, "revision": transport.bindings.read()["revision"]})
         else:
             projects = transport.bindings.projects
@@ -514,13 +543,23 @@ def change_private_binding(method, body):
             replacement = next(row for row in transport.bindings.read()["bindings"] if row["transport_ref"] == profile)
             assert transport.profiles()[profile]["binding_id"] == replacement["binding_id"] != before
             assert replacement["provider_ref"] == identity_ref(provider.profile_apps[profile])
-        assert consumer_processes[0].exited.wait(8)
-        assert len(consumer_processes) == 1
-        release.set()
-        assert next_stream_ready.wait(8), service.health_snapshot()
-        expected = group_events[:1] if different_transport else group_events[:2]
-        if removing:
-            expected += group_events[2:]
+        if unreadable and not retarget_on_retry:
+            release.set()
+            assert second_replied.wait(8), service.health_snapshot()
+            assert inventory_retried.wait(8)
+            assert service._workers[profile] == original_worker
+            assert len(consumer_processes) == 1
+            expected = group_events[:2]
+        else:
+            assert consumer_processes[0].exited.wait(8)
+            assert len(consumer_processes) == 1
+            release.set()
+            assert next_stream_ready.wait(8), service.health_snapshot()
+            expected = group_events[:1] if different_transport or retarget_on_retry else group_events[:2]
+            if removing or unreadable:
+                expected += group_events[2:]
+            if unreadable:
+                assert inventory_retried.is_set()
         assert answered == [event["message_id"] for event in expected]
         reply_targets = [call[call.index("--message-id") + 1] for call in replies["calls"]
                          if "+messages-reply" in call and "--dry-run" not in call]

From a24f23d6d5a2094aeac92a84523fa5844492e6a6 Mon Sep 17 00:00:00 2001
From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com>
Date: Fri, 13 Nov 2026 17:20:04 +0800
Subject: [PATCH 6/8] Ignore superseded Lark listener discovery results

Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com>
---
 .../lark/goal_topic_runtime_service.py        | 22 +++--
 tests/test_lark_private_ingress.py            | 88 +++++++++++++++++--
 2 files changed, 96 insertions(+), 14 deletions(-)

diff --git a/loopx/extensions/lark/goal_topic_runtime_service.py b/loopx/extensions/lark/goal_topic_runtime_service.py
index 6964ea360b..373ad133ef 100644
--- a/loopx/extensions/lark/goal_topic_runtime_service.py
+++ b/loopx/extensions/lark/goal_topic_runtime_service.py
@@ -82,6 +82,8 @@ def __init__(
         self._delivery_wakeup = threading.Event()
         self._startup_thread: threading.Thread | None = None
         self._refresh_retry: threading.Timer | None = None
+        self._profile_refresh_sequence = 0
+        self._applied_profile_refresh_sequence = 0
         self._delivery_thread: threading.Thread | None = None
         self._delivery_worker = HostedWorker(target=self._reconcile_private, stop=self._closed,
             name="loopx-lark-private-delivery", retry_seconds=5)
@@ -289,15 +291,15 @@ def _refresh_on_start(self) -> None:
 
     def _retry_profile_refresh(self) -> None:
         try:
-            unavailable = self._refresh_profile_workers(resume_queues=False)
+            retry_needed = self._refresh_profile_workers(resume_queues=False)
         except (OSError, LarkCliUnavailableError):
-            unavailable = True
+            retry_needed = True
         except Exception:
             logging.getLogger(__name__).warning("Lark profile refresh failed")
-            unavailable = False
+            retry_needed = False
         with self._lock:
             self._refresh_retry = None
-            if unavailable:
+            if retry_needed:
                 self._schedule_profile_refresh()
 
     def _schedule_profile_refresh(self) -> None:
@@ -699,8 +701,11 @@ def refresh(self) -> None:
 
     def _refresh_profile_workers(self, *, resume_queues: bool = True) -> bool:
         """Apply known listener changes and report whether discovery must retry."""
-        if self._closed.is_set():
-            return False
+        with self._lock:
+            if self._closed.is_set():
+                return False
+            self._profile_refresh_sequence += 1
+            sequence = self._profile_refresh_sequence
         snapshot = self.snapshot_provider()
         profile_configs, unavailable = self._listener_configs(snapshot)
         desired = {
@@ -720,6 +725,11 @@ def _refresh_profile_workers(self, *, resume_queues: bool = True) -> bool:
         with self._lock:
             if self._closed.is_set():
                 return False
+            if sequence < self._applied_profile_refresh_sequence:
+                # A newer refresh already owns the listener configuration.
+                # Recheck with the existing timer before changing any worker.
+                return True
+            self._applied_profile_refresh_sequence = sequence
             for profile in unavailable:
                 replacement = desired.pop(profile)
                 worker = self._workers.get(profile)
diff --git a/tests/test_lark_private_ingress.py b/tests/test_lark_private_ingress.py
index 104d4b79b7..8f976b440a 100644
--- a/tests/test_lark_private_ingress.py
+++ b/tests/test_lark_private_ingress.py
@@ -195,7 +195,8 @@ def test_stale_or_unreadable_health_is_not_verified(source_monitor):
 class _ListenerProcess:
     """Yield fixture events and let the real service end the stream."""
 
-    def __init__(self, events):
+    def __init__(self, events, *, hold_seconds=8):
+        self.hold_seconds = hold_seconds
         self.exited = threading.Event()
         self.waited = threading.Event()
         self.stdout = self.lines(events)
@@ -204,7 +205,7 @@ def lines(self, events):
         yield "[event] ready event_key=im.message.receive_v1\n"
         for event in events:
             yield json.dumps(event)
-        assert self.exited.wait(8)
+        assert self.exited.wait(self.hold_seconds)
         yield "[event] exited (reason: signal)\n"
 
     def poll(self):
@@ -351,6 +352,8 @@ def isolated_stream(**kwargs):
     (False, "add_cli_alias"), (False, "remove_cli_alias"),
     (False, "add_different_cli"), (False, "remove_different_cli"),
     (False, "profile_read_fails"), (False, "profile_read_fails_then_retargets"),
+    (False, "stale_inventory"), (False, "stale_inventory_after_error"),
+    (False, "removed_inventory"), (False, "closed_during_inventory"),
 ])
 def test_shared_listener_drains_group_inbox_when_private_workspace_changes(
     ordinary,  # noqa: F811
@@ -366,7 +369,7 @@ def test_shared_listener_drains_group_inbox_when_private_workspace_changes(
     from loopx.extensions.lark import goal_topic_runtime as runtime
     from loopx.extensions.lark.cli_resolution import LarkCliResolution
     from loopx.extensions.lark.conversation_identity import identity_ref
-    from loopx.extensions.lark.event_inbox import inspect_lark_event_inbox
+    from loopx.extensions.lark.event_inbox import inspect_lark_event_inbox, load_lark_event_inbox_config
     from loopx.extensions.lark.goal_channel_contracts import read_goal_channel_binding
     from loopx.extensions.lark.goal_channel_targets import read_goal_channel_targets
     from test_lark_private_conversations import connect
@@ -397,7 +400,11 @@ def test_shared_listener_drains_group_inbox_when_private_workspace_changes(
     removing = binding_change.startswith("remove")
     retargeted = binding_change.endswith("_retarget")
     different_transport = retargeted or binding_change.endswith("_different_cli")
-    unreadable = binding_change.startswith("profile_read_fails")
+    stale_inventory = binding_change in {"stale_inventory", "stale_inventory_after_error"}
+    removed_inventory = binding_change == "removed_inventory"
+    closing = binding_change == "closed_during_inventory"
+    delayed_inventory = stale_inventory or removed_inventory or closing
+    unreadable = binding_change.startswith("profile_read_fails") or delayed_inventory
     retarget_on_retry = binding_change == "profile_read_fails_then_retargets"
     if adding:
         transport.bindings.disconnect(before, expected_revision=transport.bindings.read()["revision"])
@@ -419,6 +426,8 @@ def snapshot():
     new_private = provider.event(profile, "after_rebind", "/status")
     in_group_answer, release, queued, next_stream_ready = (threading.Event() for _ in range(4))
     second_replied, inventory_retried = threading.Event(), threading.Event()
+    inventory_captured, return_inventory = threading.Event(), threading.Event()
+    timer_reads = []
     answered, consumer_processes = [], []
     replies = {}
 
@@ -446,13 +455,17 @@ def first_connection():
         queued.set()
 
     def replacement_connection():
-        yield group_events[2] if removing or unreadable else new_private
+        if not removed_inventory and not closing:
+            yield group_events[2] if removing or unreadable else new_private
 
     def spawn(args):
         assert args[3:6] == ["event", "consume", "im.message.receive_v1"]
         if consumer_processes:
             assert len(consumer_processes) == 1 and consumer_processes[0].waited.is_set()
-        consumer = _ListenerProcess(replacement_connection() if consumer_processes else first_connection())
+        consumer = _ListenerProcess(
+            replacement_connection() if consumer_processes else first_connection(),
+            hold_seconds=22 if delayed_inventory else 8,
+        )
         consumer_processes.append(consumer)
         return consumer
 
@@ -514,13 +527,28 @@ def change_private_binding(method, body):
         if unreadable:
             original_runner = transport.runner
             inventory_failed = threading.Event()
+            unreadable_inventory = {"returncode": 1, "stdout": "", "stderr": "temporary profile read failure"}
 
             def read_profile(args, cwd=None, timeout=None):
                 if args[1:] == ["profile", "list"]:
                     if not inventory_failed.is_set():
                         inventory_failed.set()
-                        return {"returncode": 1, "stdout": "", "stderr": "temporary profile read failure"}
+                        return unreadable_inventory
                     inventory_retried.set()
+                    if delayed_inventory and threading.current_thread() is service._refresh_retry:
+                        timer_reads.append(threading.current_thread())
+                        if not inventory_captured.is_set():
+                            if removed_inventory:
+                                provider.profile_apps.pop(profile)
+                            response = original_runner(args, cwd, timeout)
+                            if stale_inventory:
+                                rows = json.loads(response["stdout"])
+                                response["stdout"] = json.dumps([row for row in rows if row["name"] != profile])
+                            inventory_captured.set()
+                            assert return_inventory.wait(8)
+                            return response
+                    if binding_change == "stale_inventory_after_error" and not return_inventory.is_set():
+                        return unreadable_inventory
                 return original_runner(args, cwd, timeout)
 
             monkeypatch.setattr(transport, "runner", read_profile)
@@ -543,7 +571,47 @@ def read_profile(args, cwd=None, timeout=None):
             replacement = next(row for row in transport.bindings.read()["bindings"] if row["transport_ref"] == profile)
             assert transport.profiles()[profile]["binding_id"] == replacement["binding_id"] != before
             assert replacement["provider_ref"] == identity_ref(provider.profile_apps[profile])
-        if unreadable and not retarget_on_retry:
+        if delayed_inventory:
+            retry = service._refresh_retry
+            assert retry is not None and inventory_captured.wait(8)
+            if stale_inventory:
+                service.refresh()
+                assert service._refresh_retry is retry
+                assert service._workers[profile] == original_worker
+            if closing:
+                release.set()
+                assert second_replied.wait(8), service.health_snapshot()
+                service.close()
+            return_inventory.set()
+            retry.join(8)
+            assert not retry.is_alive()
+            if removed_inventory:
+                assert original_worker[0].is_set()
+                assert consumer_processes[0].exited.wait(8)
+                release.set()
+                original_worker[1].join(8)
+                assert not original_worker[1].is_alive()
+                expected = group_events[:1]
+            else:
+                if stale_inventory:
+                    assert not original_worker[0].is_set()
+                    release.set()
+                    assert second_replied.wait(8), service.health_snapshot()
+                    with service._lock:
+                        recheck = service._refresh_retry
+                    assert recheck is not None and recheck is not retry
+                    recheck.join(8)
+                    assert not recheck.is_alive()
+                    assert timer_reads == [retry, recheck]
+                    assert service._workers[profile] == original_worker
+                else:
+                    assert original_worker[0].is_set()
+                    assert service.active_profiles() == []
+                    assert timer_reads == [retry]
+                assert service._refresh_retry is None
+                assert len(consumer_processes) == 1
+                expected = group_events[:2]
+        elif unreadable and not retarget_on_retry:
             release.set()
             assert second_replied.wait(8), service.health_snapshot()
             assert inventory_retried.wait(8)
@@ -566,11 +634,15 @@ def read_profile(args, cwd=None, timeout=None):
         assert reply_targets == answered
         config_path, = (store.root.parent / ".loopx/config/lark-goal-topics").glob("*.json")
         inbox = inspect_lark_event_inbox(project=store.root.parent, config_path=config_path)
+        config = load_lark_event_inbox_config(project=store.root.parent, config_path=config_path)
+        acked = json.loads((config["inbox_path"] / "processed.json").read_text())
+        assert set(acked["message_ids"]) == set(reply_targets)
         assert inbox["processed_count"] == len(expected)
         assert inbox["pending_count"] == 0
         if not stored_app and (adding or removing):
             assert [identity["cli_bin"], "profile", "list"] in provider.calls
     finally:
+        return_inventory.set()
         release.set()
         service.close()
         server.shutdown()

From f957afcb2af5675b96257496f72c4bdfcd74b31f Mon Sep 17 00:00:00 2001
From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com>
Date: Fri, 13 Nov 2026 17:39:45 +0800
Subject: [PATCH 7/8] Retain Lark discovery retry requests during timer
 completion

Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com>
---
 .../lark/goal_topic_runtime_service.py        | 11 ++-
 tests/test_lark_private_ingress.py            | 75 +++++++++++++++++++
 2 files changed, 84 insertions(+), 2 deletions(-)

diff --git a/loopx/extensions/lark/goal_topic_runtime_service.py b/loopx/extensions/lark/goal_topic_runtime_service.py
index 373ad133ef..6cdd355bf3 100644
--- a/loopx/extensions/lark/goal_topic_runtime_service.py
+++ b/loopx/extensions/lark/goal_topic_runtime_service.py
@@ -82,6 +82,7 @@ def __init__(
         self._delivery_wakeup = threading.Event()
         self._startup_thread: threading.Thread | None = None
         self._refresh_retry: threading.Timer | None = None
+        self._refresh_retry_requested = False
         self._profile_refresh_sequence = 0
         self._applied_profile_refresh_sequence = 0
         self._delivery_thread: threading.Thread | None = None
@@ -290,6 +291,8 @@ def _refresh_on_start(self) -> None:
             self._closed.wait(5)
 
     def _retry_profile_refresh(self) -> None:
+        with self._lock:
+            self._refresh_retry_requested = False
         try:
             retry_needed = self._refresh_profile_workers(resume_queues=False)
         except (OSError, LarkCliUnavailableError):
@@ -299,12 +302,16 @@ def _retry_profile_refresh(self) -> None:
             retry_needed = False
         with self._lock:
             self._refresh_retry = None
-            if retry_needed:
+            if retry_needed or self._refresh_retry_requested:
                 self._schedule_profile_refresh()
 
     def _schedule_profile_refresh(self) -> None:
         # Called under the service lock. Only the configuration read is retried.
-        if self._closed.is_set() or self._refresh_retry is not None:
+        if self._closed.is_set():
+            return
+        if self._refresh_retry is not None:
+            # A foreground failure can arrive while the timer is finishing.
+            self._refresh_retry_requested = True
             return
         self._refresh_retry = threading.Timer(5, self._retry_profile_refresh)
         self._refresh_retry.daemon = True
diff --git a/tests/test_lark_private_ingress.py b/tests/test_lark_private_ingress.py
index 8f976b440a..ce00b6f7d2 100644
--- a/tests/test_lark_private_ingress.py
+++ b/tests/test_lark_private_ingress.py
@@ -651,6 +651,81 @@ def read_profile(args, cwd=None, timeout=None):
         controller.close()
 
 
+def test_foreground_inventory_failure_rearms_a_finishing_profile_retry(tmp_path, monkeypatch):
+    from extensions.test_lark_goal_topic_runtime import _seed_legacy_topic
+    from loopx.extensions.lark import goal_topic_runtime as runtime
+    from loopx.extensions.lark.goal_channel_contracts import read_goal_channel_binding
+    from loopx.extensions.lark.goal_channel_targets import read_goal_channel_targets
+
+    target_path, binding_path = tmp_path / "targets.json", tmp_path / "binding.json"
+    _seed_legacy_topic(target_path, binding_path)
+    targets = read_goal_channel_targets(target_path)
+    identity = targets["targets"]["fixture"]["identity"]
+    profile = identity["sender_profile"]
+    app_id = identity.pop("bot_app_id", None) or "cli_public_fixture"
+    snapshot = {
+        "target_payload": targets,
+        "binding_payloads": {"goal-alpha": read_goal_channel_binding(binding_path)},
+    }
+    available = threading.Event()
+    available.set()
+    observed, finish = threading.Event(), threading.Event()
+    timer_reads = []
+
+    def inventory(*, cli_bin):
+        if threading.current_thread() is service._refresh_retry:
+            timer_reads.append(threading.current_thread())
+        if not available.is_set():
+            raise OSError("profile inventory unavailable")
+        return {profile: app_id}
+
+    service = runtime.LarkGoalTopicRuntimeService(
+        snapshot_provider=lambda: snapshot,
+        runtime_root=tmp_path,
+        runtime_controller=object(),
+        private_conversations=SimpleNamespace(configured_profile_apps=inventory),
+        profile_poller=lambda _profile, stop: stop.wait(20),
+    )
+    original_refresh = service._refresh_profile_workers
+
+    def pause_after_observation(*, resume_queues=True):
+        retry_needed = original_refresh(resume_queues=resume_queues)
+        if not resume_queues and not observed.is_set():
+            assert not retry_needed
+            observed.set()
+            assert finish.wait(8)
+        return retry_needed
+
+    monkeypatch.setattr(service, "_refresh_profile_workers", pause_after_observation)
+    try:
+        service.refresh()
+        worker = service._workers[profile]
+        available.clear()
+        service.refresh()
+        retry = service._refresh_retry
+        assert retry is not None
+        available.set()
+        assert observed.wait(8)
+        available.clear()
+        service.refresh()
+        assert service._refresh_retry is retry
+        assert service._workers[profile] == worker
+        available.set()
+        finish.set()
+        retry.join(8)
+        assert not retry.is_alive()
+        recheck = service._refresh_retry
+        assert recheck is not None and recheck is not retry
+        recheck.join(8)
+        assert not recheck.is_alive()
+        assert service._refresh_retry is None
+        assert service._workers[profile] == worker
+        assert timer_reads == [retry, recheck]
+    finally:
+        finish.set()
+        service.close()
+
+
 def test_unknown_failed_capture_is_not_replayed(source_monitor):
     f = source_monitor
     native = {**f.carrier, "status": "failed", "event": {**f.carrier["event"], "message_id": "om_missed"}}

From 20a2b6e747999cc36f995e7c9a654decf19d592d Mon Sep 17 00:00:00 2001
From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com>
Date: Fri, 13 Nov 2026 18:14:07 +0800
Subject: [PATCH 8/8] Fence stale Lark inventory during queue recovery

Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com>
---
 .../lark/goal_topic_runtime_service.py        | 16 ++++++++++-----
 tests/test_lark_private_ingress.py            | 20 +++++++++++++++++--
 2 files changed, 29 insertions(+), 7 deletions(-)

diff --git a/loopx/extensions/lark/goal_topic_runtime_service.py b/loopx/extensions/lark/goal_topic_runtime_service.py
index 6cdd355bf3..b79571b4d6 100644
--- a/loopx/extensions/lark/goal_topic_runtime_service.py
+++ b/loopx/extensions/lark/goal_topic_runtime_service.py
@@ -84,7 +84,7 @@ def __init__(
         self._refresh_retry: threading.Timer | None = None
         self._refresh_retry_requested = False
         self._profile_refresh_sequence = 0
-        self._applied_profile_refresh_sequence = 0
+        self._observed_profile_refresh_sequence = 0
         self._delivery_thread: threading.Thread | None = None
         self._delivery_worker = HostedWorker(target=self._reconcile_private, stop=self._closed,
             name="loopx-lark-private-delivery", retry_seconds=5)
@@ -715,6 +715,14 @@ def _refresh_profile_workers(self, *, resume_queues: bool = True) -> bool:
             sequence = self._profile_refresh_sequence
         snapshot = self.snapshot_provider()
         profile_configs, unavailable = self._listener_configs(snapshot)
+        with self._lock:
+            if self._closed.is_set():
+                return False
+            # Record discovery before queue recovery can block, so an older
+            # inventory retry cannot stop a worker that still has queued events.
+            self._observed_profile_refresh_sequence = max(
+                self._observed_profile_refresh_sequence, sequence
+            )
         desired = {
             profile: self._profile_worker_fingerprint(
                 snapshot,
@@ -732,11 +740,9 @@ def _refresh_profile_workers(self, *, resume_queues: bool = True) -> bool:
         with self._lock:
             if self._closed.is_set():
                 return False
-            if sequence < self._applied_profile_refresh_sequence:
-                # A newer refresh already owns the listener configuration.
-                # Recheck with the existing timer before changing any worker.
+            if sequence < self._observed_profile_refresh_sequence:
+                # Discovery can advance while this refresh is resuming queues.
                 return True
-            self._applied_profile_refresh_sequence = sequence
             for profile in unavailable:
                 replacement = desired.pop(profile)
                 worker = self._workers.get(profile)
diff --git a/tests/test_lark_private_ingress.py b/tests/test_lark_private_ingress.py
index ce00b6f7d2..9b483fffd8 100644
--- a/tests/test_lark_private_ingress.py
+++ b/tests/test_lark_private_ingress.py
@@ -353,6 +353,7 @@ def isolated_stream(**kwargs):
     (False, "add_different_cli"), (False, "remove_different_cli"),
     (False, "profile_read_fails"), (False, "profile_read_fails_then_retargets"),
     (False, "stale_inventory"), (False, "stale_inventory_after_error"),
+    (False, "stale_inventory_during_queue_recovery"),
     (False, "removed_inventory"), (False, "closed_during_inventory"),
 ])
 def test_shared_listener_drains_group_inbox_when_private_workspace_changes(
@@ -400,7 +401,8 @@ def test_shared_listener_drains_group_inbox_when_private_workspace_changes(
     removing = binding_change.startswith("remove")
     retargeted = binding_change.endswith("_retarget")
     different_transport = retargeted or binding_change.endswith("_different_cli")
-    stale_inventory = binding_change in {"stale_inventory", "stale_inventory_after_error"}
+    stale_inventory = binding_change.startswith("stale_inventory")
+    inventory_during_queue_recovery = binding_change == "stale_inventory_during_queue_recovery"
     removed_inventory = binding_change == "removed_inventory"
     closing = binding_change == "closed_during_inventory"
     delayed_inventory = stale_inventory or removed_inventory or closing
@@ -574,9 +576,23 @@ def read_profile(args, cwd=None, timeout=None):
         if delayed_inventory:
             retry = service._refresh_retry
             assert retry is not None and inventory_captured.wait(8)
+            if inventory_during_queue_recovery:
+                resume_session_queues = service._resume_session_queues
+
+                def resume_after_retry(snapshot):
+                    # The foreground has the current inventory but has yet
+                    # to update the listener. Finish the older read here.
+                    return_inventory.set()
+                    retry.join(8)
+                    assert not retry.is_alive()
+                    assert not original_worker[0].is_set()
+                    resume_session_queues(snapshot)
+
+                monkeypatch.setattr(service, "_resume_session_queues", resume_after_retry)
             if stale_inventory:
                 service.refresh()
-                assert service._refresh_retry is retry
+                if not inventory_during_queue_recovery:
+                    assert service._refresh_retry is retry
                 assert service._workers[profile] == original_worker
             if closing:
                 release.set()