fix(chat): retire stopped quota failures from pending readback

docs/reference/protocols/manager-evidence-and-continuity-v0.md

@@ -450,6 +450,18 @@ readback remain in force, preserving the failed Turn and its original notice.
 仍保留待核对，不能走此路径重放。退避、三次上限、稳定接续身份、失败历史和逐次
 原受众送达读回均保持原约束。
 
+The existing pending count retains delivered quota failures while work remains
+outstanding. A later same-source `/stop` or `/new`, or closure of the original
+Session, retires that backlog from the read-only count without deleting the
+failed Turn or notice. Temporary binding/provider unavailability, backoff and
+retry exhaustion are not retirement evidence. Health reads neither admit
+recovery nor contact a provider.
+
+既有待处理数量保留已送达失败通知、仍未结束的额度失败工作。后续同来源 `/stop`、
+`/new` 或原 Session 关闭后，只读计数不再包含该积压；失败 Turn 和通知仍保留。
+临时绑定／provider 不可用、退避和重试耗尽不代表退休。健康读取不准入恢复，
+也不访问 provider。
+
 Default manager duties allow authorized reads and summaries; reminders use an
 existing authorized destination/subscription. Goal changes, steering,
 cross-Goal dispatch, pause/resume, publishing, and trading keep their existing

loopx/capabilities/native_chat/external_conversations.py

@@ -75,14 +75,6 @@ def _context_recovery_plan(self, row: dict[str, Any], *,
                 or reservation.get("read_only_review") or reservation.get("work_attempt"))
             and reservation.get("failed_turn_id") == row["turn_id"] else None)
         events = store.events_after(session["session_id"], turn["turn_id"], None)
-        # A later source control also retires an unexecuted failed request.
-        # The source fence makes this inventory consistent with admission.
-        stop = any(event.get("kind") in {"turn.interrupt_requested", "turn.interrupted"} for event in events)
-        if turn.get("status") in {"failed", "completed"}:
-            stop = stop or any(item.get("binding_id") == row["binding_id"]
-                and item.get("source") == row["source"] and item.get("command") in {"stop", "new"}
-                and datetime.fromisoformat(item["created_at"]) >= datetime.fromisoformat(row["created_at"])
-                for item in self.pending())
         return effect_runtime_result("collaboration.conversation.request", {
             "request": {key: value for key, value in row.items() if key != "attachments"},
             "current_session": session, "binding": selected_binding,
@@ -91,10 +83,35 @@ def _context_recovery_plan(self, row: dict[str, Any], *,
                 "admitted_turn": admitted,
                 "attachments_present": bool(row.get("attachments")),
                 "steering_observed": any(event.get("kind") == "turn.steered" for event in events),
-                "stop_observed": stop,
+                "stop_observed": self._source_stop_observed(row, turn, events),
                 "native_observation": observation, "source_verified": source_verified},
         })
 
+    def _source_stop_observed(self, row: dict[str, Any], turn: dict[str, Any],
+                              events: list[dict[str, Any]]) -> bool:
+        """Read the same durable source controls for recovery and display."""
+        return (any(event.get("kind") in {"turn.interrupt_requested", "turn.interrupted"} for event in events)
+            or (turn.get("status") in {"failed", "completed"} and any(
+                item.get("binding_id") == row["binding_id"] and item.get("source") == row["source"]
+                and item.get("command") in {"stop", "new"}
+                and datetime.fromisoformat(item["created_at"]) >= datetime.fromisoformat(row["created_at"])
+                for item in self.pending())))
+
+    def quota_backlog_pending(self, row: dict[str, Any]) -> bool:
+        """Project existing work without resolving providers or admitting recovery."""
+        from ...control_plane.effect_runtime import effect_runtime_result
+
+        store = self.controller.store
+        session = store.load_session(row["session_id"])
+        turn = store.load_turn(row["session_id"], row["turn_id"])
+        events = store.events_after(row["session_id"], row["turn_id"], None) if turn else []
+        return effect_runtime_result("collaboration.conversation.request", {
+            "request": {key: value for key, value in row.items() if key != "attachments"},
+            "current_session": session,
+            "pending_observation": {"turn": turn,
+                "stop_observed": self._source_stop_observed(row, turn, events) if turn else False},
+        }).get("pending") is True
+
     def context_recovery_candidate(self, request_ref: str) -> bool:
         """Discovery only; every authority and native fact is re-read at admission."""
         try:

loopx/control_plane/collaboration/conversation_binding.ts

@@ -437,6 +437,18 @@ export function planBoundConversationRequest(params: JsonObject): JsonObject {
     throw new EffectRuntimeRequestError("unsupported external conversation command");
   }
   const current = params.current_session === null ? null : requireJsonObject(params.current_session, "current Session");
+  if (params.pending_observation !== undefined) {
+    const observation = requireJsonObject(params.pending_observation, "pending work observation");
+    const turn = observation.turn == null ? null : requireJsonObject(observation.turn, "observed Turn");
+    // Pending work is not current retry eligibility: backoff, disabled bindings,
+    // exhausted attempts and unavailable providers do not retire the request.
+    // Only explicit lifecycle evidence removes a delivered quota failure.
+    return {pending: row.command === null && row.status === "accepted"
+      && turn?.session_id === row.session_id && turn?.turn_id === row.turn_id
+      && turn?.status === "failed" && turn.error_code === "usage_limit_exceeded"
+      && !(current?.session_id === row.session_id && current?.status === "closed")
+      && observation.stop_observed !== true};
+  }
   if (params.context_recovery !== undefined) {
     return planContextRecovery(params, row, current);
   }

loopx/extensions/lark/private_conversations.py

@@ -235,9 +235,7 @@ def health(self) -> dict[str, dict[str, int]]:
                 if (native.get("status") == "accepted" and native.get("command") is None
                         and all(native.get(key) == row.get(key) for key in
                             ("binding_id", "source", "session_id", "turn_id"))):
-                    turn = self.core.controller.store.load_turn(row["session_id"], row["turn_id"])
-                    pending = bool(turn and turn.get("status") == "failed"
-                                   and turn.get("error_code") == "usage_limit_exceeded")
+                    pending = self.core.quota_backlog_pending(native)
             if pending:
                 entry["pending_count"] += 1
                 entry["recovery_count"] += int(any(phase.get("started") and not phase.get("verified")

tests/control_plane_ts/conversation_binding.test.ts

@@ -702,3 +702,31 @@ test("local input cannot impersonate a group participant or inherit its workspac
   assert.equal(resolveConversationScope({goal_id: null, channel_id: selected.channel_id,
     project_context: selected.context, origin: "web"}).kind, "unavailable");
 });
+
+test("quota backlog projection retires explicit lifecycle stops, not unavailable recovery", () => {
+  const original = {request_ref: "f".repeat(24), command: null, status: "accepted",
+    session_id: "original", turn_id: "failed"};
+  const turn = {session_id: "original", turn_id: "failed", status: "failed", error_code: "usage_limit_exceeded"};
+  const packet = {request: original, current_session: {session_id: "original", status: "ready"},
+    pending_observation: {turn, stop_observed: false}};
+  assert.equal(planBoundConversationRequest(packet).pending, true);
+  // No binding/provider/native allowance input is necessary for this read.
+  assert.equal(planBoundConversationRequest({...packet, request: {...original,
+    context_recovery: {quota_attempt: 3, not_before: "2099-01-01T00:00:00Z"}}}).pending, true);
+  assert.equal(planBoundConversationRequest({...packet, current_session: null}).pending, true,
+    "missing session evidence must not silently retire known quota work");
+  assert.equal(planBoundConversationRequest({...packet,
+    current_session: {session_id: "foreign", status: "closed"}}).pending, true);
+  assert.equal(planBoundConversationRequest({...packet,
+    current_session: {...packet.current_session, status: "closed"}}).pending, false);
+  assert.equal(planBoundConversationRequest({...packet,
+    pending_observation: {...packet.pending_observation, stop_observed: true}}).pending, false);
+  for (const changed of [null, {...turn, error_code: "policy"}, {...turn, status: "completed"},
+    {...turn, session_id: "foreign"}, {...turn, turn_id: "foreign"}]) {
+    assert.equal(planBoundConversationRequest({...packet,
+      pending_observation: {turn: changed, stop_observed: false}}).pending, false);
+  }
+  for (const changed of [{...original, command: "stop"}, {...original, status: "rejected"}]) {
+    assert.equal(planBoundConversationRequest({...packet, request: changed}).pending, false);
+  }
+});

tests/test_chat_conversation_bindings.py

@@ -12,7 +12,8 @@
 from loopx.extensions.lark.conversation_identity import identity_ref
 
 
-def test_quota_backlog_observes_a_later_source_stop_before_new_admission(ordinary, monkeypatch):  # noqa: F811
+@pytest.mark.parametrize("command", ["stop", "new"])
+def test_quota_backlog_observes_a_later_source_stop_before_new_admission(ordinary, monkeypatch, command):  # noqa: F811
     from datetime import datetime, timedelta, timezone
     from loopx.capabilities.native_chat import external_conversations
     from test_lark_private_conversations import connect
@@ -38,7 +39,10 @@ def now(cls, tz=None):
         assert transport.core._context_recovery_plan(original, discovery_only=True)["recovery_candidate"]
         transport.admit("steward-app", provider.event("steward-app", "unrelated-stop", "/stop"))
         assert transport.core._context_recovery_plan(original, discovery_only=True)["recovery_candidate"]
-        transport.admit("notes-app", provider.event("notes-app", "later-stop", "/stop"))
+        assert transport.health()[row["binding_id"]]["pending_count"] == 1
+        transport.admit("notes-app", provider.event("notes-app", "later-stop", f"/{command}"))
+        transport.reconcile()
+        assert transport.health()[row["binding_id"]]["pending_count"] == 0
         calls = ordinary[4].read_bytes()
         assert not transport.core._context_recovery_plan(original, discovery_only=True)["recovery_candidate"]
         assert ordinary[4].read_bytes() == calls
@@ -47,6 +51,13 @@ def now(cls, tz=None):
         newer = next(x for x in transport.core.pending() if x.get("message") == "A new independent request")
         completed = runtime.wait_for_turn(session_id=newer["session_id"], turn_id=newer["turn_id"], timeout_sec=10)
         assert completed["status"] == "completed"
+        transport.reconcile()
+        from loopx.extensions.lark.private_conversations import LarkPrivateConversations
+        restarted = LarkPrivateConversations(controller=runtime, runtime_root=transport.runtime_root,
+            runner=provider, cli_bin="lark-cli")
+        before = ordinary[4].read_bytes(), list(provider.calls)
+        assert restarted.health()[row["binding_id"]]["pending_count"] == 0
+        assert (ordinary[4].read_bytes(), provider.calls) == before
         assert store.load_turn(row["session_id"], row["turn_id"]) == failed
     finally:
         runtime.close()