diff --git a/docs/reference/protocols/manager-evidence-and-continuity-v0.md b/docs/reference/protocols/manager-evidence-and-continuity-v0.md index cade73a3b0..e96a7ca543 100644 --- a/docs/reference/protocols/manager-evidence-and-continuity-v0.md +++ b/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 diff --git a/loopx/capabilities/native_chat/external_conversations.py b/loopx/capabilities/native_chat/external_conversations.py index d8cb47d187..e35f070e45 100644 --- a/loopx/capabilities/native_chat/external_conversations.py +++ b/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: diff --git a/loopx/control_plane/collaboration/conversation_binding.ts b/loopx/control_plane/collaboration/conversation_binding.ts index 600c498f7d..845cd4fd8c 100644 --- a/loopx/control_plane/collaboration/conversation_binding.ts +++ b/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); } diff --git a/loopx/extensions/lark/private_conversations.py b/loopx/extensions/lark/private_conversations.py index a0bdcdf704..a67d880860 100644 --- a/loopx/extensions/lark/private_conversations.py +++ b/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") diff --git a/tests/control_plane_ts/conversation_binding.test.ts b/tests/control_plane_ts/conversation_binding.test.ts index 6e4f8b1a4d..9059b11ae3 100644 --- a/tests/control_plane_ts/conversation_binding.test.ts +++ b/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); + } +}); diff --git a/tests/test_chat_conversation_bindings.py b/tests/test_chat_conversation_bindings.py index 697294fe04..c43667324f 100644 --- a/tests/test_chat_conversation_bindings.py +++ b/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()