diff --git a/loopx/extensions/lark/goal_topic_runtime_service.py b/loopx/extensions/lark/goal_topic_runtime_service.py index 81e6057d25..3a870b07fd 100644 --- a/loopx/extensions/lark/goal_topic_runtime_service.py +++ b/loopx/extensions/lark/goal_topic_runtime_service.py @@ -176,7 +176,13 @@ def step() -> None: and time.monotonic() >= reports_due and report_provider.pending()) if report_provider is not None and reports is None and time.monotonic() >= reports_due: reports_due = time.monotonic() + 30 - for path in sorted(paths, key=lambda path: attempted.get(path, 0)): + recovery_path = getattr(transport, "is_recovery_path", lambda path: False) + # Fresh bodies take the next released slot ahead of old + # recovery probes, including discoveries retained while + # 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))): 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) >= 4 or self._closed.is_set()): break diff --git a/loopx/extensions/lark/private_conversations.py b/loopx/extensions/lark/private_conversations.py index 679fd4e7f5..c7b2723467 100644 --- a/loopx/extensions/lark/private_conversations.py +++ b/loopx/extensions/lark/private_conversations.py @@ -1323,6 +1323,14 @@ def pending_delivery_paths(self) -> list[Path]: paths.append(path) return [*paths, *self._web_projection_paths()] + def is_recovery_path(self, path: Path) -> bool: + """Local scheduling hint; reconciliation still rechecks every effect. + + A verified old notice may need a native recovery probe, but must not + precede an unsent body just because both have zero worker attempts. + """ + return path.parent != self.root / "web-projections" and _read_json(path).get("status") == "delivered" + def _web_projection_paths(self) -> list[Path]: """Discover committed App messages, using the existing reply workers. diff --git a/tests/test_lark_private_conversations.py b/tests/test_lark_private_conversations.py index 0252185a64..135784a273 100644 --- a/tests/test_lark_private_conversations.py +++ b/tests/test_lark_private_conversations.py @@ -5030,6 +5030,82 @@ def drive(step, **_): assert all(transport._context_recovery_probe_at[path] > clock[0] for path in paths) +@pytest.mark.parametrize("background_slots", [0, 2]) +@pytest.mark.parametrize("origin", ["lark", "web"]) +def test_current_body_uses_next_slot_before_unattempted_recovery_backlog(tmp_path, monkeypatch, background_slots, origin): + from loopx.extensions.lark import goal_topic_runtime_service as worker + + release_old, release_rest = threading.Event(), threading.Event() + changed = threading.Condition() + started, finished = [], [] + slots = 4 - background_slots + transport = LarkPrivateConversations.__new__(LarkPrivateConversations) + transport.root = tmp_path + transport._context_recovery_probe_at = {} + transport.core = SimpleNamespace(context_recovery_observer=lambda: lambda ref: True) + transport.reconcile_app_feedback = lambda: 0 + web = tmp_path / "web-projections" / "current.json" + transport._web_projection_paths = lambda: [web] if web.exists() else [] + fresh = web if origin == "web" else tmp_path / "z-current.json" + for index in range(slots * 3): + (tmp_path / f"old-{index:02}.json").write_text(json.dumps({"status": "delivered", + "request_ref": str(index), "session_id": "original", "turn_id": "failed"})) + + def reconcile(path): + with changed: + started.append(path) + first_wave = len(started) <= slots + changed.notify_all() + assert (release_old if first_wave else release_rest).wait(5) + with changed: + finished.append(path) + changed.notify_all() + return 0 + + transport.reconcile_request = reconcile + if background_slots: + transport.owner_attention = SimpleNamespace(has_enabled=lambda: True, + reconcile=lambda: release_rest.wait(5) and 0) + transport.steward_reports = SimpleNamespace(pending=lambda: True, + reconcile=lambda: release_rest.wait(5) and 0) + service = worker.LarkGoalTopicRuntimeService(snapshot_provider=lambda: {}, runtime_root=tmp_path, + runtime_controller=None, private_conversations=transport) + + def drive(step, **_): + step() + with changed: + assert changed.wait_for(lambda: len(started) == slots, 2) + fresh.parent.mkdir(exist_ok=True) + fresh.write_text(json.dumps({"status": "pending" if origin == "web" else "accepted"})) + step() # Retain the new discovery while every existing slot is held. + release_old.set() + with changed: + assert changed.wait_for(lambda: len(finished) == slots, 2) + # The provider returns just after its last observable checkpoint. + # Let the actual Future settle instead of racing its done flag. + deadline = time.monotonic() + 2 + while time.monotonic() < deadline: + step() + with changed: + if changed.wait_for(lambda: len(started) == slots * 2, .01): + break + with changed: + assert len(started) == slots * 2 + assert fresh in started[slots:], "new body remained behind untouched historical recovery probes" + service._closed.set() + release_rest.set() + + monkeypatch.setattr(service._delivery_worker, "repeat", drive) + try: + service._reconcile_private() + finally: + service._closed.set() + release_old.set() + release_rest.set() + assert len(started) == slots * 2 and len(set(started)) == len(started) + assert sum(path.name.startswith("old-") for path in started) == slots * 2 - 1 + + def test_private_recovery_discovery_keeps_negative_and_attempt_backoff(tmp_path, monkeypatch): from loopx.extensions.lark import private_conversations as private