fix(reports): continue proven quota refusals once

docs/product/periodic-reports.md

@@ -23,6 +23,23 @@ The full frozen source, authored response, bundle, candidate and body must still
 match; an unsupported or altered presentation remains unavailable. New reports
 always use the shared source-link rule above.
 
+A report refused because its model allowance was exhausted can receive one
+new, correlated Turn when the original host independently reads a stable native
+failure containing only its original user input and observes usable allowance
+for the current authentication. A reset time or successful authentication RPC
+alone does not qualify. The shared typed conversation owner also checks the
+original failed kernel receipt, unchanged actor, subscription, window and Task,
+and the absence of any authored response, generation or delivery attempt.
+The original Chat Turn and failed kernel journal remain unchanged. The new Turn
+uses ordinary admission, lease, quota, validation and original-route publication;
+user work retains priority. A second failure is retained, never chained into
+another automatic attempt. Unknown effects, unavailable native metadata,
+revocation, cancellation and an unavailable allowance retain the existing hold.
+The admitted continuation refreshes only its own native adapter to reload
+current authentication while retaining its native thread. This applies to
+original steward reports and ordinary managed Session reports; authentication
+selection and unrelated clients retain their existing owners.
+
 The following synthetic reports use the production formatters. Text views use
 the compiled App Markdown component and its stylesheet; HTML views use the
 standalone report artifact. The same normalized records retain their outcomes,

loopx/capabilities/native_chat/steward_managed_mode.py

@@ -255,6 +255,7 @@ def admit(self, request: dict[str, Any], *, turn: dict[str, Any] | None = None,
               settlement_recovery: dict[str, Any] | None = None,
               selection_recovery: dict[str, Any] | None = None) -> dict[str, Any]:
         from .conversation_bindings import ChatConversationBindings
+        from .steward_turn import StewardTurnHost
         if self.controller.closed.is_set():
             raise ValueError("the original Chat host is stopped")
         actor = request["actor"]
@@ -284,6 +285,7 @@ def admit(self, request: dict[str, Any], *, turn: dict[str, Any] | None = None,
             "monitor_todo": monitor, **report,
             **({"report_settlement_recovery": settlement_recovery} if settlement_recovery is not None else {}),
             **({"report_selection_recovery": selection_recovery} if selection_recovery is not None else {}),
+            **StewardTurnHost(self.controller).report_quota_basis(request),
             "queued_turn_count": len(self.store.queued_turns(actor["session_id"]))})
 
     def inspect(self, session_id: str, *, agent_id: str | None = None,
@@ -351,7 +353,19 @@ def submit(self, session_id: str, *, agent_id: str, operation_id: str,
                 if frozen != request:
                     raise ValueError("managed Turn request identity conflict; inspect the original operation")
                 from .steward_turn import StewardTurnHost
-                selection_recovery = StewardTurnHost(controller).report_selection_recovery(existing) if report_window is not None else None
+                host = StewardTurnHost(controller)
+                if report_window is not None:
+                    from .steward_turn import report_turn
+                    retained = report_turn(self.store, request["actor"], report_window)
+                    if retained is not None and retained["turn_id"] != existing["turn_id"]:
+                        return retained, False
+                    quota_request = host.report_quota_request(existing)
+                    if quota_request is not None:
+                        request = quota_request
+                        client_id += ".quota-recovery"
+                        existing = None
+                selection_recovery = host.report_selection_recovery(existing) if existing is not None and report_window is not None else None
+            if existing is not None:
                 if selection_recovery is None:
                     return existing, False  # Failed/unknown work is never replayed.
                 client_id += ".admission-retry"

loopx/capabilities/native_chat/steward_reports.py

@@ -73,13 +73,12 @@ def _published_window(self, window: dict[str, Any]) -> bool:
         return bool(cursor and trigger["selected_trigger_id"] in cursor["covered_trigger_ids"])
 
     def _known_generation_failure(self, actor: dict[str, Any], window: dict[str, Any]) -> dict[str, str] | None:
-        from .steward_turn import native_steward_client_id
+        from .steward_turn import report_turn
         folder = attempt_directory(self.root, window)
         if (not (folder / "editorial-response.json").is_file()
                 or (folder / "private-generation.json").exists() or (folder / "private-delivery.json").exists()):
             return None
-        turn = self.controller.store.turn_for_client(actor["session_id"],
-            native_steward_client_id(actor, "report." + window["window_id"]))
+        turn = report_turn(self.controller.store, actor, window)
         request = (turn or {}).get("loopx_request") or {}
         native = ((turn or {}).get("response") or {}).get("native_loopx") or {}
         if (turn and turn.get("status") == "completed" and native.get("status") == "failed"
@@ -519,9 +518,9 @@ def publish(self, session_id: str, window: dict[str, Any]) -> bool:
         import hashlib
         from ..periodic_report.private_generation import validate_private_generation
         from ..periodic_report.incremental import commit_periodic_report_publication_cursor
-        from .steward_turn import native_steward_client_id
+        from .steward_turn import report_turn
         actor = self.selected(session_id)["actor"]
-        turn = self.controller.store.turn_for_client(session_id, native_steward_client_id(actor, "report." + window["window_id"]))
+        turn = report_turn(self.controller.store, actor, window)
         if not turn:
             return False
         request = turn.get("loopx_request") or {}
@@ -574,9 +573,8 @@ def health(self, session_id: str) -> dict[str, Any]:
             journal = read_cadence_journal(runtime_root=self.root, goal_id=actor["goal_id"])
             window = journal["window"] if journal else None
             folder = attempt_directory(self.root, window) if window else None
-            from .steward_turn import native_steward_client_id
-            turn = self.controller.store.turn_for_client(session_id,
-                native_steward_client_id(actor, "report." + window["window_id"])) if window else None
+            from .steward_turn import report_turn
+            turn = report_turn(self.controller.store, actor, window) if window else None
             native = ((turn or {}).get("response") or {}).get("native_loopx") or {}
             failed = bool(turn and (turn["status"] in {"failed", "interrupted", "timed_out"}
                 or native.get("status") == "failed"))

loopx/capabilities/native_chat/steward_turn.py

@@ -1,8 +1,9 @@
 """Original-Session app-server transport for the existing governed Turn driver.
 
-No scheduler, task store, consumer or CLI process is created here. The caller
+No scheduler, task store or message consumer is created here. The caller
 selects native work; the existing Turn command owns fresh admission, validation,
-writeback, quota and scheduler handoff. Chat retains its own execution fence.
+writeback, quota and scheduler handoff. Provider metadata uses the existing
+read-only native control client. Chat retains its own execution fence.
 """
 from __future__ import annotations
 
@@ -25,6 +26,20 @@ def native_steward_client_id(actor: Mapping[str, Any], turn_instance_id: str) ->
     return "native." + hashlib.sha256(json.dumps([actor, turn_instance_id], sort_keys=True).encode()).hexdigest()[:40]
 
 
+def report_turn(store: Any, actor: Mapping[str, Any], window: Mapping[str, Any]) -> dict[str, Any] | None:
+    """Read the one bounded continuation, retaining the original failed Turn."""
+    operation = "report." + window["window_id"]
+    original = store.turn_for_client(actor["session_id"], native_steward_client_id(actor, operation))
+    continued = store.turn_for_client(actor["session_id"], native_steward_client_id(actor, operation + ".quota-recovery"))
+    if continued is not None:
+        request = continued.get("loopx_request") or {}
+        if (not original or (request.get("report_quota_recovery") or {}).get("original_turn_id") != original["turn_id"]
+                or request.get("turn_instance_id") != original["loopx_request"]["turn_instance_id"] + ".quota-recovery"):
+            raise ValueError("the report quota continuation changed its original identity")
+        return continued
+    return original
+
+
 class StewardTurnHost:
     def __init__(self, controller: Any) -> None:
         self.controller = controller
@@ -96,11 +111,57 @@ def scope_id(ids):
             **({"report_settlement_recovery": settlement_recovery} if settlement_recovery is not None else {}),
             **({"report_launch_recovery": launch_recovery} if launch_recovery is not None else {}),
             **({"report_selection_recovery": selection_recovery} if selection_recovery is not None else {}),
+            **self.report_quota_basis(request),
             **({"preparing_model_context": True} if prepare_context else {}), **report})
         if admitted["refresh_original_scope"]:
             self.store.update_session(session["session_id"], manager_authorization_scope_id=scope["current"])
         return {**admitted, "observed_scope_id": scope["current"] if scope else None}
 
+    def report_quota_basis(self, request: Mapping[str, Any]) -> dict[str, Any]:
+        """Transport original journal observations; the typed owner admits work."""
+        recovery = request.get("report_quota_recovery")
+        if recovery is None:
+            return {}
+        from ...control_plane.turn_driver import inspect_loopx_turn_journal
+        from ..periodic_report.private_generation import attempt_directory
+        from .steward_reports import report_service
+        import time
+        actor = request["actor"]
+        original = self.store.load_turn(actor["session_id"], recovery["original_turn_id"])
+        result = ((original or {}).get("response") or {}).get("native_loopx") or {}
+        reports, owner = report_service(self.controller, actor)
+        _, window = reports.current_window(owner)
+        folder = attempt_directory(reports.root, window) if window else None
+        return {"report_quota_basis": {"original": original,
+            "inspection": inspect_loopx_turn_journal(reports.root, goal_id=actor["goal_id"],
+                agent_id=actor["agent_id"], turn_key=result["resume_turn_key"]),
+            "generation_present": folder is None or any((folder / name).exists() for name in
+                ("editorial-response.json", "private-generation.json", "private-delivery.json")),
+            "observed_at": time.time()}}
+
+    def report_quota_request(self, original: Mapping[str, Any]) -> dict[str, Any] | None:
+        """Observe quota recovery, never resume or rewrite the failed native Turn."""
+        result = (original.get("response") or {}).get("native_loopx") or {}
+        request = original.get("loopx_request") or {}
+        if (original.get("status") != "completed" or request.get("work_kind") != "report"
+                or request.get("report_quota_recovery") is not None
+                or result.get("result_kind") != "host_failure" or not original.get("upstream_turn_id")):
+            return None
+        from .codex_adapter import CodexAppServerAdapter
+        from .codex_context import checked_session_home
+        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
+        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"])
+        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
+        return {**request, "turn_instance_id": request["turn_instance_id"] + ".quota-recovery",
+            "report_quota_recovery": {"original_turn_id": original["turn_id"], "native_observation": proof}}
+
     def report_selection_recovery(self, original: dict[str, Any], *,
                                   probe_turn: dict[str, Any] | None = None) -> dict[str, Any] | None:
         """The typed owner qualifies this observation before any new worker."""
@@ -260,8 +321,7 @@ def recover_report_settlement(self, actor: dict[str, Any], window: dict[str, Any
         from ..periodic_report.private_generation import attempt_directory, validate_private_generation
         from .steward_reports import report_service
         with self.controller.lock, self._report_fence({"actor": actor}):
-            original = self.store.turn_for_client(actor["session_id"],
-                native_steward_client_id(actor, "report." + window["window_id"]))
+            original = report_turn(self.store, actor, window)
             if not original:
                 return
             request = original.get("loopx_request") or {}
@@ -340,6 +400,16 @@ def submit(self, binding_id: str, *, turn_instance_id: str, todo_id: str | None
                 frozen.setdefault("work_kind", "native_turn")
                 if frozen != request or work_kind == "monitor" and existing["loopx_request"].get("monitor_deadline") != monitor_deadline:
                     raise ValueError("native steward request identity conflict")
+                if work_kind == "report":
+                    retained = report_turn(self.store, actor, report_window)
+                    if retained is not None and retained["turn_id"] != existing["turn_id"]:
+                        return retained, False
+                    quota_request = self.report_quota_request(existing)
+                    if quota_request is not None:
+                        request = quota_request
+                        client_id = native_steward_client_id(actor, request["turn_instance_id"])
+                        existing = None
+            if existing is not None:
                 monitor_retry = work_kind == "monitor" and self._monitor_selection_failed_before_poll(existing)
                 if not monitor_retry and work_kind != "report":
                     return existing, False

loopx/chat_runtime.py

@@ -812,6 +812,11 @@ def _ensure_adapter_locked(
         if self.project_contexts.conversation_bindings is not None:
             from .capabilities.native_chat.external_conversations import ChatExternalConversations
             recovery = ChatExternalConversations(self).queued_recovery_context(session, active, accepted_turn_id)
+        native_quota_refresh = bool(active and active_id == accepted_turn_id
+            and active.get("status") == "queued" and active.get("origin") == "native_loopx"
+            and active.get("loopx_execution") is True
+            and (active.get("loopx_request") or {}).get("work_kind") == "report"
+            and isinstance((active.get("loopx_request") or {}).get("report_quota_recovery"), dict))
         with self.lock:
             current = self.adapters.get(session_id)
             manager_profile_changed = bool(
@@ -830,6 +835,7 @@ def _ensure_adapter_locked(
                 and current.healthcheck()
                 and not manager_profile_changed
                 and not recovery.get("quota_attempt")
+                and not native_quota_refresh
                 and (isinstance(current, CodexAppServerAdapter) and current.session.read_only_review) == bool(recovery.get("read_only_review"))
                 and model_allocation is None
                 and not settings_changed