From 7bfd33574b5abd7d029d8b3fd0fefefacc2d10f4 Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Wed, 11 Nov 2026 10:39:18 +0800 Subject: [PATCH] fix(lark): give ready private replies an available worker Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> --- .../lark/goal_topic_runtime_service.py | 15 +++- tests/test_lark_private_conversations.py | 82 +++++++++++++++++++ 2 files changed, 94 insertions(+), 3 deletions(-) diff --git a/loopx/extensions/lark/goal_topic_runtime_service.py b/loopx/extensions/lark/goal_topic_runtime_service.py index 3c716b1fcb..a6a260fd30 100644 --- a/loopx/extensions/lark/goal_topic_runtime_service.py +++ b/loopx/extensions/lark/goal_topic_runtime_service.py @@ -197,11 +197,20 @@ 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(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()): + recovering = {path: recovery_path(path) for path in ready} + active_background = sum(job is not None for job in (feedback, reports, attention, ingress, discovery)) + due_background = sum((reports_ready, attention_ready, ingress_ready)) + scheduled_body = False + for path in sorted(ready, key=lambda path: (recovering[path], attempted.get(path, 0))): + occupied = len(pending) + active_background + if occupied >= 4 or self._closed.is_set(): + break + # Give an unsent body the next available slot when due + # checks would otherwise reserve every free slot. + if occupied + due_background >= 4 and (recovering[path] or scheduled_body): break pending[path] = pool.submit(transport.reconcile_request, path) + scheduled_body = scheduled_body or not recovering[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(): diff --git a/tests/test_lark_private_conversations.py b/tests/test_lark_private_conversations.py index d9630e8153..f789506ebe 100644 --- a/tests/test_lark_private_conversations.py +++ b/tests/test_lark_private_conversations.py @@ -6487,6 +6487,88 @@ def reconcile(path): runtime.close() +def test_original_reply_gets_free_worker_when_background_checks_are_due(ordinary, monkeypatch): # noqa: F811 + from types import SimpleNamespace + + from loopx.chat_store import _read_json + from loopx.extensions.lark import goal_topic_runtime_service as module + + store, runtime, provider, transport = connect(ordinary) + discovery_started, background_due, release, delivered, ingress_started = ( + threading.Event() for _ in range(5) + ) + # Advance only the scheduler's due checks. The four worker threads, Core + # admission and provider readback still use their normal entry points. + monkeypatch.setattr(module, "time", SimpleNamespace( + monotonic=lambda: 61.0 if background_due.is_set() else 0.0, + )) + pending_bodies = transport.pending_body_paths + monkeypatch.setattr(transport, "pending_body_paths", + lambda: pending_bodies() if background_due.is_set() else []) + monkeypatch.setattr(transport, "reconcile_app_feedback", None) + + def discover_history(): + discovery_started.set() + assert release.wait(12) + return [] + + def reconcile_background(): + assert release.wait(12) + return 0 + + def observe_ingress(*, cancelled): + ingress_started.set() + assert release.wait(12) + return [] + + monkeypatch.setattr(transport, "pending_delivery_paths", discover_history) + monkeypatch.setattr(transport.steward_reports, "pending", background_due.is_set) + monkeypatch.setattr(transport.steward_reports, "reconcile", reconcile_background) + monkeypatch.setattr(transport.owner_attention, "has_enabled", background_due.is_set) + monkeypatch.setattr(transport.owner_attention, "reconcile", reconcile_background) + monkeypatch.setattr(transport, "observe_ingress", observe_ingress) + event = provider.event("notes-app", "reply-during-due-checks", "Explain the original request") + ref = identity_ref(transport._binding("notes-app")["provider_ref"], event["message_id"]) + target = transport.root / f"{ref}.json" + reconcile_original = transport.reconcile_request + + def reconcile(path): + result = reconcile_original(path) + if path == target and result == 1: + delivered.set() + return result + + monkeypatch.setattr(transport, "reconcile_request", reconcile) + service = module.LarkGoalTopicRuntimeService(snapshot_provider=lambda: {}, runtime_root=transport.runtime_root, + runtime_controller=runtime, private_conversations=transport, + profile_poller=lambda _profile, stop: stop.wait(12)) + service.start() + try: + assert discovery_started.wait(5) + assert transport.admit("notes-app", event)["status"] == "durably_accepted" + request = transport.core.read_request(ref) + turn = runtime.wait_for_turn(session_id=request["session_id"], turn_id=request["turn_id"], timeout_sec=5) + assert turn["status"] == "completed" + background_due.set() + service._delivery_wakeup.set() + assert delivered.wait(4), "Due background checks occupied every worker ahead of an original reply" + assert ingress_started.wait(4) + terminal = _read_json(target)["deliveries"]["terminal"] + assert terminal["verified"] and terminal["text"] == turn["response"]["message"] + reply = provider.messages[terminal["attempt"]["message_ref"]] + assert reply["chat_id"] == event["chat_id"] + assert json.loads(reply["body"]["content"])["zh_cn"]["content"][0][0]["text"] == terminal["text"] + 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 + 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