diff --git a/loopx/extensions/lark/goal_topic_runtime_service.py b/loopx/extensions/lark/goal_topic_runtime_service.py index 203374f6ad..48ea2c2ce5 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,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, 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() - if discovery is not None and discovery.done(): - try: - for path in discovery.result(): - attempted.setdefault(path, 0) - 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() @@ -161,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() @@ -168,14 +173,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()} + # 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: - 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])) + 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 +197,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..9c2ec18775 100644 --- a/tests/test_lark_private_conversations.py +++ b/tests/test_lark_private_conversations.py @@ -5940,6 +5940,101 @@ def drive(step, **_): assert sum(path.name.startswith("old-") for path in started) == slots * 2 - 1 +@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 + 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, recovery_times = [], [], [] + 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") + recovery_times.append(clock[0]) + 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 + 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, 102, 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) + if body_scan_fails: + assert recovery_times == [102.0] + 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