From e902e44e387e8d966c44e3ec7c23bb07136acda5 Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Tue, 20 Oct 2026 03:33:38 +0800 Subject: [PATCH] fix(lark): retain recovery fairness across discovery backoff Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> --- .../app-conversation-and-async-inbox-v0.md | 5 +++ ...p-conversation-and-async-inbox-v0.zh-CN.md | 4 ++ .../lark/goal_topic_runtime_service.py | 5 ++- tests/test_lark_delivery_wakeup.py | 43 +++++++++++++++++++ 4 files changed, 56 insertions(+), 1 deletion(-) diff --git a/docs/architecture/rfcs/app-conversation-and-async-inbox-v0.md b/docs/architecture/rfcs/app-conversation-and-async-inbox-v0.md index 1483869d36..a0ee180641 100644 --- a/docs/architecture/rfcs/app-conversation-and-async-inbox-v0.md +++ b/docs/architecture/rfcs/app-conversation-and-async-inbox-v0.md @@ -1336,6 +1336,11 @@ existing thread, without a history migration or error-message text matching. This proves absence of execution, not provider health or eventual success. A failed continuation is returned and retained; it cannot generate another retry. +Recovery discovery backoff is only a temporary hint. The existing reply worker +retains its per-request fairness count while that canonical request still exists; +an omitted hint does not retire work or let earlier unresolved requests repeatedly +take every available slot. The four-worker and no-extra-backlog limits stay intact. + Continuation admission, progress, cursor and final delivery use a version-scoped presentation journal. They never reopen the acknowledged original Inbox source or edit its failure notice as a new draft. The final replaces only the new draft. diff --git a/docs/architecture/rfcs/app-conversation-and-async-inbox-v0.zh-CN.md b/docs/architecture/rfcs/app-conversation-and-async-inbox-v0.zh-CN.md index 448273bbdc..bd59f27c1c 100644 --- a/docs/architecture/rfcs/app-conversation-and-async-inbox-v0.zh-CN.md +++ b/docs/architecture/rfcs/app-conversation-and-async-inbox-v0.zh-CN.md @@ -849,6 +849,10 @@ Turn，确认完整条目清单仅包含原用户输入后，Core 才能按每 线程，不迁移历史，也不匹配错误消息文本。这证明尚未执行，不证明 provider 健康 或最终成功；续办若再次失败，就回传并保留失败，不能再生成下一次重试。 +恢复发现的退避只是临时提示。规范请求仍存在时，原回复 worker 保留该请求的公平 +计数；某轮没有提示不代表工作已退役，也不能让前面的未结请求反复占满全部空位。 +四 worker 和不增加执行积压的上限保持。 + 续办的接收提示、进展、游标与最终回传按版本独立记账，不重新打开原 Inbox 已确认的 消息，也不把原失败通知改成新草稿；最终结果只替换本次新草稿。对于已经回传、但旧版 误用原失败通知的续办，原 worker 可凭确切已确认的编辑凭证恢复旧通知，并结束随后 diff --git a/loopx/extensions/lark/goal_topic_runtime_service.py b/loopx/extensions/lark/goal_topic_runtime_service.py index b446ec9077..0e6de6a0f3 100644 --- a/loopx/extensions/lark/goal_topic_runtime_service.py +++ b/loopx/extensions/lark/goal_topic_runtime_service.py @@ -130,7 +130,10 @@ def _reconcile_private(self) -> None: logging.getLogger(__name__).warning("Private Chat delivery reconciliation is pending") try: paths = transport.pending_delivery_paths() - attempted = {path: count for path, count in attempted.items() if path in 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()} 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()) diff --git a/tests/test_lark_delivery_wakeup.py b/tests/test_lark_delivery_wakeup.py index 4f50e029e5..41e551ccaa 100644 --- a/tests/test_lark_delivery_wakeup.py +++ b/tests/test_lark_delivery_wakeup.py @@ -225,3 +225,46 @@ def unexpected(*_args): runtime_controller=SimpleNamespace(turn_release_wakeup=unexpected)) service._reconcile_private() service.close() + + +def test_temporary_discovery_backoff_does_not_starve_later_durable_request(tmp_path, monkeypatch): + from concurrent.futures import Future + import loopx.extensions.lark.goal_topic_runtime_service as service_module + + paths = [tmp_path / str(index) for index in range(5)] + for path in paths: + path.write_text("durable original request") + seen, batches = [], [] + + class CompletedPool: + def __init__(self, **_kwargs): pass + def __enter__(self): return self + def __exit__(self, *_args): pass + def submit(self, operation, path): + future = Future() + future.set_result(operation(path)) + return future + + class BackoffTransport: + polls = 0 + def pending_delivery_paths(self): + self.polls += 1 + if self.polls == 7: + service._closed.set() + # The canonical requests remain present during the hint's backoff. + return paths if self.polls % 2 and self.polls < 7 else [] + def reconcile_request(self, path): + seen.append(path) + batches.append(self.polls) + return 0 # An unresolved request remains canonical, without effects. + + monkeypatch.setattr(service_module, "ThreadPoolExecutor", CompletedPool) + service = LarkGoalTopicRuntimeService(snapshot_provider=lambda: {}, runtime_root=tmp_path, + runtime_controller=None, private_conversations=BackoffTransport()) + monkeypatch.setattr(service._delivery_wakeup, "wait", lambda _timeout: None) + service._reconcile_private() + assert set(seen) == set(paths) + assert paths[-1] in seen[4:8] # The previously skipped request gets the next chance. + assert all(batches.count(batch) <= 4 for batch in set(batches)) + assert all(path.read_text() == "durable original request" for path in paths) + assert not service._workers # Original reply scheduling creates no consumer.