fix(agent): make cost_status sticky by priority (#67764)

agent/codex_runtime.py

@@ -89,7 +89,7 @@ def _record_codex_app_server_usage(agent, turn) -> dict[str, Any]:
                 )
         return {}
 
-    from agent.usage_pricing import CanonicalUsage, estimate_usage_cost
+    from agent.usage_pricing import CanonicalUsage, estimate_usage_cost, sticky_cost_status
 
     input_tokens = _coerce_usage_int(usage.get("inputTokens"))
     cache_read_tokens = _coerce_usage_int(usage.get("cachedInputTokens"))
@@ -147,7 +147,9 @@ def _record_codex_app_server_usage(agent, turn) -> dict[str, Any]:
     )
     if cost_result.amount_usd is not None:
         agent.session_estimated_cost_usd += float(cost_result.amount_usd)
-    agent.session_cost_status = cost_result.status
+    agent.session_cost_status = sticky_cost_status(
+        agent.session_cost_status, cost_result.status
+    )
     agent.session_cost_source = cost_result.source
 
     if agent._session_db and agent.session_id:

agent/conversation_loop.py

@@ -68,7 +68,7 @@
     zai_coding_overload_retry_ceiling,
 )
 from agent.trajectory import has_incomplete_scratchpad
-from agent.usage_pricing import estimate_usage_cost, normalize_usage
+from agent.usage_pricing import estimate_usage_cost, normalize_usage, sticky_cost_status
 from hermes_constants import PARTIAL_STREAM_STUB_ID
 from hermes_logging import set_session_context
 from tools.skill_provenance import set_current_write_origin
@@ -2318,7 +2318,9 @@ def _perform_api_call(next_api_kwargs):
                             agent.session_estimated_cost_usd += float(_moa_ref_cost)
                         except (TypeError, ValueError):  # pragma: no cover - defensive
                             pass
-                    agent.session_cost_status = cost_result.status
+                    agent.session_cost_status = sticky_cost_status(
+                        agent.session_cost_status, cost_result.status
+                    )
                     agent.session_cost_source = cost_result.source
 
                     # Persist token counts to session DB for /insights.

agent/insights.py

@@ -28,6 +28,7 @@
     estimate_usage_cost,
     format_duration_compact,
     has_known_pricing,
+    sticky_cost_status,
 )
 
 
@@ -580,7 +581,7 @@ def _accumulate(model, provider, base_url, session_id, inp, out,
                 status = cost_status or "unknown"
             d["cost"] += estimate
             d["actual_cost"] += float(actual_cost or 0.0)
-            d["cost_status"] = status
+            d["cost_status"] = sticky_cost_status(d.get("cost_status"), status)
             if has_known_pricing(model, provider or None, base_url):
                 d["has_pricing"] = True
             else:

agent/usage_pricing.py

@@ -26,6 +26,32 @@
     "none",
 ]
 
+# Priority order for sticky cost_status: once a session reaches a higher
+# accuracy tier, it should not silently downgrade on a single provider
+# hiccup.  "actual" > "included" > "estimated" > "unknown".
+_COST_STATUS_PRIORITY: dict[str, int] = {
+    "actual": 4,
+    "included": 3,
+    "estimated": 2,
+    "unknown": 1,
+}
+
+
+def sticky_cost_status(current: Optional[str], incoming: Optional[str]) -> Optional[str]:
+    """Return the higher-priority cost status between *current* and *incoming*.
+
+    If *incoming* is ``None`` the current value is kept.  If *current* is
+    ``None`` the incoming value is used.  Otherwise the one with the higher
+    priority wins.  Equal priority keeps the incoming value (latest wins).
+    """
+    if incoming is None:
+        return current
+    if current is None:
+        return incoming
+    cur_prio = _COST_STATUS_PRIORITY.get(current, 0)
+    inc_prio = _COST_STATUS_PRIORITY.get(incoming, 0)
+    return incoming if inc_prio >= cur_prio else current
+
 
 @dataclass(frozen=True)
 class CanonicalUsage: