diff --git a/loopx/extensions/lark/goal_topic_runtime_service.py b/loopx/extensions/lark/goal_topic_runtime_service.py index a6a260fd30..b79571b4d6 100644 --- a/loopx/extensions/lark/goal_topic_runtime_service.py +++ b/loopx/extensions/lark/goal_topic_runtime_service.py @@ -5,10 +5,13 @@ 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 from pathlib import Path +import shutil +from subprocess import Popen import threading import time from typing import Any @@ -16,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 ( @@ -28,8 +32,25 @@ 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] class LarkGoalTopicRuntimeService: @@ -55,10 +76,15 @@ 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() 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._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) @@ -128,14 +154,14 @@ 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 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 @@ -241,10 +267,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. @@ -262,6 +290,33 @@ 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): + retry_needed = True + except Exception: + logging.getLogger(__name__).warning("Lark profile refresh failed") + retry_needed = False + with self._lock: + self._refresh_retry = None + 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(): + 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 + self._refresh_retry.start() + @staticmethod def _now() -> str: return datetime.now(timezone.utc).isoformat() @@ -297,6 +352,65 @@ 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 _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) + 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) + unavailable: set[str] = set() + if not callable(inventory): + 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): + 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 + # snapshot and its separate group and owner grants. + profiles[profile] = { + **config, + "bot_app_id": app_id, + "consumer_ref": self._consumer_lease_key(profile, config), + } + return profiles, unavailable + def _profile_worker_fingerprint( self, snapshot: Mapping[str, Any], @@ -308,11 +422,22 @@ def _profile_worker_fingerprint( if self.action_service is not None else () ) - return ( - str(profile_config.get("cli_bin") or "lark-cli"), - str(profile_config.get("bot_app_id") or profile_config.get("binding_id") or ""), - str(profile_config.get("consumer_ref") or ""), - callback_chats, + 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( + # 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, + 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: @@ -411,21 +536,58 @@ def handle_review_callback( profile_config=profile_config, ) - profile_config = ( - runtime._active_profile_configs(self.snapshot_provider()).get( - profile + snapshot = self.snapshot_provider() + 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" ) - or {} + 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. 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) + 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. 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} + # 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]) # A legacy/shared consumer holds both identities: wait for @@ -447,28 +609,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 self.private_conversations is not None 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 +648,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, ) @@ -532,12 +702,27 @@ def capture(route: Mapping[str, object]) -> dict[str, object]: self.refresh() def refresh(self) -> None: - from . import goal_topic_runtime as runtime + if self._refresh_profile_workers(): + with self._lock: + self._schedule_profile_refresh() - if self._closed.is_set(): - return + def _refresh_profile_workers(self, *, resume_queues: bool = True) -> bool: + """Apply known listener changes and report whether discovery must retry.""" + 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 = runtime._active_profile_configs(snapshot) + 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, @@ -547,20 +732,54 @@ 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 + if sequence < self._observed_profile_refresh_sequence: + # Discovery can advance while this refresh is resuming queues. + return True + 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) 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): @@ -582,6 +801,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") @@ -633,6 +853,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/loopx/extensions/lark/private_conversations.py b/loopx/extensions/lark/private_conversations.py index a84c0687f8..2419fbc1c6 100644 --- a/loopx/extensions/lark/private_conversations.py +++ b/loopx/extensions/lark/private_conversations.py @@ -254,7 +254,21 @@ def profiles(self) -> dict[str, dict[str, str]]: bindings = self.bindings.read()["bindings"] if not bindings: return {} - result = call(self.runner, [self.cli_bin, "profile", "list"]) + 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, *, cli_bin: str | None = None) -> dict[str, str]: + """Read local App identities for listener leases, without granting access.""" + 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: @@ -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]]: @@ -822,10 +827,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..9b483fffd8 100644 --- a/tests/test_lark_private_ingress.py +++ b/tests/test_lark_private_ingress.py @@ -192,37 +192,554 @@ 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 - +class _ListenerProcess: + """Yield fixture events and let the real service end the stream.""" + + 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) + + 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(self.hold_seconds) + 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 + 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) + 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) + + def first_connection(): + yield first + assert in_admission.wait(8) + yield second + waiting.set() # The reader advanced past submission of the second line. + + 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 = _ListenerProcess(replacement_connection() if consumers else first_connection()) + 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 + + +@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"), + (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, "stale_inventory_during_queue_recovery"), + (False, "removed_inventory"), (False, "closed_during_inventory"), +]) +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 - monkeypatch.setattr(service._delivery_wakeup, "wait", wait) - service._reconcile_private() - assert observed == [True] and own.is_set() and not other.is_set() + 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, 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 + + 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) + 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") + 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") + 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 + 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"]) + 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} + + def snapshot(): + 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", + 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)) + second_replied, inventory_retried = threading.Event(), threading.Event() + inventory_captured, return_inventory = threading.Event(), threading.Event() + timer_reads = [] + 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" + + original_admit = transport.admit + + def 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() + return result + + def first_connection(): + yield group_events[0] + assert in_group_answer.wait(8) + yield group_events[1] + queued.set() + + def replacement_connection(): + 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(), + hold_seconds=22 if delayed_inventory else 8, + ) + 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 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)) + + 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) + 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=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=transport.cli_bin, 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 retargeted or retarget_on_retry: + provider.profile_apps[profile] = "cli_retargeted_app" + 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 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) + 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 + 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)) + 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]) + 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() + if not inventory_during_queue_recovery: + 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) + 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] + 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() + server.server_close() + server_thread.join(3) + 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):