fix(lark): retain recovery fairness across discovery backoff

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.

docs/architecture/rfcs/app-conversation-and-async-inbox-v0.zh-CN.md

@@ -849,6 +849,10 @@ Turn，确认完整条目清单仅包含原用户输入后，Core 才能按每
 线程，不迁移历史，也不匹配错误消息文本。这证明尚未执行，不证明 provider 健康
 或最终成功；续办若再次失败，就回传并保留失败，不能再生成下一次重试。
 
+恢复发现的退避只是临时提示。规范请求仍存在时，原回复 worker 保留该请求的公平
+计数；某轮没有提示不代表工作已退役，也不能让前面的未结请求反复占满全部空位。
+四 worker 和不增加执行积压的上限保持。
+
 续办的接收提示、进展、游标与最终回传按版本独立记账，不重新打开原 Inbox 已确认的
 消息，也不把原失败通知改成新草稿；最终结果只替换本次新草稿。对于已经回传、但旧版
 误用原失败通知的续办，原 worker 可凭确切已确认的编辑凭证恢复旧通知，并结束随后

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())

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.