From 924c2bbc9ccff969afbe92d14139569dfd27aac8 Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Wed, 4 Nov 2026 00:47:18 +0800 Subject: [PATCH] fix(chat): safely resume unstarted native quota failures Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> --- .../app-conversation-and-async-inbox-v0.md | 13 +++ ...p-conversation-and-async-inbox-v0.zh-CN.md | 9 ++ loopx/canary/module_metric_baseline.json | 4 +- .../native_chat/antigravity_adapter.py | 64 +++++++++++ .../native_chat/external_conversations.py | 24 +++- loopx/chat_runtime.py | 11 +- loopx/chat_store.py | 6 +- .../collaboration/conversation_binding.ts | 68 ++++++++++- .../conversation_binding.test.ts | 65 +++++++++++ tests/test_antigravity_steward_adapter.py | 60 +++++++++- tests/test_chat_manager_model_adoption.py | 31 +++++ tests/test_lark_private_conversations.py | 106 ++++++++++++++++++ 12 files changed, 444 insertions(+), 17 deletions(-) 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 6d667b7541..76d515a39a 100644 --- a/docs/architecture/rfcs/app-conversation-and-async-inbox-v0.md +++ b/docs/architecture/rfcs/app-conversation-and-async-inbox-v0.md @@ -1847,6 +1847,19 @@ card cannot substitute for final-message readback. ### Recovering a native result after Core failure +A quota-failure notice remains pending work. For Antigravity, the original worker +may reserve one new quota-continuation Turn only after two independent reads of +the same idle native trajectory prove its latest invocation contains exactly the +original input and a structured `QUOTA_EXHAUSTED` error, without model output or +tool steps. Admission binds the prompt hash, source, execution time, native index, +current authorization and fresh capacity. Later source input, steering, stop, +queued input, unknown or cleared steps veto continuation. An unconfirmed cancel +alone confers no replay eligibility. An enabled flexible endpoint pool may select +Codex through the existing quota owner; dispatch rechecks current configuration +and capacity. Keep the failed Turn and its original notice, and adopt the same +reservation after a crash. The original Bot still delivers and independently +reads back its normal result in the original audience. + Historical native-result discovery waits while the same Core Session is busy or has an active Turn, leaving shared reply workers available for current progress. Already verified result metadata keeps its existing correlation owner. 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 3618874cd9..4ed9b5d4ed 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 @@ -1210,6 +1210,15 @@ reaction 清理未完成时，也只恢复原结果 proof，不能再次发送 ### Core 失败后的原生结果恢复 +额度耗尽的失败提示仍是待办。对于 Antigravity，只在原 worker 两次独立读回同一 +空闲轨迹，证明最新一次调用恰好是原始输入和结构化 `QUOTA_EXHAUSTED` 错误、 +没有任何模型输出或工具步骤时，现有队列才可预约一次新的 quota 续办 Turn。 +准入绑定原 prompt hash、来源、执行时间、原生索引、当前授权与最新额度；后续 +来源输入、steering、stop、新队列、未知或已清理步骤均阻止续办。取消未确认本身 +不构成重放资格。已启用的灵活端点池可沿既有额度 owner 选择 Codex，派发前再次 +核验当前配置与容量。失败 Turn 和原失败通知保留，崩溃恢复采用同一个预约身份； +正常结果仍须由原 Bot 沿原受众实际回传并独立读回。 + 同一 Core Session 忙碌或仍有活跃 Turn 时，历史原生结果暂不进入读回发现，避免 占用当前进度所需的共享回复 worker。已核验结果的元数据仍由原关联修复路径负责。 diff --git a/loopx/canary/module_metric_baseline.json b/loopx/canary/module_metric_baseline.json index 7f94359675..51ded7160f 100644 --- a/loopx/canary/module_metric_baseline.json +++ b/loopx/canary/module_metric_baseline.json @@ -36,7 +36,7 @@ "loopx/chat_runtime.py": { "any_count": 37, "dict_any_count": 0, - "lines": 2395 + "lines": 2398 }, "loopx/chat_server.py": { "any_count": 25, @@ -44,7 +44,7 @@ "lines": 2039 }, "loopx/chat_store.py": { - "lines": 2111 + "lines": 2113 }, "loopx/codex_cli_probe.py": { "any_count": 37, diff --git a/loopx/capabilities/native_chat/antigravity_adapter.py b/loopx/capabilities/native_chat/antigravity_adapter.py index 4d8c7151d0..db82ceef39 100644 --- a/loopx/capabilities/native_chat/antigravity_adapter.py +++ b/loopx/capabilities/native_chat/antigravity_adapter.py @@ -380,6 +380,70 @@ class AntigravityAdapter: adapter_kind = "antigravity_ls" + @staticmethod + def quota_failure_observation(*, cascade_id: str, request_ref: str, message: str, + started_at: str, completed_at: str, + input_anchor: dict[str, Any]) -> dict[str, Any] | None: + """Prove one native input ended at quota before any model/tool output. + + An unconfirmed cancel is not proof of effects or of their absence. Read + the exact original input twice from a stable idle cascade instead. Only + the latest two-step input/quota-error sequence qualifies; any output, + tool, later input, cleared step or unknown native state fails closed. + """ + try: + found = discover_antigravity_language_server() + start = input_anchor.get("start_step_index") + if found is None or input_anchor.get("thread_id") != cascade_id or type(start) is not int or start < 0: + return None + def read(method: str, **payload: Any) -> dict[str, Any]: + return _rpc_post(found["url"], found["csrf_token"], method, + {"cascadeId": cascade_id, **payload}, timeout=5.0) + before = read("GetCascadeTrajectory") + total = before.get("numTotalSteps") + if (before.get("status") != "CASCADE_RUN_STATUS_IDLE" or type(total) is not int or total != start + 2 + or (before.get("trajectory") or {}).get("cascadeId") != cascade_id): + return None + raw = read("GetCascadeTrajectorySteps", startIndex=start, endIndex=total)["steps"] + indexed, covered = _observed_turn_steps(raw, start, total) + again = read("GetCascadeTrajectorySteps", startIndex=start, endIndex=total)["steps"] + after = read("GetCascadeTrajectory") + if covered != total or raw != again or before != after or len(indexed) != 2: + return None + user, terminal = [step for _, step in indexed] + if (user.get("type") != "CORTEX_STEP_TYPE_USER_INPUT" + or terminal.get("type") != "CORTEX_STEP_TYPE_ERROR_MESSAGE" + or any(step.get("status") != "CORTEX_STEP_STATUS_DONE" for step in (user, terminal)) + or set(user) - {"type", "status", "metadata", "userInput"} + or set(terminal) - {"type", "status", "metadata", "errorMessage"}): + return None + metadata = user["metadata"] + execution = metadata.get("executionId") + created = datetime.fromisoformat(metadata["createdAt"]) + ended = datetime.fromisoformat(terminal["metadata"]["createdAt"]) + prompt = _native_user_input_text(user) + if (not isinstance(execution, str) or not execution + or terminal["metadata"].get("executionId") not in (None, execution) + or not datetime.fromisoformat(started_at) <= created <= ended <= datetime.fromisoformat(completed_at) + or prompt is None or not manager_input_matches(prompt, message=message, request_ref=request_ref)): + return None + digest = hashlib.sha256(prompt.encode()).hexdigest() + if input_anchor.get("input_sha256") != digest: + return None + error = terminal["errorMessage"]["error"] + details = [json.loads(value) for value in error.get("rpcErrorDetails", [])] + quota = [value for value in details if value.get("@type") == "type.googleapis.com/google.rpc.ErrorInfo" + and value.get("domain") == "cloudcode-pa.googleapis.com" and value.get("reason") == "QUOTA_EXHAUSTED"] + if error.get("errorCode") != 429 or len(quota) != 1 or not quota[0].get("metadata", {}).get("model"): + return None + return {"thread_id": cascade_id, "request_ref": request_ref, "execution_id": execution, + "input_created_at": metadata["createdAt"], "start_step_index": start, "total_steps": total, + "input_sha256": digest, "model_id": quota[0]["metadata"]["model"], + "pre_execution_quota_failure": True, "only_original_user_item": True, + "latest_input_verified": True, "observed_at": time.time()} + except (KeyError, TypeError, ValueError, AttributeError, CodexChatAgentError): + return None + @staticmethod def metadata_review_capacity(*, cascade_id: str, model_id: str) -> dict[str, Any] | None: """Observe the existing model's quota; never probe it with generation. diff --git a/loopx/capabilities/native_chat/external_conversations.py b/loopx/capabilities/native_chat/external_conversations.py index fd1b91e37c..b4078e6fdc 100644 --- a/loopx/capabilities/native_chat/external_conversations.py +++ b/loopx/capabilities/native_chat/external_conversations.py @@ -114,6 +114,10 @@ def _context_recovery_plan(self, row: dict[str, Any], *, "preparation_observation": preparation, "later_equivalent_input_present": later_equivalent, "preparation_scope_available": preparation_scope_available, + "later_source_input_present": any(item.get("binding_id") == row["binding_id"] + and item.get("source") == row["source"] and item.get("command") is None + and datetime.fromisoformat(item["created_at"]) > datetime.fromisoformat(row["created_at"]) + for item in self.pending()) if turn.get("error_code") == "antigravity_step_error" else None, "native_input": anchors[0] if len(anchors) == 1 else None, "response_literal_open": bool(open_literal), "metadata_capacity": metadata_capacity, @@ -250,12 +254,20 @@ def source_requests() -> list[dict[str, Any]]: # checks and the typed eligibility decision remain below that fence. proof = None capacity = None - if not plan.get("admitted_turn_id") and metadata_review: - proof = self._original_native_result(row, session, turn, context, include_raw_response=True, - include_retained_result=turn.get("status") == "completed") + if not plan.get("admitted_turn_id") and (metadata_review or turn.get("error_code") == "antigravity_step_error"): + from .antigravity_adapter import AntigravityAdapter, ANTIGRAVITY_DEFAULT_MODEL + events = self.controller.store.events_after(row["session_id"], row["turn_id"], None) + anchors = [item["payload"]["native_input"] for item in events + if item.get("kind") == "turn.started" and item.get("payload", {}).get("native_input")] + native_session = self._original_native_session(session, anchors) + if metadata_review: + proof = self._original_native_result(row, session, turn, context, include_raw_response=True, + include_retained_result=turn.get("status") == "completed") + elif len(anchors) == 1: + proof = AntigravityAdapter.quota_failure_observation(cascade_id=native_session["upstream_thread_id"], + request_ref=row["request_ref"], message=row["message"], started_at=turn["started_at"], + completed_at=turn["completed_at"], input_anchor=anchors[0]) if proof is not None: - from .antigravity_adapter import AntigravityAdapter, ANTIGRAVITY_DEFAULT_MODEL - native_session = self._original_native_session(session, [proof]) capacity = AntigravityAdapter.metadata_review_capacity( cascade_id=native_session["upstream_thread_id"], model_id=native_session.get("agent_model_id") or ANTIGRAVITY_DEFAULT_MODEL) @@ -342,6 +354,8 @@ def source_requests() -> list[dict[str, Any]]: recovery.update({f"{kind}_attempt": plan[f"{kind}_attempt"], "presentation_key": f"{MANAGER_CONTEXT_VERSION}.{kind}.{plan[f'{kind}_attempt']}", "not_before": plan["not_before"]}) + if plan.get("quota_executor"): + recovery["quota_executor"] = plan["quota_executor"] if preparation_retry: recovery["preparation_observation"] = plan["preparation_observation"] row["context_recovery"] = recovery diff --git a/loopx/chat_runtime.py b/loopx/chat_runtime.py index 0e058e1048..e77e631b23 100644 --- a/loopx/chat_runtime.py +++ b/loopx/chat_runtime.py @@ -832,14 +832,15 @@ def _ensure_adapter_locked( recovery = ChatExternalConversations(self).queued_recovery_context(session, active, accepted_turn_id) quota_route = None metadata_fallback = recovery.get("metadata_review") is True and recovery.get("metadata_executor") == "codex" + quota_fallback = bool(recovery.get("quota_attempt") and recovery.get("quota_executor") == "codex") if (manager_runtime is not None and active and active_id == accepted_turn_id and active.get("status") == "queued" and not active.get("upstream_turn_id") - and active.get("origin") != "native_loopx" and (not recovery or metadata_fallback) - and ("-recovery-" not in str(active.get("client_turn_id")) or metadata_fallback) + and active.get("origin") != "native_loopx" and (not recovery or metadata_fallback or quota_fallback) + and ("-recovery-" not in str(active.get("client_turn_id")) or metadata_fallback or quota_fallback) and session.get("upstream_mode") in {"default", "chat"}): quota_route = manager_session_quota_route(self, session) - if metadata_fallback and (not quota_route or quota_route["allocation"]["executor_endpoint"] != "codex"): - raise CodexChatAgentError("The reserved formatter no longer has verified Codex capacity.", error_code="manager_authorization_unavailable", gate=None) + if (metadata_fallback or quota_fallback) and (not quota_route or quota_route["allocation"]["executor_endpoint"] != "codex"): + raise CodexChatAgentError("The reserved recovery no longer has verified Codex capacity.", error_code="manager_authorization_unavailable", gate=None) if quota_route and quota_route["allocation"]["executor_endpoint"] == session.get("agent_id"): quota_route = None if quota_route is not None: @@ -971,6 +972,7 @@ def check_route_configuration(): session.get("manager_context_version") != MANAGER_CONTEXT_VERSION or manager_profile_changed or model_allocation is not None + or quota_fallback # Do not resume a prior tools-disabled formatter as business execution. ) ) legacy_codex_goal_thread = ( @@ -1096,6 +1098,7 @@ def check_route_configuration(): agent_model_id=adapter.current_model_id, **observed_settings, manager_quota_block=quota_route["blocked_primary"], metadata_review=metadata_fallback, + quota_recovery=quota_fallback, expected_upstream_thread_id=session["upstream_thread_id"]) if selected_agent == "codex": changes["codex_home"] = str(selected_codex_home) diff --git a/loopx/chat_store.py b/loopx/chat_store.py index 5c525959d8..8218adef2d 100644 --- a/loopx/chat_store.py +++ b/loopx/chat_store.py @@ -440,6 +440,7 @@ def reserve_loopx_outcome_request( def update_session(self, session_id: str, *, expected_statuses: set[str] | None = None, expected_upstream_thread_id: str | None = None, metadata_review: bool = False, + quota_recovery: bool = False, **changes: Any) -> dict[str, Any]: path = self._session_path(session_id) with self._session_lock(session_id): @@ -502,9 +503,10 @@ def update_session(self, session_id: str, *, expected_statuses: set[str] | None or payload.get("upstream_mode") not in {"default", "chat"} or not active or active.get("status") != "queued" or active.get("upstream_turn_id") or active.get("origin") == "native_loopx" - or ("-recovery-" in str(active.get("client_turn_id")) and not (metadata_review + or ("-recovery-" in str(active.get("client_turn_id")) and not ((metadata_review or quota_recovery) and payload.get("steward_context") is not None and endpoint == "codex" - and str(active.get("client_turn_id")).endswith("-metadata-1"))) + and str(active.get("client_turn_id")).endswith( + "-metadata-1" if metadata_review else "-quota-1"))) or allocation["selection_policy"] != "flexible" or allocation["configured_endpoint"] != "antigravity" or set(allocation["eligible_endpoints"]) != {"antigravity", "codex"} diff --git a/loopx/control_plane/collaboration/conversation_binding.ts b/loopx/control_plane/collaboration/conversation_binding.ts index 2844648db1..9a0eafc153 100644 --- a/loopx/control_plane/collaboration/conversation_binding.ts +++ b/loopx/control_plane/collaboration/conversation_binding.ts @@ -637,6 +637,9 @@ function planContextRecovery(params: JsonObject, row: JsonObject, current: JsonO || (Array.isArray(row.context_recovery_history) && row.context_recovery_history.some(value => requireJsonObject(value, "prior preparation recovery").preparation_attempt != null))) return no; if (current.agent_id === "antigravity") { + if (failed.error_code === "antigravity_step_error") { + return planNativeQuotaRecovery(recovery, row, current, failed, version); + } const originalResponse = failed.response && typeof failed.response === "object" && !Array.isArray(failed.response) ? failed.response as JsonObject : null; const anchor = recovery.native_input && typeof recovery.native_input === "object" && !Array.isArray(recovery.native_input) @@ -885,14 +888,72 @@ function planNativeMetadataReview(recovery: JsonObject, row: JsonObject, current targets.add(target); transcribed.push({goal_id: item.goal_id, agent_id: item.agent_id, brief}); } - return metadataReviewCapacity(recovery, row, current, {...candidate, metadata_handoffs: transcribed, + return nativeRecoveryCapacity(recovery, row, current, {...candidate, metadata_handoffs: transcribed, ...(adopted === null && recovery.source_verified === true ? {retain_metadata_result: true} : {})}); } +/** A quota error alone never permits replay. One exact native input/error pair, + * no later source input, fresh capacity and the original source permit a new + * queue Turn. Effects, ambiguous terminal state and missing anchors stay held. + */ +function planNativeQuotaRecovery(recovery: JsonObject, row: JsonObject, current: JsonObject, + failed: JsonObject, version: number): JsonObject { + const no = {operation: "admit_turn", recovery_candidate: false, recovery_eligible: false}; + const originalClient = `external-${ref(row.request_ref, "original request")}`; + if (failed.session_id !== current.session_id || failed.turn_id !== row.turn_id + || failed.client_turn_id !== originalClient || failed.message !== row.message || failed.origin !== row.origin + || failed.status !== "failed" || failed.response !== null || failed.delta_count !== 0 + || typeof failed.upstream_turn_id !== "string" || !failed.upstream_turn_id + || recovery.attachments_present !== false || recovery.steering_observed !== false) return no; + const client = `${originalClient}-recovery-${version}-quota-1`; + const previous = row.context_recovery == null ? null : requireJsonObject(row.context_recovery, "quota reservation"); + if (row.context_recovery_history != null + && (!Array.isArray(row.context_recovery_history) || row.context_recovery_history.length) + || previous !== null && (previous.version !== version || previous.quota_attempt !== 1 + || previous.failed_turn_id !== failed.turn_id || previous.client_turn_id !== client + || ["routing_attempt", "preparation_attempt", "read_only_review", "metadata_review", "work_attempt"] + .some(key => previous[key] != null))) return no; + if (recovery.admitted_turn != null && previous !== null) { + const admitted = requireJsonObject(recovery.admitted_turn, "admitted quota Turn"); + if (admitted.client_turn_id !== client || admitted.session_id !== current.session_id + || admitted.turn_id === failed.turn_id || admitted.message !== row.message || admitted.origin !== row.origin + || !Array.isArray(admitted.attachments ?? []) || (admitted.attachments as unknown[] | undefined)?.length) return no; + return {...no, recovery_candidate: true, recovery_eligible: recovery.source_verified === true, + client_turn_id: client, admitted_turn_id: sessionIdentity(admitted.turn_id, "admitted quota Turn"), + session_id: current.session_id, native_thread_ids: []}; + } + const now = Date.parse(String(recovery.observed_at)); + const notBefore = Date.parse(String(failed.completed_at)) + 60_000; + const anchor = recovery.native_input as JsonObject | null | undefined; + if (recovery.stop_observed !== false || recovery.later_source_input_present !== false + || current.status !== "ready" || current.active_turn_id != null || recovery.queued_input_present !== false + || !Number.isFinite(now) || !Number.isFinite(notBefore) || now < notBefore || !anchor + || anchor.thread_id !== current.upstream_thread_id || !Number.isSafeInteger(anchor.start_step_index) + || Number(anchor.start_step_index) < 0 || typeof anchor.input_sha256 !== "string" + || !BARE_SHA256_PATTERN.test(anchor.input_sha256)) return no; + const candidate = {...no, recovery_candidate: true, quota_attempt: 1, client_turn_id: client, + not_before: new Date(notBefore).toISOString(), failed_turn_id: failed.turn_id, session_id: current.session_id, + native_thread_ids: [current.upstream_thread_id]}; + const proof = recovery.native_observation as JsonObject | null | undefined; + if (!proof) return candidate; + const created = Date.parse(String(proof.input_created_at)); + const started = Date.parse(String(failed.started_at)), completed = Date.parse(String(failed.completed_at)); + if (proof.thread_id !== anchor.thread_id || proof.request_ref !== row.request_ref + || proof.start_step_index !== anchor.start_step_index || proof.input_sha256 !== anchor.input_sha256 + || proof.total_steps !== Number(anchor.start_step_index) + 2 || proof.latest_input_verified !== true + || proof.pre_execution_quota_failure !== true || proof.only_original_user_item !== true + || proof.model_id !== (current.agent_model_id ?? "gemini-4-argon-high") + || typeof proof.execution_id !== "string" || !proof.execution_id + || typeof proof.observed_at !== "number" || !Number.isFinite(proof.observed_at) + || now / 1000 < proof.observed_at || now / 1000 - proof.observed_at > 120 + || ![created, started, completed].every(Number.isFinite) || created < started || created > completed) return candidate; + return nativeRecoveryCapacity(recovery, row, current, candidate); +} + /** A model list/positive remainder is not proof a known quota failure cleared. * Keep the existing request pending without reserving its one format attempt. */ -function metadataReviewCapacity(recovery: JsonObject, row: JsonObject, current: JsonObject, +function nativeRecoveryCapacity(recovery: JsonObject, row: JsonObject, current: JsonObject, candidate: JsonObject): JsonObject { if (recovery.source_verified !== true) return candidate; const capacity = recovery.metadata_capacity as JsonObject | null | undefined; @@ -923,7 +984,8 @@ function metadataReviewCapacity(recovery: JsonObject, row: JsonObject, current: && configuration.eligible_endpoints.includes("antigravity") && configuration.eligible_endpoints.includes("codex") && typeof allocation.configuration_revision === "string" && allocation.configuration_revision.length > 0 && allocation.configuration_revision === configuration.configuration_revision) { - return {...candidate, recovery_eligible: true, metadata_executor: "codex"}; + return {...candidate, recovery_eligible: true, + ...(candidate.metadata_review === true ? {metadata_executor: "codex"} : {quota_executor: "codex"})}; } const waiting = row.metadata_quota_wait as JsonObject | null | undefined; if (waiting != null && (typeof waiting.model_id !== "string" || typeof waiting.remaining_fraction !== "number" diff --git a/tests/control_plane_ts/conversation_binding.test.ts b/tests/control_plane_ts/conversation_binding.test.ts index 2e345091f9..1e11b306d2 100644 --- a/tests/control_plane_ts/conversation_binding.test.ts +++ b/tests/control_plane_ts/conversation_binding.test.ts @@ -19,6 +19,71 @@ const current = {schema_version: "loopx_chat_conversation_bindings_v0", revision const request = {current, expected_revision: 0, operation: "configure", binding: row, observation, available_projects: [project]}; +test("an anchored pre-execution Argon quota failure resumes only through fresh original-source admission", () => { + const selected = {...row, context_kind: "steward", grant: "portfolio_read", goal_ids: [], executor_endpoint_id: "antigravity"}; + const configured = planConversationBinding({...request, binding: selected}).state; + const source = {source_ref: "e".repeat(24), sender_ref: row.operator_ref, private_human_message: true}; + const context = resolveBoundConversation({current: configured, binding_id: row.binding_id, + observation, available_projects: [project], ...source}).context; + const ref = "f".repeat(24), message = "Explain the original change"; + const session = {session_id: "original", goal_id: "loopx-manager", agent_id: "antigravity", + agent_model_id: "gemini-4-argon-high", upstream_mode: "default", status: "ready", upstream_thread_id: "native", + active_turn_id: null, channel_id: `manager.external.native.${row.binding_id}.${source.source_ref}`, steward_context: context}; + const failed = {session_id: "original", turn_id: "failed", client_turn_id: `external-${ref}`, message, origin: "lark", + status: "failed", delta_count: 0, response: null, upstream_turn_id: "native-failed", error_code: "antigravity_step_error", + started_at: "2026-01-01T00:00:00Z", completed_at: "2026-01-01T00:00:10Z"}; + const at = Date.parse("2026-01-01T00:10:00Z") / 1000; + const anchor = {thread_id: "native", start_step_index: 10, input_sha256: "a".repeat(64)}; + const proof = {...anchor, request_ref: ref, execution_id: "original-execution", total_steps: 12, + input_created_at: "2026-01-01T00:00:05Z", model_id: session.agent_model_id, + pre_execution_quota_failure: true, only_original_user_item: true, latest_input_verified: true, observed_at: at}; + const allocation = {schema_version: "manager_executor_allocation_v0", selection_policy: "flexible", + configured_endpoint: "antigravity", executor_endpoint: "codex", allocation_reason: "flexible_availability_fallback", + configuration_revision: "current-pool", available: true, eligible_endpoints: ["antigravity", "codex"]}; + const configuration = {status: "ready", selection_policy: "flexible", executor_endpoint: "antigravity", + eligible_endpoints: ["antigravity", "codex"], configuration_revision: "current-pool"}; + const capacity = {model_id: session.agent_model_id, account_ref: "c".repeat(64), remaining_fraction: 0, + reset_time: "2026-01-08T00:00:00Z", observed_at: at, quota_exhausted: true, normal_completion: false, + alternate_executor: {allocation, observed_at: at}}; + const packet = {request: {request_ref: ref, command: null, status: "accepted", delivery_verified: true, + message, origin: "lark", binding_id: row.binding_id, session_id: "original", turn_id: "failed", source}, + binding: selected, current_session: session, context_recovery: {version: 22, failed_turn: failed, + observed_at: "2026-01-01T00:10:00Z", steering_observed: false, stop_observed: false, attachments_present: false, + source_verified: true, queued_input_present: false, later_source_input_present: false, + native_input: anchor, native_observation: proof, metadata_capacity: capacity, executor_configuration: configuration}}; + const plan = planBoundConversationRequest(packet); + assert.equal(plan.recovery_eligible, true); + assert.equal(plan.quota_executor, "codex"); + assert.equal(plan.quota_attempt, 1); + assert.equal(plan.client_turn_id, `external-${ref}-recovery-22-quota-1`); + for (const change of [{source_verified: false}, {stop_observed: true}, {steering_observed: true}, + {attachments_present: true}, {queued_input_present: true}, {later_source_input_present: true}, + {native_input: null}, {native_observation: null}, {metadata_capacity: null}, + {executor_configuration: {...configuration, configuration_revision: "changed"}}, + {metadata_capacity: {...capacity, alternate_executor: null}}, + ...[{pre_execution_quota_failure: false}, {only_original_user_item: false}, {latest_input_verified: false}, + {total_steps: 13}, {input_sha256: "b".repeat(64)}, {request_ref: "foreign"}, {model_id: "other-model"}, + {observed_at: at - 121}, {observed_at: NaN}].map(value => ({native_observation: {...proof, ...value}})), + ...[{delta_count: 1}, {response: {}}, {client_turn_id: "foreign"}, {origin: "web"}, {started_at: "unknown"}] + .map(value => ({failed_turn: {...failed, ...value}}))]) { + assert.notEqual(planBoundConversationRequest({...packet, + context_recovery: {...packet.context_recovery, ...change}}).recovery_eligible, true); + } + const reserved = {...packet.request, context_recovery: {version: 22, failed_turn_id: "failed", + quota_attempt: 1, client_turn_id: plan.client_turn_id}}; + const admitted = {session_id: "original", turn_id: "continued", client_turn_id: plan.client_turn_id, message, origin: "lark"}; + const adopted = planBoundConversationRequest({...packet, request: reserved, + context_recovery: {...packet.context_recovery, native_observation: null, admitted_turn: admitted}}); + assert.equal(adopted.admitted_turn_id, "continued"); + assert.equal(adopted.recovery_eligible, true); + const correlation = planBoundConversationRequest({...packet, request: reserved, + current_session: {...session, status: "busy", active_turn_id: "newer"}, + context_recovery: {...packet.context_recovery, native_observation: null, stop_observed: true, + later_source_input_present: true, queued_input_present: true, admitted_turn: admitted}}); + assert.equal(correlation.admitted_turn_id, "continued", "Later control may not lose already admitted correlation"); + assert.equal(correlation.recovery_eligible, true); +}); + test("Core preparation failures resume through one bounded original-source queue, never from missing native evidence", () => { for (const endpoint of ["codex", "antigravity"]) { const selected = {...row, context_kind: "steward", grant: "portfolio_read", goal_ids: [], executor_endpoint_id: endpoint}; diff --git a/tests/test_antigravity_steward_adapter.py b/tests/test_antigravity_steward_adapter.py index ddf9985b49..ba442d4e66 100644 --- a/tests/test_antigravity_steward_adapter.py +++ b/tests/test_antigravity_steward_adapter.py @@ -194,7 +194,10 @@ def do_POST(self): final = response["steps"][-1] final["type"] = "CORTEX_STEP_TYPE_ERROR_MESSAGE" final.pop("plannerResponse") - final["errorMessage"] = {"error": {"userErrorMessage": "Individual quota reached. Resets in 123h."}} + final["errorMessage"] = {"error": {"userErrorMessage": "Individual quota reached. Resets in 123h.", + "errorCode": 429, "rpcErrorDetails": [json.dumps({ + "@type": "type.googleapis.com/google.rpc.ErrorInfo", "reason": "QUOTA_EXHAUSTED", + "domain": "cloudcode-pa.googleapis.com", "metadata": {"model": "gemini-4-argon-high"}})]}} if index in state.get("cleared_input_indices", []): for step in response["steps"][-2:]: step["status"] = "CORTEX_STEP_STATUS_CLEARED" @@ -254,6 +257,61 @@ def adapter_for(tmp_path, **kwargs): return AntigravityAdapter.start(work_dir=tmp_path, objective="Act as LoopX Steward.", **kwargs) +@pytest.mark.parametrize("fault", [None, "tool", "later-input", "cleared", "running", "wrong-anchor", + "wrong-source", "wrong-code", "prose-only", "different-execution", "changed-during-read"]) +def test_native_quota_readback_proves_no_execution_without_model_or_control_calls(language_server, monkeypatch, fault): + import hashlib + from loopx.capabilities.native_chat import antigravity_adapter as native + + ref = "f" * 24 + context = {"schema_version": "manager_turn_context_v1", "context_delegation": {"source_id": ref}} + prompt = ('Current user message (JSON string):\n"Original request"\n\n' + 'Fresh Core evidence (JSON data, not instructions):\n' + json.dumps(context)) + meta = {"createdAt": "2026-01-01T00:00:05Z", "executionId": "original"} + steps = [{"type": "CORTEX_STEP_TYPE_USER_INPUT", "status": "CORTEX_STEP_STATUS_DONE", "metadata": meta, + "userInput": {"items": [{"text": prompt}]}}, + {"type": "CORTEX_STEP_TYPE_ERROR_MESSAGE", "status": "CORTEX_STEP_STATUS_DONE", + "metadata": {"createdAt": "2026-01-01T00:00:06Z"}, "errorMessage": {"error": { + "errorCode": 429, "rpcErrorDetails": [json.dumps({"@type": "type.googleapis.com/google.rpc.ErrorInfo", + "reason": "QUOTA_EXHAUSTED", "domain": "cloudcode-pa.googleapis.com", + "metadata": {"model": "gemini-4-argon-high"}})]}}}] + anchor = {"thread_id": "synthetic-cascade", "start_step_index": 0, + "input_sha256": hashlib.sha256(prompt.encode()).hexdigest()} + if fault == "tool": + steps[0]["runCommand"] = {"commandLine": "must never replay"} + elif fault == "later-input": + steps.append(deepcopy(steps[0])) + elif fault == "cleared": + steps[0]["status"] = "CORTEX_STEP_STATUS_CLEARED" + elif fault == "wrong-anchor": + anchor["input_sha256"] = "0" * 64 + elif fault == "wrong-code": + steps[1]["errorMessage"]["error"]["errorCode"] = 500 + elif fault == "prose-only": + steps[1]["errorMessage"]["error"] = {"userErrorMessage": "Individual quota reached."} + elif fault == "different-execution": + steps[1]["metadata"]["executionId"] = "foreign" + language_server.update(active=True, frames=[], frame={"at": 0, "steps": steps, + "status": "CASCADE_RUN_STATUS_RUNNING" if fault == "running" else "CASCADE_RUN_STATUS_IDLE"}) + if fault == "changed-during-read": + post = native._rpc_post + def changed(*args, **kwargs): + result = post(*args, **kwargs) + if args[2] == "GetCascadeTrajectorySteps": + steps[0]["userInput"]["items"][0]["text"] += "Changed" + return result + monkeypatch.setattr(native, "_rpc_post", changed) + proof = AntigravityAdapter.quota_failure_observation(cascade_id="synthetic-cascade", + request_ref="foreign" if fault == "wrong-source" else ref, message="Original request", + started_at="2026-01-01T00:00:00Z", completed_at="2026-01-01T00:00:10Z", input_anchor=anchor) + assert (proof is not None) == (fault is None) + if proof: + assert proof["pre_execution_quota_failure"] and proof["only_original_user_item"] + assert proof["input_sha256"] == anchor["input_sha256"] + assert {method for method, _ in language_server["calls"]} <= { + "GetCascadeModelConfigData", "GetCascadeTrajectory", "GetCascadeTrajectorySteps"} + + def _visible_history(): return [{"role": "user" if index % 2 == 0 else "assistant", "content": (" EARLIEST-CONSTRAINT\r\n不要重复已完成的动作\t " if index == 0 else diff --git a/tests/test_chat_manager_model_adoption.py b/tests/test_chat_manager_model_adoption.py index d8a4674a18..1745a8b74d 100644 --- a/tests/test_chat_manager_model_adoption.py +++ b/tests/test_chat_manager_model_adoption.py @@ -433,6 +433,37 @@ def start(**kwargs): assert restored["channel_id"] == original["channel_id"] +def test_proven_quota_recovery_opens_fresh_context_when_codex_already_owns_the_session(conversation, monkeypatch): + from types import SimpleNamespace + import loopx.chat_runtime as runtime_module + from loopx.capabilities.native_chat.external_conversations import ChatExternalConversations + + root, store, runtime, session, starts = conversation + configure(root, model="gemini-4-argon-high", effort="high", endpoint="antigravity", policy="flexible") + capabilities = [{"agent_id": name, "available": name == "codex", "adapter_kind": kind} for name, kind in + [("antigravity", "antigravity_ls"), ("codex", "codex_app_server")]] + monkeypatch.setattr(runtime, "capabilities", lambda: capabilities) + allocation = manager_executor_allocation(runtime, environ={}, endpoint_capabilities=capabilities) + session = store.update_session(session["session_id"], manager_executor_allocation=allocation, + agent_model_id="gpt-6.1-sol") + assert runtime_module.manager_session_model_allocation(runtime, session) is None + store.append_message(session["session_id"], role="user", text="Keep the original source and completed work.") + turn, _ = store.create_queued_turn(session["session_id"], client_turn_id="external-original-recovery-22-quota-1", + message="Continue the proven unstarted request", origin="lark") + session = store.update_session(session["session_id"], active_turn_id=turn["turn_id"], status="busy") + runtime.project_contexts.conversation_bindings = SimpleNamespace() + monkeypatch.setattr(ChatExternalConversations, "queued_recovery_context", lambda *_: + {"quota_attempt": 1, "quota_executor": "codex"}) + monkeypatch.setattr(runtime_module, "manager_session_quota_route", lambda *_: + {"allocation": allocation, "blocked_primary": None}) + runtime._ensure_adapter(session, work_dir=root, objective="manager") + assert len(starts) == 2 and starts[0][1].closed + assert starts[1][0]["resume_thread_id"] is None # A formatter's native permissions cannot become business permissions. + assert starts[1][0]["read_only_review"] is False and starts[1][0]["metadata_review_context"] is None + assert starts[1][0]["history"] == [{"role": "user", "content": "Keep the original source and completed work."}] + assert store.load_turn(session["session_id"], turn["turn_id"])["status"] == "queued" + + @pytest.mark.parametrize("status", ["queued", "running"]) def test_native_account_epoch_reopens_only_unstarted_codex_adapter(conversation, monkeypatch, status): import sys diff --git a/tests/test_lark_private_conversations.py b/tests/test_lark_private_conversations.py index e637574c0b..7ab427fd10 100644 --- a/tests/test_lark_private_conversations.py +++ b/tests/test_lark_private_conversations.py @@ -19,6 +19,112 @@ 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"]) +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 + from pathlib import Path + from test_native_steward_private import steward + from test_chat_manager_model_adoption import configure + from loopx.chat_manager import manager_executor_allocation + import loopx.chat_manager as manager_module + import loopx.chat_runtime as runtime_module + from loopx.capabilities.native_chat import external_conversations, antigravity_adapter + + monkeypatch.setenv("LOOPX_MANAGER_MODEL", "gemini-4-argon-high") + monkeypatch.setenv("LOOPX_MANAGER_REASONING_EFFORT", "high") + fixture = steward.__wrapped__(tmp_path) + store, runtime, provider, transport, binding, capture, _ = next(fixture) + try: + binding = transport.bindings.configure(transport_ref="steward-app", project_ref=binding["project_ref"], + executor_endpoint_id="antigravity", context_kind="steward") + language_server.update(quota_error_indices=[0], quota_fraction=0) + event = provider.event("steward-app", "quota-before-execution", "Explain this change") + transport.admit("steward-app", event) + ref = identity_ref(binding["provider_ref"], event["message_id"]) + original = transport.core.read_request(ref) + failed = runtime.wait_for_turn(session_id=original["session_id"], turn_id=original["turn_id"], timeout_sec=15) + assert failed["status"] == "failed" and failed["error_code"] == "antigravity_step_error" + original_bytes = store._turn_path(original["session_id"], original["turn_id"]).read_bytes() + original_events = store.events_after(original["session_id"], original["turn_id"], None) + path = transport.root / f"{ref}.json" + transport.reconcile_request(path) # Preserve original failure delivery. + configure(runtime.coordination_runtime_root, model="gemini-4-argon-high", effort="high", + endpoint="antigravity", policy="flexible") + capabilities = [{"agent_id": name, "available": True, "adapter_kind": kind} for name, kind in + [("antigravity", "antigravity_ls"), ("codex", "codex_app_server")]] + monkeypatch.setattr(runtime, "capabilities", lambda: capabilities) + allocation = manager_executor_allocation(runtime, environ={}, endpoint_capabilities=[ + {**item, "available": item["agent_id"] == "codex"} for item in capabilities]) + fake = Path(runtime.codex_bin) + fake.write_text(fake.read_text().replace('result = {"thread": {"id": "durable-thread"}}', + 'result = {"thread": {"id": "durable-thread"}, "model": "gpt-6.1-sol"}')) + route = lambda *_: None if fault == "no-capacity" else {"allocation": allocation, + "blocked_primary": None, "observed_at": time.time()} + monkeypatch.setattr(manager_module, "manager_session_quota_route", route) + monkeypatch.setattr(runtime_module, "manager_session_quota_route", route) + class Ahead(datetime): + @classmethod + def now(cls, tz=None): + return super().now(tz) + timedelta(seconds=65) + monkeypatch.setattr(external_conversations, "datetime", Ahead) + update_session = store.update_session + transitions = [] + def checked_update(session_id, **changes): + if changes.get("quota_recovery"): + before = store.load_session(session_id) + without_proof = {key: value for key, value in changes.items() if key != "quota_recovery"} + with pytest.raises(ValueError, match="executor transition"): + update_session(session_id, **without_proof) + assert store.load_session(session_id) == before + transitions.append(session_id) + return update_session(session_id, **changes) + monkeypatch.setattr(store, "update_session", checked_update) + if fault == "stop": + 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": + provider.messages[event["message_id"]]["content"] = "Changed original input" + elif fault == "later-source": + transport.admit("steward-app", provider.event("steward-app", "source-correction", "/status")) + # A later ordinary source is observed without creating model work. + later = deepcopy(original) + later.update(request_ref="e" * 24, created_at=(datetime.now().astimezone() + timedelta(seconds=1)).isoformat()) + from loopx.chat_store import _atomic_write_json + _atomic_write_json(transport.core.root / f"{later['request_ref']}.json", later) + elif fault == "native-tool": + post = antigravity_adapter._rpc_post + def with_tool(*args, **kwargs): + result = post(*args, **kwargs) + if args[2] == "GetCascadeTrajectorySteps": + result["steps"][0]["runCommand"] = {"commandLine": "unknown effects"} + return result + monkeypatch.setattr(antigravity_adapter, "_rpc_post", with_tool) + transport.reconcile_request(path) + recovered = transport.core.read_request(ref) + assert store._turn_path(original["session_id"], original["turn_id"]).read_bytes() == original_bytes + assert store.events_after(original["session_id"], original["turn_id"], None) == original_events + if fault: + assert recovered["turn_id"] == original["turn_id"] + assert not recovered.get("context_recovery") + else: + assert recovered["context_recovery"]["quota_executor"] == "codex" + done = runtime.wait_for_turn(session_id=recovered["session_id"], turn_id=recovered["turn_id"], timeout_sec=15) + assert done["status"] == "completed", done + transport.reconcile_request(path) + assert transport.core.read_request(ref)["delivery_verified"] + session = store.load_session(recovered["session_id"]) + assert session["agent_id"] == "codex" and session["agent_model_id"] == "gpt-6.1-sol" + assert transitions == [session["session_id"]] + after = capture.read_bytes() + transport.reconcile_request(path) + assert capture.read_bytes() == after # Original worker reentry cannot dispatch twice. + assert language_server["sent_count"] == 1 # The quota-failed Argon business is never sent again. + finally: + runtime.close() + fixture.close() + + class Provider: def __init__(self): self.messages = {}