From dc47894fc1141ca2789a3b2a29a1cc8ec16e6759 Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Fri, 30 Oct 2026 12:50:04 +0800 Subject: [PATCH] fix(chat): recover report quota refusals across retained contexts Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> --- .../capabilities/native_chat/steward_turn.py | 10 ++- .../collaboration/steward_managed_mode.ts | 12 ++- tests/test_native_steward_reports.py | 78 +++++++++++++++++-- 3 files changed, 93 insertions(+), 7 deletions(-) diff --git a/loopx/capabilities/native_chat/steward_turn.py b/loopx/capabilities/native_chat/steward_turn.py index ec6137d3ae..623ac4c0b6 100644 --- a/loopx/capabilities/native_chat/steward_turn.py +++ b/loopx/capabilities/native_chat/steward_turn.py @@ -232,10 +232,18 @@ def report_quota_request(self, original: Mapping[str, Any]) -> dict[str, Any] | session = self.store.load_session(original["session_id"]) if not session or not session.get("codex_home") or not session.get("upstream_thread_id"): return None + history = session.get("upstream_context_history") or [] + if not isinstance(history, list) or any(not isinstance(item, dict) + or not isinstance(item.get("thread_id"), str) or not item["thread_id"] + for item in history): + return None + # Context replacement retains native provenance on this same Session. + # Read the original refusal there; never resume a historical thread. + threads = [session["upstream_thread_id"], *(item["thread_id"] for item in reversed(history))] proof = CodexAppServerAdapter.failed_request_observation(codex_bin=self.controller.codex_bin, codex_home=Path(session["codex_home"]), model_auth_home=checked_session_home(self.controller.codex_home, session), - thread_ids=[session["upstream_thread_id"]], turn_id=original["upstream_turn_id"]) + thread_ids=list(dict.fromkeys(threads)), turn_id=original["upstream_turn_id"]) if (not proof or proof.get("only_original_user_item") is not True or proof.get("codex_error_info") != "usageLimitExceeded" or proof.get("quota_available") is not True): return None diff --git a/loopx/control_plane/collaboration/steward_managed_mode.ts b/loopx/control_plane/collaboration/steward_managed_mode.ts index 64bf60096e..e64523a058 100644 --- a/loopx/control_plane/collaboration/steward_managed_mode.ts +++ b/loopx/control_plane/collaboration/steward_managed_mode.ts @@ -96,6 +96,16 @@ function admitReportQuotaRecovery(params: JsonObject, request: JsonObject, sessi const lineage = requireJsonObject(receipt.lineage, "original report lineage"); const keys = Object.keys(request).filter(key => key !== "report_quota_recovery"); const accepting = params.turn == null && params.report_settlement_recovery == null; + const history = session.upstream_context_history ?? []; + if (!Array.isArray(history)) throw new EffectRuntimeRequestError("invalid native context provenance"); + const nativeThreads = [session.upstream_thread_id, ...history.map(value => { + const item = requireJsonObject(value, "native context provenance"); + if (item.manager_context_version !== null && (!Number.isSafeInteger(item.manager_context_version) + || Number(item.manager_context_version) < 0)) { + throw new EffectRuntimeRequestError("invalid native context provenance"); + } + return requireNonEmptyString(item.thread_id, "original native context thread"); + })]; if (request.work_kind !== "report" || prior.work_kind !== "report" || prior.report_quota_recovery !== undefined || request.resume_turn_key !== undefined || prior.resume_turn_key !== undefined || Object.keys(recovery).sort().join() !== "native_observation,original_turn_id" @@ -119,7 +129,7 @@ function admitReportQuotaRecovery(params: JsonObject, request: JsonObject, sessi || ["goal_matches", "owner_matches", "turn_key_matches", "journal_consistent"].some(key => inspection[key] !== true) || inspection.journal_status !== "failed" || ![null, true].includes(recorded.host_invoked as null | boolean) - || observation.thread_id !== session.upstream_thread_id || typeof observation.thread_id !== "string" || !observation.thread_id + || !nativeThreads.includes(observation.thread_id) || typeof observation.thread_id !== "string" || !observation.thread_id || observation.turn_id !== original.upstream_turn_id || observation.only_original_user_item !== true || observation.codex_error_info !== "usageLimitExceeded" || observation.quota_available !== true || typeof observation.quota_observed_at !== "number" || !Number.isFinite(observation.quota_observed_at) diff --git a/tests/test_native_steward_reports.py b/tests/test_native_steward_reports.py index 25acd3ff8f..8183d5add0 100644 --- a/tests/test_native_steward_reports.py +++ b/tests/test_native_steward_reports.py @@ -507,8 +507,9 @@ def install_model(steward, tmp_path, sid, *, result_kind="validated_completion") @pytest.mark.parametrize("provider", ["file", "sqlite"]) @pytest.mark.parametrize("local_session", [False, True]) @pytest.mark.parametrize("recovery_succeeds", [False, True]) +@pytest.mark.parametrize("prior_context", [False, True], ids=["current_context", "prior_context"]) def test_report_quota_failure_continues_once_after_native_no_effects_and_available_allowance( - steward, tmp_path, monkeypatch, provider, local_session, recovery_succeeds): + steward, tmp_path, monkeypatch, provider, local_session, recovery_succeeds, prior_context): import time from datetime import timezone from loopx.chat_agent import CodexChatAgentError @@ -570,14 +571,28 @@ def exhausted(adapter, _prompt, event_sink): assert not reports.health(owner)["execution_failed"] assert frozen == (original_path.read_bytes(), journal_path.read_bytes()) observations = [] - proof = {"thread_id": store.load_session(sid)["upstream_thread_id"], + session = store.load_session(sid) + refused_thread = session["upstream_thread_id"] + if prior_context: + # The native host resumes the requested thread, rather than the base + # fixture's fixed id. Keep context identity checks enabled. + fake = Path(runtime.codex_bin) + fake.write_text(fake.read_text().replace('"durable-thread"', + 'request.get("params", {}).get("threadId", "durable-thread")')) + store.update_session(sid, upstream_context_history=[*(session.get("upstream_context_history") or []), + {"thread_id": refused_thread, "manager_context_version": session.get("manager_context_version")}], + upstream_thread_id="native-context-after-quota-refusal") + expected_threads = [store.load_session(sid)["upstream_thread_id"]] + if prior_context: + expected_threads.append(refused_thread) + proof = {"thread_id": refused_thread, "turn_id": original["upstream_turn_id"], "only_original_user_item": True, "codex_error_info": "usageLimitExceeded", "quota_available": False, "quota_observed_at": time.time()} def observe(**kwargs): observations.append(kwargs) - return dict(proof) + return dict(proof) if refused_thread in kwargs["thread_ids"] else None monkeypatch.setattr(CodexAppServerAdapter, "failed_request_observation", staticmethod(observe)) runtime._steward_monitor_next_check.clear() @@ -590,7 +605,8 @@ def observe(**kwargs): old_adapter = runtime.adapters[sid] old_thread = store.load_session(sid)["upstream_thread_id"] continued = pump_steward_monitors(runtime) - assert len(continued) == 1, "the original report must not remain stranded after proven quota recovery" + assert len(continued) == 1, ("the original report must not remain stranded after proven quota recovery; " + f"native contexts observed: {[row['thread_ids'] for row in observations]}") result = finish(runtime, continued[0]) assert runtime.adapters[sid] is not old_adapter assert old_adapter.session.process.poll() is not None @@ -605,7 +621,7 @@ def observe(**kwargs): assert reports.health(owner)["execution_failed"] is not recovery_succeeds assert frozen == (original_path.read_bytes(), journal_path.read_bytes()) assert observations and all(x["turn_id"] == original["upstream_turn_id"] for x in observations) - assert all(x["thread_ids"] == [proof["thread_id"]] for x in observations) + assert all(x["thread_ids"] == expected_threads for x in observations) if not recovery_succeeds: assert not reports.health(owner)["publication_verified"] assert len(calls) == 2 @@ -634,6 +650,58 @@ def observe(**kwargs): assert frozen == (original_path.read_bytes(), journal_path.read_bytes()) +@pytest.mark.parametrize("provider", ["file", "sqlite"]) +@pytest.mark.parametrize("unsafe_proof", ["foreign_thread", "observed_effects"]) +def test_report_quota_context_history_does_not_admit_foreign_or_effectful_failure( + steward, tmp_path, monkeypatch, provider, unsafe_proof): + import time + from loopx.chat_agent import CodexChatAgentError + from loopx.capabilities.native_chat.codex_adapter import CodexAppServerAdapter + from loopx.control_plane.turn_driver.journal_store import turn_journal_path + + sid, actor, reports, _ = configured(steward, tmp_path, monkeypatch, provider) + store, runtime, _, _, binding, capture, _ = steward + calls = [] + + def exhausted(adapter, _prompt, event_sink): + calls.append(adapter.upstream_thread_id) + event_sink("turn.started", {"upstream_turn_id": "quota-refused-native-turn"}) + raise CodexChatAgentError("The provider allowance was exhausted.") + + monkeypatch.setattr(CodexAppServerAdapter, "start_turn", exhausted) + original = finish(runtime, pump_steward_monitors(runtime)[0]) + native = original["response"]["native_loopx"] + assert native["status"] == "failed" and native["result_kind"] == "host_failure" + session = store.load_session(sid) + refused_thread = session["upstream_thread_id"] + fake = Path(runtime.codex_bin) + fake.write_text(fake.read_text().replace('"durable-thread"', + 'request.get("params", {}).get("threadId", "durable-thread")')) + store.update_session(sid, upstream_context_history=[*(session.get("upstream_context_history") or []), + {"thread_id": refused_thread, "manager_context_version": session.get("manager_context_version")}], + upstream_thread_id="native-context-after-quota-refusal") + journal = turn_journal_path(reports.root, goal_id=actor["goal_id"], turn_key=native["resume_turn_key"]) + turn_folder = store._turn_path(sid, original["turn_id"]).parent + before = {path.name: path.read_bytes() for path in turn_folder.glob("*.json")} + frozen, messages, model_calls = journal.read_bytes(), store.messages(sid), capture.read_bytes() + observed = [] + + def observe(**kwargs): + observed.append(kwargs) + return {"thread_id": "unrecorded-foreign-thread" if unsafe_proof == "foreign_thread" else refused_thread, + "turn_id": original["upstream_turn_id"], "only_original_user_item": unsafe_proof != "observed_effects", + "codex_error_info": "usageLimitExceeded", "quota_available": True, "quota_observed_at": time.time()} + + monkeypatch.setattr(CodexAppServerAdapter, "failed_request_observation", staticmethod(observe)) + runtime._steward_monitor_next_check.clear() + assert pump_steward_monitors(runtime) == [] + assert len(calls) == 1 and observed + assert {path.name: path.read_bytes() for path in turn_folder.glob("*.json")} == before + assert (journal.read_bytes(), store.messages(sid), capture.read_bytes()) == (frozen, messages, model_calls) + assert reports.health(binding["binding_id"])["execution_failed"] + assert not reports.health(binding["binding_id"])["publication_verified"] + + @pytest.mark.parametrize("provider", ["file", "sqlite"]) @pytest.mark.parametrize("change", ["unrelated_input", "read_counter", "queued_input", "binding_scope", "original_changed", "original_changed_after_identity"])