fix(chat): keep report quota checks outside publication locks

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"]:

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"])