fix(lark): keep unattempted recovery work ready when capacity returns

loopx/extensions/lark/goal_topic_runtime_service.py

@@ -157,6 +157,14 @@ def step() -> None:
                     # 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()}
+                    for path in paths:
+                        attempted.setdefault(path, 0)
+                    # A scan is not an attempt. Keep unsubmitted discoveries in
+                    # this same fairness map when the pool is full; discovery's
+                    # read-only backoff must not hide them after a slot opens.
+                    # Reconciliation re-reads the source and Core authority.
+                    paths = list(dict.fromkeys([
+                        *(path for path, count in attempted.items() if count == 0), *paths]))
                     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())

loopx/extensions/lark/private_conversations.py

@@ -1639,6 +1639,11 @@ def reconcile_request(self, path: Path) -> int:
         # and the shared Session queue/fence; transport workers never run models.
         row = _read_json(path)
         record_event_timing(row["profile"], row["event"]["message_id"], "reconciliation_started")
+        if row["status"] == "delivered" and row.get("session_id") and row.get("turn_id"):
+            # Discovery may find more candidates than the shared pool can run.
+            # Refresh the probe delay on the actual attempt, which may follow
+            # discovery after a full pool. Core still revalidates every effect.
+            self._context_recovery_probe_at[path] = time.monotonic() + 60
         if row["status"] == "delivered" and row.get("context_recovery"):
             # Repair historical presentation first. A delivered failure of a
             # continuation still needs Core's fresh recovery judgment; only a

tests/test_lark_private_conversations.py

@@ -19,7 +19,8 @@
 from loopx.extensions.lark.goal_topic_runtime import poll_lark_goal_topic_profile_once
 
 
-@pytest.mark.parametrize("fault", [None, "stop", "source-changed", "later-source", "no-capacity", "native-tool"])
+@pytest.mark.parametrize("fault", [None, "stop", "source-changed", "later-source", "no-capacity", "native-tool",
+    "stop-after-discovery", "source-changed-after-discovery"])
 def test_original_worker_continues_proven_unstarted_argon_quota_on_codex(tmp_path, language_server, monkeypatch, fault):  # noqa: F811
     from copy import deepcopy
     from datetime import timedelta
@@ -81,10 +82,12 @@ def checked_update(session_id, **changes):
                 transitions.append(session_id)
             return update_session(session_id, **changes)
         monkeypatch.setattr(store, "update_session", checked_update)
-        if fault == "stop":
+        if fault in {"stop-after-discovery", "source-changed-after-discovery"}:
+            assert path in transport.pending_delivery_paths()
+        if fault in {"stop", "stop-after-discovery"}:
             store.append_event(original["session_id"], original["turn_id"], kind="turn.interrupt_requested", payload={})
             original_events = store.events_after(original["session_id"], original["turn_id"], None)
-        elif fault == "source-changed":
+        elif fault in {"source-changed", "source-changed-after-discovery"}:
             provider.messages[event["message_id"]]["content"] = "Changed original input"
         elif fault == "later-source":
             transport.admit("steward-app", provider.event("steward-app", "source-correction", "/status"))
@@ -4501,6 +4504,137 @@ def held_reaction(self, args, cwd=None, timeout=None):
     assert all(phase.get("verified") for phase in projection["deliveries"].values())
 
 
+@pytest.mark.parametrize("background_slots", [0, 2])
+def test_private_recovery_backlog_uses_released_capacity_before_retry_delay(tmp_path, monkeypatch, background_slots):
+    from loopx.extensions.lark import goal_topic_runtime_service as worker
+    from loopx.extensions.lark import private_conversations as private
+
+    # Exercise the real discovery, bounded executor, fairness counters and
+    # reconciliation entrypoint. Only native/provider work and the clock are
+    # synthetic: discovering work while full must not count as attempting it.
+    clock = [100.0]
+    release, full = threading.Event(), threading.Event()
+    lock = threading.Lock()
+    attempts, completed = {}, set()
+    slots = 4 - background_slots
+    native_root = tmp_path / "native"
+    native_root.mkdir()
+    transport = LarkPrivateConversations.__new__(LarkPrivateConversations)
+    transport.root = tmp_path
+    transport._context_recovery_probe_at = {}
+    transport._web_projection_paths = lambda: []
+    transport.reconcile_app_feedback = lambda: 0
+
+    def recover(ref, **_):
+        with lock:
+            attempts.setdefault(ref, clock[0])
+            if len(attempts) == slots:
+                full.set()
+        assert release.wait(5)
+        completed.add(ref)
+        return {"request_ref": ref, "turn_id": "original-failed-turn"}
+
+    transport.core = SimpleNamespace(root=native_root,
+        context_recovery_candidate=lambda ref: ref != "ineligible",
+        context_recovery_observer=lambda: transport.core.context_recovery_candidate,
+        recover_failed_request=recover)
+    if background_slots:
+        transport.owner_attention = SimpleNamespace(has_enabled=lambda: True,
+            reconcile=lambda: release.wait(5) and 0)
+        transport.steward_reports = SimpleNamespace(pending=lambda: True,
+            reconcile=lambda: release.wait(5) and 0)
+
+    def write(ref):
+        path = tmp_path / f"{ref}.json"
+        path.write_text(json.dumps({"status": "delivered", "request_ref": ref,
+            "session_id": "original-session", "turn_id": "original-failed-turn",
+            "profile": "fixture", "event": {"message_id": ref}}))
+        return path
+
+    paths = [write(f"old-{i:02}") for i in range(9)]
+    write("ineligible")
+    original_bytes = {path: path.read_bytes() for path in tmp_path.glob("*.json")}
+    service = worker.LarkGoalTopicRuntimeService(snapshot_provider=lambda: {}, runtime_root=tmp_path,
+        runtime_controller=None, private_conversations=transport)
+    monkeypatch.setattr(private, "time", SimpleNamespace(monotonic=lambda: clock[0]))
+    monkeypatch.setattr(worker, "time", SimpleNamespace(monotonic=lambda: clock[0]))
+
+    def drive(step, **_):
+        step()
+        assert full.wait(2)
+        clock[0] = 101
+        step()  # No free capacity: do not put the remaining old requests off.
+        paths.append(write("late"))
+        clock[0] = 102
+        step()  # The later arrival is also discovered while all slots are held.
+        release.set()
+        for tick in range(103, 115):
+            clock[0] = tick
+            step()
+            deadline = time.monotonic() + .1
+            while len(completed) < min(10, (tick - 102) * slots) and time.monotonic() < deadline:
+                time.sleep(.001)
+            if len(completed) == 10:
+                break
+        service._closed.set()
+
+    monkeypatch.setattr(service._delivery_worker, "repeat", drive)
+    try:
+        service._reconcile_private()
+    finally:
+        release.set()
+    assert set(attempts) == {path.stem for path in paths}
+    assert attempts["late"] < 115
+    assert all(path.read_bytes() == body for path, body in original_bytes.items())
+    assert all(transport._context_recovery_probe_at[path] > clock[0] for path in paths)
+
+
+def test_private_recovery_discovery_keeps_negative_and_attempt_backoff(tmp_path, monkeypatch):
+    from loopx.extensions.lark import private_conversations as private
+
+    clock, checks, attempts = [100.0], [], []
+    eligible = {"candidate": True, "ineligible": False}
+    transport = LarkPrivateConversations.__new__(LarkPrivateConversations)
+    transport.root = tmp_path
+    transport._context_recovery_probe_at = {}
+    transport._web_projection_paths = lambda: []
+    monkeypatch.setattr(private, "time", SimpleNamespace(monotonic=lambda: clock[0]))
+
+    def candidate(ref):
+        checks.append(ref)
+        return eligible[ref]
+
+    def unavailable(ref, **_):
+        attempts.append(ref)
+        raise OSError("synthetic native observation unavailable")
+
+    transport.core = SimpleNamespace(context_recovery_candidate=candidate,
+        context_recovery_observer=lambda: transport.core.context_recovery_candidate,
+        recover_failed_request=unavailable)
+    paths = {}
+    for ref in eligible:
+        path = tmp_path / f"{ref}.json"
+        path.write_text(json.dumps({"status": "delivered", "request_ref": ref,
+            "session_id": "original", "turn_id": "failed", "profile": "fixture",
+            "event": {"message_id": ref}}))
+        paths[ref] = path
+    assert transport.pending_delivery_paths() == [paths["candidate"]]
+    assert transport.pending_delivery_paths() == []
+    assert checks.count("ineligible") == 1 and attempts == []
+    clock[0] += 30  # The worker finally gets a slot after discovery.
+    with pytest.raises(OSError, match="observation unavailable"):
+        transport.reconcile_request(paths["candidate"])
+    assert attempts == ["candidate"]
+    clock[0] += 31  # Discovery's old hint elapsed; the actual probe's has not.
+    assert transport.pending_delivery_paths() == []
+    clock[0] += 30
+    # Fresh eligibility wins over the old discovery hint; no stale cached
+    # candidate is an execution grant after a source/permission/stop change.
+    eligible["candidate"] = False
+    assert transport.pending_delivery_paths() == []
+    assert attempts == ["candidate"]
+
+
 def test_private_reply_workers_have_no_executor_backlog_and_stop_scheduling(tmp_path):
     import threading
     from loopx.extensions.lark.goal_topic_runtime_service import LarkGoalTopicRuntimeService