feat: add opt-in estimated API cost reporting

docs/docs/usage-guide/additional_configurations.md

@@ -30,6 +30,14 @@ To see which model actually answered, how many tokens the run consumed, and how
 /review --config.output_run_details=true
 ```
 
+API-cost collection is a separate, default-off option controlled by `config.output_run_cost`. Enable both flags to collect it and add it inside the run-details section:
+
+```
+/review --config.output_run_details=true --config.output_run_cost=true
+```
+
+`config.output_run_details` remains the public-output gate: setting only `config.output_run_cost=true` collects run-level cost data but never adds it to a PR comment.
+
 On providers that support GitHub-Flavored Markdown this appends a collapsible section to the generated comment; elsewhere `/review` and `/describe` append the same information as plain text:
 
 ```
@@ -38,10 +46,17 @@ On providers that support GitHub-Flavored Markdown this appends a collapsible se
 - Tokens: 12,340 in / 1,205 out / 13,545 total
 - Time cost: 8.2s
 - AI calls: 1
+- Estimated API cost: $0.08 USD
+  - anthropic/claude-opus-5: $0.07 USD
+  - anthropic/claude-sonnet-5: $0.01 USD
 ```
 
 `Model` shows the model that produced the answer, marked `(fallback)` when the primary model failed and a fallback took over. The `Tokens` line appears only when the model provider reports usage. `AI calls` counts the successful LLM invocations made during the run. The flag is disabled by default.
 
+`Estimated API cost` is derived synchronously from each completed LiteLLM response and its finalized usage. LiteLLM can account for cache reads, cache writes, reasoning tokens, and provider-specific usage categories when the response and its pricing data include them. Multi-model runs show a compact breakdown of the known costs. Exact `Decimal` values are retained for aggregation, while public currency output is rounded to two decimal places; a tiny positive value that would round to zero is shown as `<$0.01` instead of `$0.00`. If only some successful calls can be priced, the total is marked `partial` with the priced-call count; if none can be priced, the line reports `unavailable`. Missing pricing is never rendered as `$0`.
+
+The amount is an estimate based on LiteLLM's pricing data, not provider-invoice-authoritative billing. Reconcile it with the provider's billing records before using it for accounting or chargeback. Streaming responses are priced only after finalized usage is available; asynchronous callbacks and transient `response_cost` callback metadata are not treated as the sole source of truth. The public section contains only aggregate costs and configured model names, never prompts, response bodies, API keys, or provider request IDs.
+
 Notes:
 
 - `/improve` appends the section only when it publishes a summary comment. If the provider lacks GFM support or `pr_code_suggestions.commitable_code_suggestions` is enabled, `/improve` posts inline comments instead, so no run details section appears.

pr_agent/algo/ai_handlers/litellm_ai_handler.py

@@ -18,9 +18,9 @@
                            USER_MESSAGE_ONLY_MODELS)
 from pr_agent.algo.ai_handlers.base_ai_handler import BaseAiHandler
 from pr_agent.algo.ai_handlers.litellm_helpers import (
-    MockResponse, _get_azure_ad_token, _handle_streaming_response,
-    _process_litellm_extra_body)
-from pr_agent.algo.run_details import record_ai_call
+    _get_azure_ad_token, _handle_streaming_response,
+    _process_litellm_extra_body, _response_field)
+from pr_agent.algo.run_details import _as_decimal_cost, record_ai_call
 from pr_agent.algo.utils import ReasoningEffort, get_version
 from pr_agent.config_loader import get_settings
 from pr_agent.log import get_logger
@@ -377,12 +377,73 @@ def prepare_logs(self, response, system, user, resp, finish_reason):
         return response_log
 
     @staticmethod
-    def _record_completion_metadata(response) -> None:
-        """Count the call and accumulate token usage when the provider reports it.
+    def _record_completion_metadata(response, model=None, display_model=None) -> None:
+        """Count a successful call and synchronously collect usage-based cost when possible."""
+        usage = _response_field(response, "usage")
+
+        cost_usd = None
+        if get_settings().get("config.output_run_cost", False):
+            # The guard covers the whole cost block, not just completion_cost:
+            # reading inline costs and probing usage call model_dump() on
+            # provider-specific objects, and a cost estimate must never fail a
+            # call that already succeeded and was billed.
+            try:
+                cost_usd = LiteLLMAIHandler._read_positive_response_cost(response, usage)
+                if cost_usd is None and model and LiteLLMAIHandler._has_priceable_usage(usage):
+                    # Preserve LiteLLM's full usage object so completion_cost can price cache,
+                    # reasoning, and provider-specific categories. Convert the small completed
+                    # stream wrapper to a dictionary while retaining `response.usage`.
+                    cost_response = response
+                    if not isinstance(response, dict) and not hasattr(response, "model_dump"):
+                        cost_response = response.dict()
+                    cost_usd = litellm.completion_cost(completion_response=cost_response, model=model)
+            except Exception as e:
+                # Treat missing model pricing or insufficient usage as an unavailable call cost.
+                # Retain the successful call so the collector marks the aggregate safely.
+                get_logger().debug(f"Unable to estimate API cost for model {model}: {type(e).__name__}")
+
+        recorded_model = display_model if display_model is not None else model
+        record_ai_call(usage, model=recorded_model, cost_usd=cost_usd)
 
-        Streaming models return a MockResponse without `usage`, so tokens stay unset.
+    @staticmethod
+    def _read_positive_response_cost(response, usage):
+        """Read a finalized inline cost, rejecting zero placeholders and invalid values."""
+        candidates = [
+            _response_field(usage, "response_cost"),
+            _response_field(usage, "cost"),
+        ]
+
+        hidden_params = _response_field(response, "_hidden_params")
+        if hasattr(hidden_params, "model_dump"):
+            hidden_params = hidden_params.model_dump()
+        if isinstance(hidden_params, dict):
+            candidates.append(hidden_params.get("response_cost"))
+
+        for candidate in candidates:
+            decimal_cost = _as_decimal_cost(candidate)
+            if decimal_cost is not None:
+                return decimal_cost
+        return None
+
+    @staticmethod
+    def _has_priceable_usage(usage) -> bool:
+        """Return true when finalized usage reports a positive token count.
+
+        Only token counters gate pricing: provider extras such as Groq's timing
+        floats (queue_time, prompt_time) are not billable quantities, and letting
+        them pass would send zero-token usage to completion_cost, which prices
+        it as 0.0 instead of raising.
         """
-        record_ai_call(getattr(response, "usage", None))
+        if usage is None:
+            return False
+        return any(
+            isinstance(count, int) and not isinstance(count, bool) and count > 0
+            for count in (
+                _response_field(usage, "prompt_tokens"),
+                _response_field(usage, "completion_tokens"),
+                _response_field(usage, "total_tokens"),
+            )
+        )
 
     def _configure_claude_extended_thinking(self, model: str, kwargs: dict) -> dict:
         """
@@ -913,19 +974,23 @@ def _as_int(value):
                     body=None,
                 ) from e
 
-            get_logger().debug(f"\nAI response:\n{resp}")
+        # Post-response bookkeeping happens outside the Bedrock IMDS lock above: it
+        # touches no os.environ credentials, and in IMDS mode the lock serializes
+        # every concurrent call, so holding it through logging and cost pricing
+        # would make each waiting coroutine pay for them serially.
+        get_logger().debug(f"\nAI response:\n{resp}")
 
-            # log the full response for debugging
-            response_log = self.prepare_logs(response_obj, system, user, resp, finish_reason)
-            get_logger().debug("Full_response", artifact=response_log)
+        # log the full response for debugging
+        response_log = self.prepare_logs(response_obj, system, user, resp, finish_reason)
+        get_logger().debug("Full_response", artifact=response_log)
 
-            # for CLI debugging
-            if get_settings().config.verbosity_level >= 2:
-                get_logger().info(f"\nAI response:\n{resp}")
+        # for CLI debugging
+        if get_settings().config.verbosity_level >= 2:
+            get_logger().info(f"\nAI response:\n{resp}")
 
-            self._record_completion_metadata(response_obj)
+        self._record_completion_metadata(response_obj, model=model, display_model=user_model)
 
-            return resp, finish_reason
+        return resp, finish_reason
 
     async def _get_completion(self, **kwargs):
         """
@@ -947,6 +1012,7 @@ async def _get_completion(self, **kwargs):
         # response normalization. Streaming avoids that conversion path.
         if model in self.streaming_required_models or force_streaming:
             kwargs["stream"] = True
+            kwargs["stream_options"] = {"include_usage": True}
             if force_streaming and model not in self.streaming_required_models:
                 get_logger().info(
                     f"Using streaming mode for model {model} "
@@ -955,10 +1021,7 @@ async def _get_completion(self, **kwargs):
             else:
                 get_logger().info(f"Using streaming mode for model {model}")
             response = await acompletion(**kwargs)
-            resp, finish_reason = await _handle_streaming_response(response)
-            # Create MockResponse for streaming since we don't have the full response object
-            mock_response = MockResponse(resp, finish_reason)
-            return resp, finish_reason, mock_response
+            return await _handle_streaming_response(response, model=model)
         else:
             response = await acompletion(**kwargs)
             if response is None or len(response["choices"]) == 0:

pr_agent/algo/ai_handlers/litellm_helpers.py

@@ -43,21 +43,42 @@
 }
 
 
-async def _handle_streaming_response(response):
+def _response_field(response, name):
+    if isinstance(response, dict):
+        return response.get(name)
+    return getattr(response, name, None)
+
+
+def _stream_usage(chunk):
+    """Read finalized usage from a regular or metadata-only LiteLLM chunk."""
+    usage = _response_field(chunk, "usage")
+    if usage is not None:
+        return usage
+    hidden_params = _response_field(chunk, "_hidden_params")
+    if isinstance(hidden_params, dict):
+        return hidden_params.get("usage")
+    return None
+
+
+async def _handle_streaming_response(response, model=None):
     """
     Handle streaming response from acompletion and collect the full response.
 
     Args:
         response: The streaming response object from acompletion
 
     Returns:
-        tuple: (full_response_content, finish_reason)
+        tuple: (full_response_content, finish_reason, completed_response)
     """
     full_response = ""
     finish_reason = None
+    finalized_usage = None
 
     try:
         async for chunk in response:
+            usage = _stream_usage(chunk)
+            if usage is not None:
+                finalized_usage = usage
             if chunk.choices and len(chunk.choices) > 0:
                 choice = chunk.choices[0]
                 delta = choice.delta
@@ -76,13 +97,14 @@ async def _handle_streaming_response(response):
     elif not full_response and finish_reason:
         get_logger().debug(f"Streaming response resulted in empty content but completed with finish_reason: {finish_reason}")
         raise openai.APIError(f"Streaming response completed with finish_reason '{finish_reason}' but no content received")
-    return full_response, finish_reason
+    return full_response, finish_reason, MockResponse(full_response, finish_reason, finalized_usage, model)
 
 
 class MockResponse:
-    """Mock response object for streaming models to enable consistent logging."""
+    """Represent a completed streaming response while retaining LiteLLM's finalized usage object."""
 
-    def __init__(self, resp, finish_reason):
+    def __init__(self, resp, finish_reason, usage=None, model=None):
+        self.usage = usage
         self._data = {
             "choices": [
                 {
@@ -91,9 +113,19 @@ def __init__(self, resp, finish_reason):
                 }
             ]
         }
+        if model is not None:
+            self._data["model"] = model
 
     def dict(self):
-        return self._data
+        data = self._data.copy()
+        if self.usage is not None:
+            if hasattr(self.usage, "model_dump"):
+                data["usage"] = self.usage.model_dump()
+            elif isinstance(self.usage, dict):
+                data["usage"] = self.usage.copy()
+            else:
+                data["usage"] = vars(self.usage).copy()
+        return data
 
 
 def _get_azure_ad_token():

pr_agent/algo/ai_handlers/openai_ai_handler.py

@@ -61,6 +61,10 @@ async def chat_completion(self, model: str, system: str, user: str, temperature:
             usage = chat_completion.usage
             get_logger().info("AI response", response=resp, messages=messages, finish_reason=finish_reason,
                               model=model, usage=usage)
+            # Count the call and its tokens but no cost: this handler has no pricing
+            # source wired up, so with output_run_cost enabled its calls render as
+            # unpriced. Left as-is while the path stays cold — no setting selects
+            # this handler, it is reachable only by injecting it programmatically.
             record_ai_call(usage)
             return resp, finish_reason
         except openai.RateLimitError as e:

pr_agent/algo/run_details.py

@@ -10,6 +10,7 @@
 import time
 from contextvars import ContextVar
 from dataclasses import dataclass, field
+from decimal import Decimal, InvalidOperation
 from typing import Optional
 
 _run_details: ContextVar[Optional["RunDetails"]] = ContextVar(
@@ -44,6 +45,13 @@ class RunDetails:
     total_tokens: int = 0
     # Successful LLM invocations, counted even when their token usage is unavailable.
     num_ai_calls: int = 0
+    # Accumulate costs only when cost output is enabled and LiteLLM can synchronously
+    # price a successful response with a positive amount. Use the known-call count to
+    # distinguish priced calls from missing pricing data. Retain per-model totals to
+    # keep fallback and multi-call runs auditable.
+    total_cost_usd: Decimal = field(default_factory=lambda: Decimal("0"))
+    known_cost_call_count: int = 0
+    model_costs_usd: dict[str, Decimal] = field(default_factory=dict)
     # Monotonic reference taken when the collector is installed, i.e. at the top of the
     # tool's run(). Monotonic so that wall-clock adjustments cannot yield a negative duration.
     start_time: float = field(default_factory=time.monotonic)
@@ -60,6 +68,15 @@ def has_token_usage(self) -> bool:
             or self.completion_tokens > 0
         )
 
+    @property
+    def cost_status(self) -> str:
+        """Return whether every, some, or none of the successful calls were priced."""
+        if self.known_cost_call_count == 0:
+            return "unavailable"
+        if self.known_cost_call_count == self.num_ai_calls:
+            return "complete"
+        return "partial"
+
 
 def init_run_details() -> RunDetails:
     """Install a fresh collector for the current run and return it."""
@@ -107,11 +124,36 @@ def add_token_usage(usage) -> None:
     details.total_tokens += total_tokens
 
 
-def record_ai_call(usage=None) -> None:
-    """Count one AI call and accumulate token usage when available."""
+def _as_decimal_cost(cost_usd) -> Optional[Decimal]:
+    """Normalize a positive finite USD value without introducing float math.
+
+    Zero is rejected on purpose: litellm.completion_cost returns 0.0 both for
+    zero-priced model entries (e.g. local/ollama models) and for usage without
+    billable tokens, so a zero here means "could not be priced", not "free" —
+    recording it would render a false "$0.00" with cost status complete.
+    """
+    if cost_usd is None or isinstance(cost_usd, bool):
+        return None
+    try:
+        cost = cost_usd if isinstance(cost_usd, Decimal) else Decimal(str(cost_usd))
+    except (InvalidOperation, TypeError, ValueError):
+        return None
+    if not cost.is_finite() or cost <= 0:
+        return None
+    return cost
+
+
+def record_ai_call(usage=None, model: Optional[str] = None, cost_usd=None) -> None:
+    """Count one successful AI call and accumulate usage and known cost."""
     details = get_run_details()
     if details is None:
         return
     details.num_ai_calls += 1
     if usage is not None:
         add_token_usage(usage)
+    cost = _as_decimal_cost(cost_usd)
+    if cost is not None:
+        details.total_cost_usd += cost
+        details.known_cost_call_count += 1
+        model_name = model or "unknown"
+        details.model_costs_usd[model_name] = details.model_costs_usd.get(model_name, Decimal("0")) + cost

pr_agent/algo/utils.py

@@ -13,6 +13,7 @@
 import time
 import traceback
 from datetime import datetime
+from decimal import ROUND_HALF_UP, Decimal
 from enum import Enum
 from importlib.metadata import PackageNotFoundError, version
 from typing import Any, Iterable, List, Tuple, TypedDict
@@ -1486,6 +1487,14 @@ def show_relevant_configurations(relevant_section: str) -> str:
     return markdown_text
 
 
+def _format_usd(cost: Decimal) -> str:
+    """Format cost at two decimals without turning a positive amount into false zero."""
+    rounded = cost.quantize(Decimal("0.01"), rounding=ROUND_HALF_UP)
+    if cost > 0 and rounded == 0:
+        return "<$0.01"
+    return f"${rounded:.2f}"
+
+
 def show_run_details(gfm_supported: bool) -> str:
     """Render the opt-in run-details section (model, tokens, time cost, AI calls).
 
@@ -1508,6 +1517,21 @@ def show_run_details(gfm_supported: bool) -> str:
     lines.append(f"- Time cost: {details.duration_seconds:.1f}s")
     if details.num_ai_calls:
         lines.append(f"- AI calls: {details.num_ai_calls}")
+    if get_settings().get("config.output_run_cost", False) and details.num_ai_calls:
+        if details.cost_status == "unavailable":
+            # No causal claim here: pricing can be missing because usage was absent
+            # (streaming without a final usage chunk) or because the active handler
+            # never collects costs (openai/langchain handlers).
+            lines.append("- Estimated API cost: unavailable (no calls could be priced)")
+        else:
+            partial = ""
+            if details.cost_status == "partial":
+                partial = (f" (partial: {details.known_cost_call_count} of "
+                           f"{details.num_ai_calls} successful calls priced)")
+            lines.append(f"- Estimated API cost: {_format_usd(details.total_cost_usd)} USD{partial}")
+            if len(details.model_costs_usd) > 1:
+                for model, cost in details.model_costs_usd.items():
+                    lines.append(f"  - {model}: {_format_usd(cost)} USD")
     body = "\n".join(lines)
 
     if gfm_supported: