Keep rediscovered Lark private recovery work scheduled

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,21 +155,31 @@ 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()
                     # Discovery can temporarily hide a durable request during
                     # 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)

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