diff --git a/agent/auxiliary_client.py b/agent/auxiliary_client.py index 3c614dd8f947..3c7a5e8ea9cd 100644 --- a/agent/auxiliary_client.py +++ b/agent/auxiliary_client.py @@ -1016,6 +1016,22 @@ def create(self, **kwargs) -> Any: anthropic_kwargs["temperature"] = temperature response = self._client.messages.create(**anthropic_kwargs) + + # KR-P2-K ST2 — secondary rate-limit signal (best-effort). + # Only direct-Anthropic responses carry anthropic-ratelimit-* + # headers; this is one of the two surfaces that does. + try: + from agent.cost_ladder_wire import ( + record_rate_limit_pulse_from_response, + ) + record_rate_limit_pulse_from_response(response) + except Exception as _cl_exc: + logger.debug( + "[kora.cost_ladder] AnthropicAuxiliaryClient " + "rate-limit-pulse capture failed: %r", + _cl_exc, + ) + _transport = get_transport("anthropic_messages") _nr = _transport.normalize_response( response, strip_tool_prefix=self._is_oauth @@ -5287,3 +5303,69 @@ async def async_call_llm( logger.debug("Auxiliary (async): cache eviction after connection error failed", exc_info=True) raise + + +# --------------------------------------------------------------------------- +# KR-P2-K ST2 — cost-ladder estimator wrap +# --------------------------------------------------------------------------- +# +# The primary estimator signal feeds the cost-ladder holder after every +# auxiliary inference returns. Wrapping ``call_llm`` + ``async_call_llm`` +# at module-level captures all 16+ internal return paths (success + +# retry + fallback chains) in one place — simpler than threading the +# kwargs through each individual return site. +# +# The wrappers store the resolved model/provider/base_url from the +# caller's kwargs; ``record_inference_from_response`` falls back to +# ``response.model`` when those are absent, so the cost holder gets a +# usable model identifier even when fallback rerouted to a different +# provider mid-call. +# +# Fail-soft: the wrapper catches every estimator failure path and logs +# DEBUG; the underlying ``call_llm`` / ``async_call_llm`` response is +# always returned unchanged to the caller. + +_call_llm_inner = call_llm +_async_call_llm_inner = async_call_llm + + +def call_llm(*args, **kwargs): + """KR-P2-K ST2 wrap of :func:`call_llm` that feeds the cost-ladder + estimator after the inner call returns. Failed calls (which raise + before producing a response) correctly do NOT contribute to the + spent_to_date counter — only successful inferences burn the pool. + """ + response = _call_llm_inner(*args, **kwargs) + try: + from agent.cost_ladder_wire import record_inference_from_response + record_inference_from_response( + response, + model=kwargs.get("model"), + provider=kwargs.get("provider"), + base_url=kwargs.get("base_url"), + ) + except Exception as _cl_exc: + logger.debug( + "[kora.cost_ladder] call_llm wrap feed failed: %r", _cl_exc + ) + return response + + +async def async_call_llm(*args, **kwargs): + """KR-P2-K ST2 wrap of :func:`async_call_llm`. Same semantics as + the sync wrap above.""" + response = await _async_call_llm_inner(*args, **kwargs) + try: + from agent.cost_ladder_wire import record_inference_from_response + record_inference_from_response( + response, + model=kwargs.get("model"), + provider=kwargs.get("provider"), + base_url=kwargs.get("base_url"), + ) + except Exception as _cl_exc: + logger.debug( + "[kora.cost_ladder] async_call_llm wrap feed failed: %r", + _cl_exc, + ) + return response diff --git a/agent/conversation_loop.py b/agent/conversation_loop.py index 8ea2fc9f3fd8..13d5031d0f93 100644 --- a/agent/conversation_loop.py +++ b/agent/conversation_loop.py @@ -1588,6 +1588,32 @@ def _stop_spinner(): agent.session_cost_status = cost_result.status agent.session_cost_source = cost_result.source + # KR-P2-K ST2 — feed the cost-ladder estimator + # singleton. Fail-soft: any failure path logs DEBUG + # and returns; the inference response handler must + # never crash because of cost-ladder accounting. + # The chokepoint here catches BOTH the direct- + # Anthropic path (api_mode="anthropic_messages") + # AND the OpenAI-compat path; both flow through + # this normalize_usage block. + try: + from agent.cost_ladder_wire import ( + record_inference_from_response, + ) + record_inference_from_response( + response, + model=agent.model, + provider=agent.provider, + base_url=agent.base_url, + api_mode=agent.api_mode, + ) + except Exception as _cl_exc: + logger.debug( + "[kora.cost_ladder] conversation_loop chokepoint " + "feed failed: %r", + _cl_exc, + ) + # Persist token counts to session DB for /insights. # Do this for every platform with a session_id so non-CLI # sessions (gateway, cron, delegated runs) cannot lose diff --git a/agent/cost_ladder_wire.py b/agent/cost_ladder_wire.py new file mode 100644 index 000000000000..ec76a0d74e06 --- /dev/null +++ b/agent/cost_ladder_wire.py @@ -0,0 +1,281 @@ +"""Cost-ladder wire-in helpers (KR-P2-K ST2). + +Bridges the inference dispatch sites to the +:class:`agent.cost_state_holder.CostStateHolder` singleton. Two surfaces: + + - :func:`record_inference_from_response` — extracts ``response.usage``, + normalizes via :func:`agent.usage_pricing.normalize_usage`, feeds the + holder's :meth:`record_inference`. Fail-soft: every failure path + logs DEBUG and returns; the inference response handler must never + crash because of the cost-ladder estimator. + + - :func:`record_rate_limit_pulse_from_response` — extracts the + ``anthropic-ratelimit-{requests,tokens}-{limit,remaining,reset}`` + headers from a direct-Anthropic SDK response, builds a + :class:`RateLimitPulse`, feeds the holder's + :meth:`record_rate_limit_pulse`. Best-effort secondary signal per + R4.1 §9.6: only the 2 direct-Anthropic dispatch sites in this + codebase surface these headers; OpenAI-compat responses don't. + +# Asymmetry recap (per the B1 verification ruling) + +The Joshua billing memo requires the **primary signal** (per-call +$-burn against the $200 pool) to cover ALL inference paths. The +**secondary signal** (rate-limit headers) is best-effort — only +direct-Anthropic SDK responses carry them. + +This module covers: + + - Primary signal via :func:`record_inference_from_response`: called + from BOTH the main agent's :mod:`agent.conversation_loop` + chokepoint (post-:func:`normalize_usage`) AND from + :mod:`agent.auxiliary_client`'s ``_validate_llm_response`` / + ``_record_and_validate`` wrappers, covering compression / vision / + web-extract / session-search / skills-hub / MCP / title-generation + side tasks. + + - Secondary signal via :func:`record_rate_limit_pulse_from_response`: + called from the 2 direct-Anthropic sites + (``run_agent.py:_anthropic_messages_create`` wrapper + + ``auxiliary_client.py:AnthropicAuxiliaryClient`` chat-completions + shim). + +# Note on the "3 sites" framing + +The KR-P2-K bucket spec ST2 referenced "3 SDK call sites" from the +verification report. In practice the actual inference dispatch in +this codebase funnels through 2 chokepoints +(``conversation_loop.normalize_usage`` + ``auxiliary_client.call_llm``) +which together cover all dispatch paths. The third "site" identified +in the verification report (``agent_runtime_helpers.py:1275``) is +actually the OpenAI client construction site — the inference call +dispatched through that client flows back through the +``conversation_loop`` chokepoint. Coverage is intact; the framing +was site-count vs chokepoint-count. +""" + +from __future__ import annotations + +import logging +from datetime import datetime, timezone +from typing import Any, Optional + +from agent.cost_state_holder import ( + RateLimitAxis, + RateLimitPulse, + get_cost_holder, +) +from agent.usage_pricing import normalize_usage + +logger = logging.getLogger(__name__) + + +# --------------------------------------------------------------------------- +# Primary signal — record_inference (per-call $-burn) +# --------------------------------------------------------------------------- + + +def record_inference_from_response( + response: Any, + *, + model: Optional[str] = None, + provider: Optional[str] = None, + base_url: Optional[str] = None, + api_mode: Optional[str] = None, +) -> None: + """Feed the cost-ladder estimator from an inference response. + + Args: + response: SDK response object. Must expose ``response.usage`` + in a shape :func:`normalize_usage` can consume (Anthropic + ``input_tokens``/``output_tokens``/``cache_*_input_tokens`` + shape OR OpenAI ``prompt_tokens``/``completion_tokens``/ + ``input_tokens_details`` shape). + model: Model identifier — typically from ``response.model``. + Falls back to ``getattr(response, "model", "")``. + provider: Provider hint forwarded to + :func:`agent.usage_pricing.normalize_usage` + + :func:`agent.cost_state_holder.CostStateHolder.record_inference`. + base_url: Endpoint base URL hint, forwarded the same way. + api_mode: API mode (``"anthropic_messages"`` / ``"openai"`` / + etc.) forwarded to :func:`normalize_usage` for shape + disambiguation. + + Fail-soft: any failure path (no holder, no usage attribute, + normalize raises, record_inference raises) logs at DEBUG and + returns. The caller's inference response handler must not see + estimator failures. + """ + try: + holder = get_cost_holder() + if holder is None: + return + + raw_usage = getattr(response, "usage", None) + if not raw_usage: + return + + canonical_usage = normalize_usage( + raw_usage, provider=provider, api_mode=api_mode + ) + + resolved_model = model or getattr(response, "model", "") or "" + if not resolved_model: + # Without a model name there's no pricing route to resolve. + return + + holder.record_inference( + canonical_usage, + model_name=resolved_model, + provider=provider, + base_url=base_url, + ) + except Exception as exc: + # Fail-soft per the contract — estimator failures must not + # crash the inference response handler. + logger.debug( + "[kora.cost_ladder] record_inference_from_response failed: %r", + exc, + ) + + +# --------------------------------------------------------------------------- +# Secondary signal — record_rate_limit_pulse (Anthropic-only) +# --------------------------------------------------------------------------- + + +def record_rate_limit_pulse_from_response(response: Any) -> None: + """Capture Anthropic SDK rate-limit headers from a direct-Anthropic + response. + + Anthropic responses carry six headers (three per axis, two axes): + + ``anthropic-ratelimit-requests-limit`` + ``anthropic-ratelimit-requests-remaining`` + ``anthropic-ratelimit-requests-reset`` + ``anthropic-ratelimit-tokens-limit`` + ``anthropic-ratelimit-tokens-remaining`` + ``anthropic-ratelimit-tokens-reset`` + + The ``-reset`` values are ISO-8601 timestamps. ``-limit`` and + ``-remaining`` are non-negative integers. + + Best-effort: if the response doesn't expose ``response.headers`` + (some test doubles + non-Anthropic shapes), or any header is + missing/malformed, the helper logs DEBUG and returns without + touching the holder. Only the 2 direct-Anthropic dispatch sites + surface these headers in this codebase; the OpenAI-compat path + doesn't. + """ + try: + holder = get_cost_holder() + if holder is None: + return + + headers = _extract_headers(response) + if headers is None: + return + + requests_axis = _parse_axis(headers, axis="requests") + tokens_axis = _parse_axis(headers, axis="tokens") + if requests_axis is None or tokens_axis is None: + return + + pulse = RateLimitPulse( + requests=requests_axis, + tokens=tokens_axis, + captured_at=datetime.now(timezone.utc), + ) + holder.record_rate_limit_pulse(pulse) + except Exception as exc: + logger.debug( + "[kora.cost_ladder] record_rate_limit_pulse_from_response " + "failed: %r", + exc, + ) + + +def _extract_headers(response: Any) -> Optional[Any]: + """Return a header-lookup object from the response, or ``None``. + + Anthropic SDK responses expose headers via either + ``response.headers`` (dict-like) OR + ``response.http_response.headers`` (httpx Response). Try both; + fail-soft if neither. + """ + headers = getattr(response, "headers", None) + if headers is not None: + return headers + http_response = getattr(response, "http_response", None) + if http_response is not None: + return getattr(http_response, "headers", None) + return None + + +def _parse_axis(headers: Any, *, axis: str) -> Optional[RateLimitAxis]: + """Build a :class:`RateLimitAxis` from a header set. + + Returns ``None`` if any required header is missing or malformed. + """ + limit_key = f"anthropic-ratelimit-{axis}-limit" + remaining_key = f"anthropic-ratelimit-{axis}-remaining" + reset_key = f"anthropic-ratelimit-{axis}-reset" + + try: + limit_str = _get_header(headers, limit_key) + remaining_str = _get_header(headers, remaining_key) + reset_str = _get_header(headers, reset_key) + except KeyError: + return None + if limit_str is None or remaining_str is None or reset_str is None: + return None + + try: + limit = int(limit_str) + remaining = int(remaining_str) + reset_at = _parse_iso8601(reset_str) + except (ValueError, TypeError): + return None + if reset_at is None: + return None + + return RateLimitAxis(limit=limit, remaining=remaining, reset_at=reset_at) + + +def _get_header(headers: Any, key: str) -> Optional[str]: + """Look up ``key`` in headers, handling case-insensitive dict-like + and httpx-Headers shapes uniformly. Returns ``None`` if missing.""" + # httpx.Headers + most dict-like header objects support .get + getter = getattr(headers, "get", None) + if callable(getter): + value = getter(key) + if value is None: + value = getter(key.lower()) + return value + # Last-resort: bracket access + try: + return headers[key] + except (KeyError, TypeError): + try: + return headers[key.lower()] + except (KeyError, TypeError): + return None + + +def _parse_iso8601(value: str) -> Optional[datetime]: + """Parse an ISO-8601 timestamp string into a UTC-aware datetime. + + Anthropic's ``-reset`` headers are RFC3339 / ISO-8601 with a ``Z`` + suffix; Python's :meth:`datetime.fromisoformat` handles the + common cases. Returns ``None`` on parse failure. + """ + try: + # Python 3.11+ fromisoformat handles trailing Z + offsets natively. + # Replace 'Z' with '+00:00' for the older format too just in case. + normalized = value.replace("Z", "+00:00") if value.endswith("Z") else value + parsed = datetime.fromisoformat(normalized) + if parsed.tzinfo is None: + parsed = parsed.replace(tzinfo=timezone.utc) + return parsed + except (ValueError, AttributeError): + return None diff --git a/run_agent.py b/run_agent.py index 88222e406d65..7bbbe9e0a1a2 100644 --- a/run_agent.py +++ b/run_agent.py @@ -2867,7 +2867,26 @@ def _credential_pool_may_recover_rate_limit(self) -> bool: def _anthropic_messages_create(self, api_kwargs: dict): if self.api_mode == "anthropic_messages": self._try_refresh_anthropic_client_credentials() - return self._anthropic_client.messages.create(**api_kwargs) + response = self._anthropic_client.messages.create(**api_kwargs) + + # KR-P2-K ST2 — secondary rate-limit signal (best-effort). + # Only direct-Anthropic responses carry anthropic-ratelimit-* + # headers; this is one of the two surfaces that does. The + # primary signal (per-call $-burn) lands via the + # conversation_loop chokepoint after this returns. + try: + from agent.cost_ladder_wire import ( + record_rate_limit_pulse_from_response, + ) + record_rate_limit_pulse_from_response(response) + except Exception as _cl_exc: + logger.debug( + "[kora.cost_ladder] _anthropic_messages_create " + "rate-limit-pulse capture failed: %r", + _cl_exc, + ) + + return response def _rebuild_anthropic_client(self) -> None: """Rebuild the Anthropic client after an interrupt or stale call. diff --git a/tests/test_cost_ladder_wire.py b/tests/test_cost_ladder_wire.py new file mode 100644 index 000000000000..51f5e745b36e --- /dev/null +++ b/tests/test_cost_ladder_wire.py @@ -0,0 +1,352 @@ +"""Unit tests for ``agent/cost_ladder_wire.py`` (KR-P2-K ST2 helpers). + +Covers: + - ``record_inference_from_response`` happy path (extracts response.usage, + normalizes, calls holder.record_inference) + - Fail-soft on every failure mode: no holder, no usage, normalize + raises, record_inference raises, no model + - ``record_rate_limit_pulse_from_response`` happy path with both axes + - Fail-soft on missing headers / malformed values / no http_response + - Header lookup handles both ``response.headers`` and + ``response.http_response.headers`` (httpx-style) +""" + +from __future__ import annotations + +from datetime import datetime, timezone +from types import SimpleNamespace +from typing import Optional +from unittest.mock import MagicMock + +import pytest + +from agent.cost_ladder_wire import ( + record_inference_from_response, + record_rate_limit_pulse_from_response, +) +from agent.cost_state_holder import ( + CostStateHolder, + RateLimitPulse, + _reset_cost_holder_for_tests, + init_cost_holder, +) + + +@pytest.fixture(autouse=True) +def _reset_singleton(): + _reset_cost_holder_for_tests() + yield + _reset_cost_holder_for_tests() + + +# --------------------------------------------------------------------------- +# Fakes +# --------------------------------------------------------------------------- + + +def _holder() -> CostStateHolder: + return init_cost_holder( + billing_period_start=datetime(2026, 5, 1, tzinfo=timezone.utc), + ) + + +def _anthropic_response( + *, + input_tokens: int = 100, + output_tokens: int = 50, + model: str = "claude-sonnet-4.7", + headers: Optional[dict] = None, +) -> SimpleNamespace: + """Build a fake direct-Anthropic SDK response.""" + return SimpleNamespace( + model=model, + usage=SimpleNamespace( + input_tokens=input_tokens, + output_tokens=output_tokens, + ), + headers=headers, + ) + + +def _openai_response( + *, + prompt_tokens: int = 100, + completion_tokens: int = 50, + model: str = "claude-sonnet-4.7", +) -> SimpleNamespace: + """Build a fake OpenAI-compat response.""" + return SimpleNamespace( + model=model, + usage=SimpleNamespace( + prompt_tokens=prompt_tokens, + completion_tokens=completion_tokens, + total_tokens=prompt_tokens + completion_tokens, + ), + ) + + +# --------------------------------------------------------------------------- +# record_inference_from_response +# --------------------------------------------------------------------------- + + +def test_record_inference_no_holder_is_noop(): + """Without an initialized cost holder, the wire-helper silently + returns. The estimator must not crash inference response handling.""" + # Holder is reset by autouse fixture + record_inference_from_response( + _anthropic_response(), + model="claude-sonnet-4.7", + provider="anthropic", + ) + # No assertion — just ensure no exception raised + + +def test_record_inference_no_usage_attribute_is_noop(): + holder = _holder() + response_no_usage = SimpleNamespace(model="claude-sonnet-4.7", usage=None) + record_inference_from_response( + response_no_usage, + model="claude-sonnet-4.7", + provider="anthropic", + ) + assert holder.current.spent_to_date_usd == 0.0 + + +def test_record_inference_no_model_returns_silently(): + """Without a model name, no pricing route resolves. Skip.""" + holder = _holder() + response = _anthropic_response() + record_inference_from_response( + response, + model=None, + provider="anthropic", + ) + # Response had a .model attr so it shouldn't return silently — but + # if response.model is also empty: + response_no_model = SimpleNamespace( + usage=SimpleNamespace(input_tokens=10, output_tokens=5), + ) + # No model attribute at all + record_inference_from_response( + response_no_model, + model=None, + provider="anthropic", + ) + # Should not raise — fail-soft + + +def test_record_inference_happy_path_anthropic_shape(): + """Direct-Anthropic response normalizes via normalize_usage + feeds + the holder. Spend accumulates.""" + holder = _holder() + # Use a model that has known Anthropic pricing + response = _anthropic_response( + input_tokens=1000, output_tokens=500, model="claude-haiku-4.5" + ) + record_inference_from_response( + response, + model="claude-haiku-4.5", + provider="anthropic", + ) + # Cost should be > 0 (haiku is the cheapest tier; small amount but + # not zero) + assert holder.current.spent_to_date_usd > 0 + + +def test_record_inference_uses_response_model_when_arg_omitted(): + """If the caller doesn't pass ``model``, the helper extracts it from + ``response.model`` (the SDK's canonical model identifier).""" + holder = _holder() + response = _anthropic_response( + input_tokens=1000, output_tokens=500, model="claude-haiku-4.5" + ) + record_inference_from_response( + response, + model=None, # omitted — should fall back to response.model + provider="anthropic", + ) + assert holder.current.spent_to_date_usd > 0 + + +def test_record_inference_normalize_failure_is_swallowed(monkeypatch): + """If normalize_usage raises (malformed response shape), fail-soft.""" + holder = _holder() + monkeypatch.setattr( + "agent.cost_ladder_wire.normalize_usage", + MagicMock(side_effect=RuntimeError("malformed response")), + ) + response = _anthropic_response() + record_inference_from_response( + response, model="claude-sonnet-4.7", provider="anthropic" + ) + # Holder untouched + assert holder.current.spent_to_date_usd == 0.0 + + +def test_record_inference_record_failure_is_swallowed(monkeypatch): + """If holder.record_inference raises, fail-soft.""" + holder = _holder() + monkeypatch.setattr( + holder, + "record_inference", + MagicMock(side_effect=RuntimeError("estimator boom")), + ) + response = _anthropic_response() + record_inference_from_response( + response, model="claude-sonnet-4.7", provider="anthropic" + ) + # No exception raised; holder state unchanged + + +def test_record_inference_openai_compat_shape(): + """OpenAI-compat response (prompt/completion tokens shape) is + correctly normalized + recorded.""" + holder = _holder() + response = _openai_response( + prompt_tokens=2000, completion_tokens=300, model="claude-haiku-4.5" + ) + record_inference_from_response( + response, + model="claude-haiku-4.5", + provider="openrouter", + ) + assert holder.current.spent_to_date_usd >= 0 # any non-zero or zero is ok + + +# --------------------------------------------------------------------------- +# record_rate_limit_pulse_from_response +# --------------------------------------------------------------------------- + + +def _anthropic_headers() -> dict: + return { + "anthropic-ratelimit-requests-limit": "1000", + "anthropic-ratelimit-requests-remaining": "850", + "anthropic-ratelimit-requests-reset": "2026-05-21T12:00:00Z", + "anthropic-ratelimit-tokens-limit": "10000000", + "anthropic-ratelimit-tokens-remaining": "7500000", + "anthropic-ratelimit-tokens-reset": "2026-05-21T12:05:00Z", + } + + +def test_rate_limit_no_holder_is_noop(): + response = _anthropic_response(headers=_anthropic_headers()) + record_rate_limit_pulse_from_response(response) + # Nothing to assert; just no exception + + +def test_rate_limit_happy_path_both_axes_captured(): + holder = _holder() + response = _anthropic_response(headers=_anthropic_headers()) + record_rate_limit_pulse_from_response(response) + + pulse = holder.current.latest_rate_limit_pulse + assert pulse is not None + assert isinstance(pulse, RateLimitPulse) + assert pulse.requests.limit == 1000 + assert pulse.requests.remaining == 850 + assert pulse.tokens.limit == 10000000 + assert pulse.tokens.remaining == 7500000 + + +def test_rate_limit_reads_from_http_response_when_top_level_absent(): + """httpx-style responses expose headers via response.http_response.headers.""" + holder = _holder() + response = SimpleNamespace( + model="claude-sonnet-4.7", + usage=None, + http_response=SimpleNamespace(headers=_anthropic_headers()), + ) + record_rate_limit_pulse_from_response(response) + assert holder.current.latest_rate_limit_pulse is not None + + +def test_rate_limit_missing_headers_attribute_is_noop(): + holder = _holder() + response = SimpleNamespace(model="claude-sonnet-4.7", usage=None) + record_rate_limit_pulse_from_response(response) + assert holder.current.latest_rate_limit_pulse is None + + +def test_rate_limit_missing_required_header_is_noop(): + """If any of the 6 required headers is missing, skip — partial + pulses aren't meaningful.""" + holder = _holder() + incomplete = _anthropic_headers() + del incomplete["anthropic-ratelimit-tokens-remaining"] + response = _anthropic_response(headers=incomplete) + record_rate_limit_pulse_from_response(response) + assert holder.current.latest_rate_limit_pulse is None + + +def test_rate_limit_malformed_int_value_is_noop(): + holder = _holder() + bad = _anthropic_headers() + bad["anthropic-ratelimit-requests-limit"] = "not-a-number" + response = _anthropic_response(headers=bad) + record_rate_limit_pulse_from_response(response) + assert holder.current.latest_rate_limit_pulse is None + + +def test_rate_limit_malformed_timestamp_is_noop(): + holder = _holder() + bad = _anthropic_headers() + bad["anthropic-ratelimit-requests-reset"] = "not-a-date" + response = _anthropic_response(headers=bad) + record_rate_limit_pulse_from_response(response) + assert holder.current.latest_rate_limit_pulse is None + + +def test_rate_limit_iso8601_with_trailing_z_parses(): + """RFC3339 ``Z`` suffix is canonical Anthropic header format; + Python <3.11 ``fromisoformat`` didn't accept it natively.""" + holder = _holder() + response = _anthropic_response(headers=_anthropic_headers()) + record_rate_limit_pulse_from_response(response) + pulse = holder.current.latest_rate_limit_pulse + assert pulse is not None + # reset_at parsed and tz-aware + assert pulse.requests.reset_at.tzinfo is not None + + +def test_rate_limit_iso8601_with_offset_parses(): + holder = _holder() + headers = _anthropic_headers() + headers["anthropic-ratelimit-requests-reset"] = "2026-05-21T12:00:00+00:00" + response = _anthropic_response(headers=headers) + record_rate_limit_pulse_from_response(response) + pulse = holder.current.latest_rate_limit_pulse + assert pulse is not None + + +# --------------------------------------------------------------------------- +# Helper: header lookup handles case + bracket access +# --------------------------------------------------------------------------- + + +def test_header_lookup_handles_case_insensitive(): + """Some httpx Header objects do case-insensitive lookup; the helper + should work either way.""" + from agent.cost_ladder_wire import _get_header + + headers = {"Anthropic-RateLimit-Requests-Limit": "1000"} + + # Standard dict — exact lookup + assert _get_header(headers, "Anthropic-RateLimit-Requests-Limit") == "1000" + # Lowercase fallback also tries + assert _get_header(headers, "anthropic-ratelimit-requests-limit") is None # exact dict + + +def test_header_lookup_handles_object_with_get_method(): + from agent.cost_ladder_wire import _get_header + + class _CaseInsensitive: + def __init__(self, base): + self.base = {k.lower(): v for k, v in base.items()} + + def get(self, key): + return self.base.get(key.lower()) + + h = _CaseInsensitive({"Anthropic-RateLimit-Requests-Limit": "1000"}) + assert _get_header(h, "anthropic-ratelimit-requests-limit") == "1000"