From 0b1329a2ea59ac4972c3089fe355e1cf8d72b2e3 Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Mon, 9 Nov 2026 14:25:14 +0800 Subject: [PATCH 1/2] fix(lark): schedule rediscovered private recovery paths Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> --- .../lark/goal_topic_runtime_service.py | 27 +++--- tests/test_lark_private_conversations.py | 82 +++++++++++++++++++ 2 files changed, 95 insertions(+), 14 deletions(-) diff --git a/loopx/extensions/lark/goal_topic_runtime_service.py b/loopx/extensions/lark/goal_topic_runtime_service.py index 203374f6ad..38df7e1f17 100644 --- a/loopx/extensions/lark/goal_topic_runtime_service.py +++ b/loopx/extensions/lark/goal_topic_runtime_service.py @@ -89,6 +89,7 @@ def _reconcile_private(self) -> None: return pending: dict[Path, Future[int]] = {} attempted: dict[Path, int] = {} + ready: dict[Path, None] = {} attention: Future[int] | None = None attention_due = 0.0 reports: Future[int] | None = None @@ -105,14 +106,14 @@ def _reconcile_private(self) -> None: with ThreadPoolExecutor(max_workers=4, thread_name_prefix="loopx-lark-private-reply") as pool, \ (wakeup_scope(self._delivery_wakeup) if callable(wakeup_scope) else nullcontext()): def step() -> None: - nonlocal attempted, attention, attention_due, reports, reports_due, ingress, ingress_due, feedback, feedback_due, discovery, discovery_due + nonlocal attempted, ready, attention, attention_due, reports, reports_due, ingress, ingress_due, feedback, feedback_due, discovery, discovery_due # Clear BEFORE scanning: an arrival during the scan must also # wake the following wait, rather than disappear in that gap. self._delivery_wakeup.clear() + discovered_paths: list[Path] = [] if discovery is not None and discovery.done(): try: - for path in discovery.result(): - attempted.setdefault(path, 0) + discovered_paths = discovery.result() except Exception: logging.getLogger(__name__).warning("Private Chat historical discovery is pending") discovery = None @@ -168,14 +169,13 @@ def step() -> None: # backoff. Retain its fairness count until the source is # actually removed, so earlier hints cannot starve later work. attempted = {path: count for path, count in attempted.items() if path.exists()} - for path in paths: - attempted.setdefault(path, 0) - # A scan is not an attempt. Keep unsubmitted discoveries in - # this same fairness map when the pool is full; discovery's - # read-only backoff must not hide them after a slot opens. - # Reconciliation re-reads the source and Core authority. - paths = list(dict.fromkeys([ - *(path for path, count in attempted.items() if count == 0), *paths])) + # Keep scheduling hints apart from lifetime attempt counts. + # A delivered body or an earlier held recovery can become + # eligible again. Retain the new hint until a slot opens. + ready = {path: None for path in ready if path.exists()} + for path in (*paths, *discovered_paths): + if path not in pending: + ready.setdefault(path, None) owner_attention = getattr(transport, "owner_attention", None) attention_ready = (owner_attention is not None and attention is None and time.monotonic() >= attention_due and owner_attention.has_enabled()) @@ -193,13 +193,12 @@ def step() -> None: # full. Attempts retain fairness within each class. This # local hint never admits/replays a request or skips its # original grant and provider-attempt checks. - for path in sorted(paths, key=lambda path: (recovery_path(path), attempted.get(path, 0))): + for path in sorted(ready, key=lambda path: (recovery_path(path), attempted.get(path, 0))): if (len(pending) + int(feedback is not None) + int(reports is not None or reports_ready) + int(attention is not None or attention_ready) + int(ingress is not None or ingress_ready) + int(discovery is not None) >= 4 or self._closed.is_set()): break - if path in pending: - continue pending[path] = pool.submit(transport.reconcile_request, path) + del ready[path] attempted[path] = attempted.get(path, 0) + 1 if reports_ready and len(pending) + int(feedback is not None) + int(attention is not None or attention_ready) + int(ingress is not None or ingress_ready) + int(discovery is not None) < 4 and not self._closed.is_set(): reports = pool.submit(report_provider.reconcile) diff --git a/tests/test_lark_private_conversations.py b/tests/test_lark_private_conversations.py index 6f6172db56..4f3ffe0816 100644 --- a/tests/test_lark_private_conversations.py +++ b/tests/test_lark_private_conversations.py @@ -5940,6 +5940,88 @@ def drive(step, **_): assert sum(path.name.startswith("old-") for path in started) == slots * 2 - 1 +@pytest.mark.parametrize("initial", ["accepted", "delivered"]) +def test_already_attempted_private_path_is_rediscovered(tmp_path, monkeypatch, initial): + from concurrent.futures import Future + from loopx.chat_store import _atomic_write_json, _read_json + from loopx.extensions.lark import goal_topic_runtime_service as worker + from loopx.extensions.lark import private_conversations as private + + class ImmediateExecutor: + """The worker still sees completed futures at the next scan boundary.""" + def __init__(self, **_): + pass + + def __enter__(self): + return self + + def __exit__(self, *_): + pass + + def submit(self, action, *args, **kwargs): + future = Future() + try: + future.set_result(action(*args, **kwargs)) + except Exception as error: + future.set_exception(error) + return future + + # Same durable path: ordinary body sends first or an earlier native probe + # left no effect. A later full discovery finds eligibility, while the fast + # body scan continues to exclude the now-delivered receipt. + path = tmp_path / "original.json" + record = {"status": initial, "request_ref": "original", "session_id": "session", + "turn_id": "turn", "event": {"message_id": "source"}, "profile": "fixture"} + _atomic_write_json(path, record) + clock = [100.0] + now = SimpleNamespace(monotonic=lambda: clock[0]) + monkeypatch.setattr(worker, "time", now) + monkeypatch.setattr(private, "time", now) + monkeypatch.setattr(worker, "ThreadPoolExecutor", ImmediateExecutor) + attempts, discoveries = [], [] + required_recoveries = 1 if initial == "accepted" else 2 + + def eligible(request_ref): + assert request_ref == record["request_ref"] + discoveries.append(clock[0]) + return attempts.count("recovery") < required_recoveries + + def reconcile(receipt): + assert receipt == path + current = _read_json(receipt) + if current["status"] == "accepted": + attempts.append("body") + _atomic_write_json(receipt, {**current, "status": "delivered"}) + else: + attempts.append("recovery") + transport._context_recovery_probe_at[path] = clock[0] + 60 + return 0 + + transport = private.LarkPrivateConversations.__new__(private.LarkPrivateConversations) + transport.root = tmp_path + transport._delivery_discovery = PrivateDeliveryDiscovery(tmp_path) + transport._body_discovery = PrivateDeliveryDiscovery(tmp_path) + transport._context_recovery_probe_at = {} + transport._web_projection_paths = lambda: [] + transport.reconcile_app_feedback = lambda: 0 + transport.core = SimpleNamespace(context_recovery_observer=lambda: eligible) + transport.reconcile_request = reconcile + service = worker.LarkGoalTopicRuntimeService(snapshot_provider=lambda: {}, runtime_root=tmp_path, + runtime_controller=None, private_conversations=transport) + + def drive(step, **_): + for tick in (100, 101, 161, 162, 222, 223): + clock[0] = float(tick) + step() + + monkeypatch.setattr(service._delivery_worker, "repeat", drive) + service._reconcile_private() + expected = ["body", "recovery"] if initial == "accepted" else ["recovery", "recovery"] + assert len(discoveries) >= 2 + assert attempts == expected, (attempts, discoveries) + assert _read_json(path) == {**record, "status": "delivered"} + + def test_private_recovery_discovery_keeps_negative_and_attempt_backoff(tmp_path, monkeypatch): from loopx.extensions.lark import private_conversations as private From 65067e410f6e0c834381a7a9e08da87f48e3703d Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Mon, 9 Nov 2026 14:35:03 +0800 Subject: [PATCH 2/2] fix(lark): keep recovery hints across body scan errors Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> --- .../lark/goal_topic_runtime_service.py | 24 +++++++++++-------- tests/test_lark_private_conversations.py | 21 ++++++++++++---- 2 files changed, 31 insertions(+), 14 deletions(-) diff --git a/loopx/extensions/lark/goal_topic_runtime_service.py b/loopx/extensions/lark/goal_topic_runtime_service.py index 38df7e1f17..48ea2c2ce5 100644 --- a/loopx/extensions/lark/goal_topic_runtime_service.py +++ b/loopx/extensions/lark/goal_topic_runtime_service.py @@ -106,18 +106,11 @@ def _reconcile_private(self) -> None: with ThreadPoolExecutor(max_workers=4, thread_name_prefix="loopx-lark-private-reply") as pool, \ (wakeup_scope(self._delivery_wakeup) if callable(wakeup_scope) else nullcontext()): def step() -> None: - nonlocal attempted, ready, attention, attention_due, reports, reports_due, ingress, ingress_due, feedback, feedback_due, discovery, discovery_due + nonlocal attempted, ready, attention, attention_due, reports, reports_due + nonlocal ingress, ingress_due, feedback, feedback_due, discovery, discovery_due # Clear BEFORE scanning: an arrival during the scan must also # wake the following wait, rather than disappear in that gap. self._delivery_wakeup.clear() - discovered_paths: list[Path] = [] - if discovery is not None and discovery.done(): - try: - discovered_paths = discovery.result() - except Exception: - logging.getLogger(__name__).warning("Private Chat historical discovery is pending") - discovery = None - discovery_due = time.monotonic() + 5 if feedback is not None and feedback.done(): try: feedback.result() @@ -162,6 +155,17 @@ def step() -> None: future.result() except Exception: logging.getLogger(__name__).warning("Private Chat delivery reconciliation is pending") + if discovery is not None and discovery.done(): + try: + # Keep new hints even if the separate body scan fails. + # A previous attempt only affects their order. + for path in discovery.result(): + if path not in pending: + ready.setdefault(path, None) + except Exception: + logging.getLogger(__name__).warning("Private Chat historical discovery is pending") + discovery = None + discovery_due = time.monotonic() + 5 try: body_paths = getattr(transport, "pending_body_paths", None) paths = body_paths() if callable(body_paths) else transport.pending_delivery_paths() @@ -173,7 +177,7 @@ def step() -> None: # A delivered body or an earlier held recovery can become # eligible again. Retain the new hint until a slot opens. ready = {path: None for path in ready if path.exists()} - for path in (*paths, *discovered_paths): + for path in paths: if path not in pending: ready.setdefault(path, None) owner_attention = getattr(transport, "owner_attention", None) diff --git a/tests/test_lark_private_conversations.py b/tests/test_lark_private_conversations.py index 4f3ffe0816..9c2ec18775 100644 --- a/tests/test_lark_private_conversations.py +++ b/tests/test_lark_private_conversations.py @@ -5940,8 +5940,10 @@ def drive(step, **_): assert sum(path.name.startswith("old-") for path in started) == slots * 2 - 1 -@pytest.mark.parametrize("initial", ["accepted", "delivered"]) -def test_already_attempted_private_path_is_rediscovered(tmp_path, monkeypatch, initial): +@pytest.mark.parametrize(("initial", "body_scan_fails"), [ + ("accepted", False), ("accepted", True), ("delivered", False), +]) +def test_already_attempted_private_path_is_rediscovered(tmp_path, monkeypatch, initial, body_scan_fails): from concurrent.futures import Future from loopx.chat_store import _atomic_write_json, _read_json from loopx.extensions.lark import goal_topic_runtime_service as worker @@ -5978,7 +5980,7 @@ def submit(self, action, *args, **kwargs): monkeypatch.setattr(worker, "time", now) monkeypatch.setattr(private, "time", now) monkeypatch.setattr(worker, "ThreadPoolExecutor", ImmediateExecutor) - attempts, discoveries = [], [] + attempts, discoveries, recovery_times = [], [], [] required_recoveries = 1 if initial == "accepted" else 2 def eligible(request_ref): @@ -5994,6 +5996,7 @@ def reconcile(receipt): _atomic_write_json(receipt, {**current, "status": "delivered"}) else: attempts.append("recovery") + recovery_times.append(clock[0]) transport._context_recovery_probe_at[path] = clock[0] + 60 return 0 @@ -6006,11 +6009,19 @@ def reconcile(receipt): transport.reconcile_app_feedback = lambda: 0 transport.core = SimpleNamespace(context_recovery_observer=lambda: eligible) transport.reconcile_request = reconcile + pending_body_paths = transport.pending_body_paths + + def scan_bodies(): + if body_scan_fails and clock[0] == 101: + raise OSError("temporary body scan failure") + return pending_body_paths() + + transport.pending_body_paths = scan_bodies service = worker.LarkGoalTopicRuntimeService(snapshot_provider=lambda: {}, runtime_root=tmp_path, runtime_controller=None, private_conversations=transport) def drive(step, **_): - for tick in (100, 101, 161, 162, 222, 223): + for tick in (100, 101, 102, 161, 162, 222, 223): clock[0] = float(tick) step() @@ -6019,6 +6030,8 @@ def drive(step, **_): expected = ["body", "recovery"] if initial == "accepted" else ["recovery", "recovery"] assert len(discoveries) >= 2 assert attempts == expected, (attempts, discoveries) + if body_scan_fails: + assert recovery_times == [102.0] assert _read_json(path) == {**record, "status": "delivered"}