diff --git a/loopx/canary/module_metric_baseline.json b/loopx/canary/module_metric_baseline.json index 081b6f04cd..cd724ea7d6 100644 --- a/loopx/canary/module_metric_baseline.json +++ b/loopx/canary/module_metric_baseline.json @@ -84,7 +84,7 @@ "lines": 2760 }, "loopx/extensions/lark/private_conversations.py": { - "lines": 2012 + "lines": 2024 }, "loopx/heartbeat_prompt.py": { "any_count": 17, diff --git a/loopx/extensions/lark/goal_topic_runtime_service.py b/loopx/extensions/lark/goal_topic_runtime_service.py index 0cea1658f9..203374f6ad 100644 --- a/loopx/extensions/lark/goal_topic_runtime_service.py +++ b/loopx/extensions/lark/goal_topic_runtime_service.py @@ -95,6 +95,8 @@ def _reconcile_private(self) -> None: reports_due = 0.0 feedback: Future[int] | None = None feedback_due = 0.0 + discovery: Future[list[Path]] | None = None + discovery_due = 0.0 ingress: Future[list[str]] | None = None ingress_due = time.monotonic() + 60 wakeup_scope = getattr(self.runtime_controller, "turn_release_wakeup", None) @@ -103,10 +105,18 @@ 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 + nonlocal attempted, 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() + 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() @@ -152,7 +162,8 @@ def step() -> None: except Exception: logging.getLogger(__name__).warning("Private Chat delivery reconciliation is pending") try: - paths = transport.pending_delivery_paths() + 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. @@ -184,25 +195,33 @@ def step() -> None: # 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()): + + 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) 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) < 4 and not self._closed.is_set(): + 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) - if attention_ready and len(pending) + int(feedback is not None) + int(reports is not None) + int(ingress is not None or ingress_ready) < 4 and not self._closed.is_set(): + if attention_ready and len(pending) + int(feedback is not None) + int(reports is not None) + int(ingress is not None or ingress_ready) + int(discovery is not None) < 4 and not self._closed.is_set(): attention = pool.submit(owner_attention.reconcile) - if ingress_ready and len(pending) + int(feedback is not None) + int(reports is not None) + int(attention is not None) < 4 and not self._closed.is_set(): + if ingress_ready and len(pending) + int(feedback is not None) + int(reports is not None) + int(attention is not None) + int(discovery is not None) < 4 and not self._closed.is_set(): ingress = pool.submit(getattr(transport, "observe_ingress", observer.observe), cancelled=self._closed.is_set) app_feedback = getattr(transport, "reconcile_app_feedback", None) # Optional feedback uses one slot in this same bounded # pool and leaves capacity for canonical body deliveries. if (callable(app_feedback) and feedback is None and time.monotonic() >= feedback_due and len(pending) + int(reports is not None) + int(attention is not None) - + int(ingress is not None) < 3 and not self._closed.is_set()): + + int(ingress is not None) + int(discovery is not None) < 3 and not self._closed.is_set()): feedback = pool.submit(app_feedback) + # One discovery pass uses this same bounded pool. Its slow + # native/history observations cannot hold the scanner that + # discovers fresh captures and already generated answers. + if (callable(body_paths) and discovery is None and time.monotonic() >= discovery_due + and len(pending) + int(feedback is not None) + int(reports is not None) + + int(attention is not None) + int(ingress is not None) < 4 and not self._closed.is_set()): + discovery = pool.submit(transport.pending_delivery_paths) + discovery.add_done_callback(lambda _: self._delivery_wakeup.set()) except Exception: logging.getLogger(__name__).warning("Private Chat delivery discovery is pending") diff --git a/loopx/extensions/lark/private_conversations.py b/loopx/extensions/lark/private_conversations.py index 975d1b98eb..8224ad7f89 100644 --- a/loopx/extensions/lark/private_conversations.py +++ b/loopx/extensions/lark/private_conversations.py @@ -59,6 +59,7 @@ def __init__(self, *, controller: Any, runtime_root: Path, runner: Any, cli_bin: self.root = controller.store.root / "lark-private-deliveries" self._context_recovery_probe_at: dict[Path, float] = {} self._delivery_discovery = PrivateDeliveryDiscovery(self.root) + self._body_discovery = PrivateDeliveryDiscovery(self.root) self._web_projection_scan_at = 0.0 self._web_projection_snapshots: dict[str, tuple[int, int]] = {} from .owner_attention import LarkOwnerAttention @@ -1271,6 +1272,17 @@ def inbox() -> Path: except (KeyError, ValueError, OSError): return # Unknown authority or provider effects stay pending. + def pending_body_paths(self) -> list[Path]: + """Find unsent bodies without waiting for historical recovery probes. + + The existing pool reads the complete discovery separately. Keep this + metadata cache local to the scanner; it grants no admission or send. + """ + paths = [path for path, row, _ in self._body_discovery.rows() + if row["status"] != "delivered"] + return [*paths, *(path for path in sorted((self.root / "web-projections").glob("*.json")) + if _read_json(path).get("status") == "pending")] + @effect_runtime_request_scope() def pending_delivery_paths(self) -> list[Path]: """The durable transport store is the queue; no in-memory admission.""" diff --git a/tests/test_lark_private_conversations.py b/tests/test_lark_private_conversations.py index 82abab90ba..6f6172db56 100644 --- a/tests/test_lark_private_conversations.py +++ b/tests/test_lark_private_conversations.py @@ -5783,6 +5783,7 @@ def test_private_recovery_backlog_uses_released_capacity_before_retry_delay(tmp_ transport = LarkPrivateConversations.__new__(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 @@ -5822,8 +5823,13 @@ def write(ref): monkeypatch.setattr(worker, "time", SimpleNamespace(monotonic=lambda: clock[0])) def drive(step, **_): - step() - assert full.wait(2) + # Historical discovery now hands its result back to the same scanner. + # Drive that real handoff before asserting the unchanged pool bound. + deadline = time.monotonic() + 2 + while not full.is_set() and time.monotonic() < deadline: + step() + full.wait(.01) + assert full.is_set() clock[0] = 101 step() # No free capacity: do not put the remaining old requests off. paths.append(write("late")) @@ -5863,6 +5869,7 @@ def test_current_body_uses_next_slot_before_unattempted_recovery_backlog(tmp_pat transport = LarkPrivateConversations.__new__(LarkPrivateConversations) transport.root = tmp_path transport._delivery_discovery = PrivateDeliveryDiscovery(tmp_path) + transport._body_discovery = PrivateDeliveryDiscovery(tmp_path) transport._context_recovery_probe_at = {} transport.core = SimpleNamespace(context_recovery_observer=lambda: lambda ref: True) transport.reconcile_app_feedback = lambda: 0 @@ -5894,9 +5901,14 @@ def reconcile(path): runtime_controller=None, private_conversations=transport) def drive(step, **_): - step() + deadline = time.monotonic() + 2 + while time.monotonic() < deadline: + step() + with changed: + if changed.wait_for(lambda: len(started) == slots, .01): + break with changed: - assert changed.wait_for(lambda: len(started) == slots, 2) + assert len(started) == slots 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. @@ -6088,3 +6100,122 @@ def test_completed_command_recovery_defers_app_probe_and_revocation_blocks_reply assert provider.writes == [] and store.list_sessions() == [] finally: runtime.close() + + +def test_original_reply_is_delivered_while_historical_recovery_discovery_waits(ordinary, monkeypatch): # noqa: F811 + from loopx.chat_store import _atomic_write_json, _read_json + from loopx.extensions.lark.goal_topic_runtime_service import LarkGoalTopicRuntimeService + + store, runtime, provider, transport = connect(ordinary) + scanning, release, delivered = threading.Event(), threading.Event(), threading.Event() + old_ref = "0" * 24 + old_path = transport.root / f"{old_ref}.json" + # This is only a synthetic historical discovery candidate, with no grant + # or Core request. Its blocked read must not delay a real original answer. + _atomic_write_json(old_path, {"status": "delivered", "request_ref": old_ref, + "session_id": "historical", "turn_id": "historical", "event": {}, "deliveries": {}}) + observer = transport.core.context_recovery_observer + + def recovery_observer(): + original = observer() + + def candidate(ref): + if ref == old_ref: + scanning.set() + assert release.wait(15) + return False + return original(ref) + return candidate + + monkeypatch.setattr(transport.core, "context_recovery_observer", recovery_observer) + original_reconcile = transport.reconcile_request + target = [None] + + def reconcile(path): + result = original_reconcile(path) + if path == target[0] and _read_json(path).get("status") == "delivered": + delivered.set() + return result + + monkeypatch.setattr(transport, "reconcile_request", reconcile) + service = LarkGoalTopicRuntimeService(snapshot_provider=lambda: {}, runtime_root=transport.runtime_root, + runtime_controller=runtime, private_conversations=transport, + profile_poller=lambda _profile, stop: stop.wait(15)) + service.start() + try: + assert scanning.wait(5) + event = provider.event("notes-app", "fresh-during-history-read", "Explain this original question") + target[0] = transport.root / f"{identity_ref(transport._binding('notes-app')['provider_ref'], event['message_id'])}.json" + assert service._admit_private("notes-app", event)["status"] == "durably_accepted" + request = next(row for row in transport.core.pending() if row.get("message") == event["content"]) + assert target[0].stem == request["request_ref"] + turn = runtime.wait_for_turn(session_id=request["session_id"], turn_id=request["turn_id"], timeout_sec=5) + assert turn["status"] == "completed" + assert delivered.wait(4), "A historical read blocked the original reply worker" + assert not release.is_set() + record = _read_json(target[0]) + terminal = record["deliveries"]["terminal"] + assert terminal["verified"] and terminal["text"] == turn["response"]["message"] + final = provider.messages[terminal["attempt"]["message_ref"]] + assert json.loads(final["body"]["content"])["zh_cn"]["content"][0][0]["text"] == turn["response"]["message"] + assert len(store.list_sessions()) == 1 + packets = [json.loads(line) for line in ordinary[4].read_text().splitlines()] + assert sum(packet["method"] == "turn/start" for packet in packets) == 1 + assert _read_json(old_path)["status"] == "delivered" + finally: + release.set() + service.close() + if service._delivery_thread is not None: + service._delivery_thread.join(5) + runtime.close() + + +def test_historical_discovery_uses_one_existing_slot_without_an_executor_backlog(tmp_path): + from loopx.extensions.lark.goal_topic_runtime_service import LarkGoalTopicRuntimeService + + scanning, release_scan, full, release_bodies = (threading.Event() for _ in range(4)) + calls, bodies = [], [] + lock = threading.Lock() + + class Pending: + def pending_body_paths(self): + return list(bodies) + + def pending_delivery_paths(self): + scanning.set() + assert release_scan.wait(15) + return [] + + def reconcile_request(self, path): + with lock: + calls.append(path) + if len(calls) == 3: + full.set() + assert release_bodies.wait(15) + return 0 + + service = LarkGoalTopicRuntimeService(snapshot_provider=lambda: {}, runtime_root=tmp_path, + runtime_controller=None, private_conversations=Pending()) + service.start() + closer = None + try: + assert scanning.wait(5) + for index in range(9): + path = tmp_path / str(index) + path.write_text("durable original request") + bodies.append(path) + service._delivery_wakeup.set() + assert full.wait(4) + assert len(calls) == 3 and len(set(calls)) == 3 # One scan plus three replies, total four. + closer = threading.Thread(target=service.close) + closer.start() + assert service._closed.wait(1) + finally: + release_scan.set() + release_bodies.set() + service.close() + if closer is not None: + closer.join(5) + service._delivery_thread.join(5) + assert not service._delivery_thread.is_alive() + assert len(calls) == 3 # The other six durable requests were never submitted to the executor.