From 7b64579ec432da987b095cd292d5730d1727ed0f Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Fri, 30 Oct 2026 10:42:49 +0800 Subject: [PATCH] fix(chat): observe report quota outside publication locks Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> --- .../capabilities/native_chat/steward_turn.py | 29 ++++- tests/test_native_steward_reports.py | 111 ++++++++++++++++++ 2 files changed, 135 insertions(+), 5 deletions(-) diff --git a/loopx/capabilities/native_chat/steward_turn.py b/loopx/capabilities/native_chat/steward_turn.py index ed279a2dae..ec6137d3ae 100644 --- a/loopx/capabilities/native_chat/steward_turn.py +++ b/loopx/capabilities/native_chat/steward_turn.py @@ -445,7 +445,8 @@ def no_model(*_args): def _prepare_submission(self, binding_id: str, *, turn_instance_id: str, todo_id: str | None, validation_command: Sequence[str] | None, work_kind: str, - monitor_deadline: str | None, report_window: dict[str, Any] | None) -> dict[str, Any]: + monitor_deadline: str | None, report_window: dict[str, Any] | None, + quota_observation: dict[str, Any] | None = None) -> dict[str, Any]: """Read original request/recovery identity without accepting any work.""" binding = next((row for row in self.bindings.read()["bindings"] if row["binding_id"] == binding_id), None) if binding is None: @@ -464,6 +465,16 @@ def _prepare_submission(self, binding_id: str, *, turn_instance_id: str, todo_id report_subscription_revision=report_window["subscription_revision"]) client_id = native_steward_client_id(actor, turn_instance_id) existing = self.store.turn_for_client(actor["session_id"], client_id) + if quota_observation is not None: + if binding != quota_observation["binding"]: + raise ValueError("the original steward authorization changed during quota observation") + # Refence execution evidence after slow metadata IO; ordinary SSE + # read counters do not change the original request or its effects. + original = quota_observation["original"] + if existing is None or any(existing.get(key) != original.get(key) for key in ( + "turn_id", "client_turn_id", "status", "loopx_request", "response", + "upstream_turn_id", "delta_count", "error_code", "error", "loopx_execution")): + raise ValueError("the original report changed during quota observation") launch_recovery = None selection_recovery = None if existing is not None: @@ -478,7 +489,9 @@ def _prepare_submission(self, binding_id: str, *, turn_instance_id: str, todo_id retained = report_turn(self.store, actor, report_window) if retained is not None and retained["turn_id"] != existing["turn_id"]: return {"turn": retained} - quota_request = self.report_quota_request(existing) + if quota_observation is None: + return {"turn": existing, "quota_probe": {"original": existing, "binding": binding}} + quota_request = quota_observation["request"] if quota_request is not None: request = quota_request client_id = native_steward_client_id(actor, request["turn_instance_id"]) @@ -538,8 +551,14 @@ def submit(self, binding_id: str, *, turn_instance_id: str, todo_id: str | None monitor_deadline=monitor_deadline, report_window=report_window) with controller.lock, exclusive_file_lock(self.bindings.path, operation="inspect_native_steward_turn"): prepared = self._prepare_submission(binding_id, **parameters) - if "turn" in prepared: - return prepared["turn"], False + quota_observation = None + if "quota_probe" in prepared: + quota_observation = prepared["quota_probe"] + quota_observation["request"] = self.report_quota_request(quota_observation["original"]) + with controller.lock, exclusive_file_lock(self.bindings.path, operation="inspect_native_steward_turn"): + prepared = self._prepare_submission(binding_id, **parameters, quota_observation=quota_observation) + if "turn" in prepared: + return prepared["turn"], False observed = self._observe(prepared["request"]["actor"]) retained = None if work_kind == "monitor": @@ -549,7 +568,7 @@ def submit(self, binding_id: str, *, turn_instance_id: str, todo_id: str | None retained = native_command(controller, ["task-lease", "inspect", "--goal-id", prepared["request"]["actor"]["goal_id"], "--todo-id", todo_id]).get("lease") with controller.lock, exclusive_file_lock(self.bindings.path, operation="admit_native_steward_turn"): - current = self._prepare_submission(binding_id, **parameters) + current = self._prepare_submission(binding_id, **parameters, quota_observation=quota_observation) if "turn" in current: return current["turn"], False if current["request"] != prepared["request"] or current["client_id"] != prepared["client_id"]: diff --git a/tests/test_native_steward_reports.py b/tests/test_native_steward_reports.py index 5523518ccf..25acd3ff8f 100644 --- a/tests/test_native_steward_reports.py +++ b/tests/test_native_steward_reports.py @@ -634,6 +634,117 @@ def observe(**kwargs): assert frozen == (original_path.read_bytes(), journal_path.read_bytes()) +@pytest.mark.parametrize("provider", ["file", "sqlite"]) +@pytest.mark.parametrize("change", ["unrelated_input", "read_counter", "queued_input", "binding_scope", + "original_changed", "original_changed_after_identity"]) +def test_report_quota_observation_keeps_publication_available_and_rechecks_original( + steward, tmp_path, monkeypatch, provider, change): + import copy + import threading + import time + from loopx.chat_agent import CodexChatAgentError + from loopx.file_lock import exclusive_file_lock + from loopx.capabilities.native_chat.codex_adapter import CodexAppServerAdapter + from loopx.capabilities.native_chat.steward_turn import StewardTurnHost, native_steward_client_id + from loopx.control_plane.turn_driver.journal_store import turn_journal_path + + sid, actor, reports, _ = configured(steward, tmp_path, monkeypatch, provider) + store, runtime, _, transport, binding, capture, _ = steward + install_model(steward, tmp_path, sid) + + def exhausted(_adapter, _prompt, event_sink): + 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["host_failure"]["kind"] == "unknown" + _, window = reports.current_window(binding["binding_id"]) + journal = turn_journal_path(reports.root, goal_id=actor["goal_id"], turn_key=native["resume_turn_key"]) + original_journal, calls = journal.read_bytes(), capture.read_bytes() + other = store.create_session(goal_id=actor["goal_id"], agent_id="codex", adapter_kind="codex_app_server", + upstream_thread_id="unrelated-user-thread") + entered, released = threading.Event(), threading.Event() + results, errors, observations = [], [], [] + proof = {"thread_id": store.load_session(sid)["upstream_thread_id"], + "turn_id": original["upstream_turn_id"], "only_original_user_item": True, + "codex_error_info": "usageLimitExceeded", "quota_available": change not in {"unrelated_input", "read_counter"}, + "quota_observed_at": time.time()} + + def observe(**_kwargs): + observations.append(_kwargs) + entered.set() + assert released.wait(10), "quota observation fixture was not released" + return dict(proof) + + def change_original(): + changed = copy.deepcopy(original["response"]) + changed["native_loopx"]["reason"] = "Retained source changed during observation" + store.update_turn(sid, original["turn_id"], response=changed) + + observe_identity = StewardTurnHost._observe + + def identity(host, actor): + observed = observe_identity(host, actor) + if change == "original_changed_after_identity": + with runtime.lock: + change_original() + return observed + + def submit(): + try: + request = original["loopx_request"] + results.append(StewardTurnHost(runtime).submit(binding["binding_id"], report_window=window, + **{key: request[key] for key in ("turn_instance_id", "todo_id", "work_kind", "validation_command")})) + except Exception as exc: + errors.append(exc) + + monkeypatch.setattr(CodexAppServerAdapter, "failed_request_observation", staticmethod(observe)) + monkeypatch.setattr(StewardTurnHost, "_observe", identity) + worker = threading.Thread(target=submit) + worker.start() + available = False + queued = None + try: + assert entered.wait(5) + available = runtime.lock.acquire(timeout=.5) + if available: + with exclusive_file_lock(transport.bindings.path, timeout_seconds=.5): + if change in {"unrelated_input", "queued_input"}: + queued, created = store.create_queued_turn(other["session_id"] if change == "unrelated_input" else sid, + client_turn_id="human-during-quota", message="Keep this ordinary input available", origin="web") + assert created + elif change == "original_changed": + change_original() + elif change == "read_counter": + store.update_turn(sid, original["turn_id"], + sse_reconnect_count=original["sse_reconnect_count"] + 1) + if change == "binding_scope": + transport.bindings.configure(transport_ref="steward-app", project_ref=binding["project_ref"], + executor_endpoint_id="codex", context_kind="steward", goal_scope="selected") + finally: + if available: + runtime.lock.release() + else: + proof["quota_available"] = False # A failing baseline must not start model work. + released.set() + worker.join(10) + assert not worker.is_alive() + assert available, "quota metadata IO prevented ordinary input and binding publication" + assert len(observations) == 1 + assert journal.read_bytes() == original_journal and capture.read_bytes() == calls + continued = native_steward_client_id(actor, original["loopx_request"]["turn_instance_id"] + ".quota-recovery") + assert store.turn_for_client(sid, continued) is None + if change in {"unrelated_input", "read_counter"}: + assert not errors and results == [(store.load_turn(sid, original["turn_id"]), False)] + if queued: + assert store.load_turn(other["session_id"], queued["turn_id"])["status"] == "queued" + else: + assert len(errors) == 1 and isinstance(errors[0], ValueError), errors + assert results == [] + + @pytest.mark.parametrize("provider", ["file", "sqlite"]) @pytest.mark.parametrize("local_session", [False, True]) @pytest.mark.parametrize("replan_outcome", ["settled", "wait", "invalid_semantic", "already_admitted", "retained_probe"])