fix(agent): per-request provider headers reach the wire

A provider's request_headers(user_initiated=True) answer (turn attribution,
e.g. a premium-billing initiator header) was consumed when the first
attempt's payload was built. A 429/5xx retry or a fallback activation
rebuilds the payload for the SAME API call, so the retried attempt went out
with the route's default (agent) attribution instead of the user's.

The route now claims the call's api_call_count; every attempt of that call
re-applies the headers, tool-loop follow-ups (higher count) do not, and a new
turn / session reset clears the claim.

Repro (fake profile + local HTTP stand-in, scratch probe_xinit.py): first
attempt carries the header, the retry did not (codex_responses 500 retry,
chat_completions 429 retry); both attempts carry it after this change.

The turn flag is consumed exactly once per user turn, before the route is
asked: a route with no attribution, or a request_headers hook that fails
(route_answer swallows plugin errors), no longer leaves it set, which billed
the next tool-loop call as user-initiated (r4-review P2-5).

agent/agent_init.py

@@ -2179,7 +2179,7 @@ def _init_usage_state(agent):
 # Per-session usage accounting.
 _USAGE_STATE: dict[str, Any] = {
     "_user_turn_count": 0,
-    "_is_user_initiated_turn": False,  # first API call of a user turn (provider request_headers attribution)
+    "_is_user_initiated_turn": False, "_user_initiated_api_call": None,  # request_headers turn attribution
     # Usage anchors (agent/usage_anchor.py): last response's exact usage + transcript
     # snapshot; invalidated on compaction/session switch so stale anchors never suppress compression.
     "_usage_anchor": None,

agent/turn_api_request.py

@@ -40,6 +40,24 @@ def _set_extra_header(api_kwargs: Any, key: str, value: str) -> None:
     api_kwargs["extra_headers"] = _xh
 
 
+def _apply_user_turn_request_headers(agent: Any, api_kwargs: Any, api_call_count: Any) -> None:
+    """Provider per-request headers (turn attribution). The first API call of a user turn consumes the
+    turn flag exactly once, whatever the route answers (or if its hook fails), and claims its
+    ``api_call_count``: every attempt of that call (retry, fallback activation) rebuilds the payload with
+    the same count and keeps the attribution; tool-loop follow-ups get the route's default."""
+    if getattr(agent, "_is_user_initiated_turn", False):
+        agent._is_user_initiated_turn = False
+        agent._user_initiated_api_call = api_call_count
+    if getattr(agent, "_user_initiated_api_call", None) != api_call_count:
+        return
+    from agent.provider_route_hooks import route_answer
+    _request_headers = route_answer(
+        "request_headers", provider=agent.provider or "", base_url=agent.base_url or "",
+        model=agent.model or "", user_initiated=True)
+    for _key, _value in (_request_headers or {}).items():
+        _set_extra_header(api_kwargs, _key, _value)
+
+
 def _fire_pre_api_request_hook(
     agent: Any, api_kwargs: Any, api_messages: Any, _llm_middleware_trace: Any, *, messages: Any,
     original_user_message: Any, approx_tokens: Any, total_chars: Any, retry_count: Any,
@@ -134,17 +152,7 @@ def build_request_payload(
     # retry must bypass the cache.
     if agent._empty_content_retries > 0 and agent._is_openrouter_url():
         _set_extra_header(api_kwargs, "X-OpenRouter-Cache", "false")
-    # Provider per-request headers (turn attribution): the first call of a user turn is consumed
-    # by the route that attributes it; tool-loop follow-ups get the route's default.
-    if getattr(agent, "_is_user_initiated_turn", False):
-        from agent.provider_route_hooks import route_answer
-        _request_headers = route_answer(
-            "request_headers", provider=agent.provider or "", base_url=agent.base_url or "",
-            model=agent.model or "", user_initiated=True)
-        if _request_headers:
-            for _key, _value in _request_headers.items():
-                _set_extra_header(api_kwargs, _key, _value)
-            agent._is_user_initiated_turn = False
+    _apply_user_turn_request_headers(agent, api_kwargs, api_call_count)
     try:
         from hermes_cli.middleware import apply_llm_request_middleware
 

agent/turn_context.py

@@ -1126,6 +1126,7 @@ def build_turn_context(
     # Turn-initiator state for per-request provider headers: the first API call of this
     # user turn is user-initiated; tool-loop follow-ups are agent-initiated.
     agent._is_user_initiated_turn = True
+    agent._user_initiated_api_call = None
 
     # Preserve the original user message (no nudge injection).
     original_user_message = persist_user_message if persist_user_message is not None else user_message

providers/base.py

@@ -467,7 +467,7 @@ class ProviderProfile:
 
     def request_headers(self, *, base_url: str, model: str = "", user_initiated: bool = False) -> dict[str, str]:
         """Per-request headers (``extra_headers``) for *base_url*; ``user_initiated`` is True only on the
-        first API call of a user turn (attribution). Pure: core memoizes the answer per argument set."""
+        first API call of a user turn, every attempt of it (attribution). Pure: core memoizes the answer per argument set."""
         return {}
 
     def max_tokens_param(self, *, base_url: str, model: str = "") -> str | None:

run_agent.py

@@ -406,8 +406,8 @@ class AIAgent(
         self._compaction_prompt_drift_logged = False
         # Who wrote the current turn. build_turn_context() sets it at the start of every turn.
         self._turn_author = None
-        # True for the first API call of a user turn, False for tool-loop follow-ups (provider request_headers).
-        self._is_user_initiated_turn = False
+        # First API call of a user turn (provider request_headers) and the api_call_count that claimed it.
+        self._is_user_initiated_turn, self._user_initiated_api_call = False, None
 
         self._transition_context_engine_session(
             old_session_id=old_session_id, new_session_id=getattr(self, "session_id", None),

tests/agent/test_provider_wire_seams.py

@@ -111,13 +111,13 @@ def test_image_request_client_carries_image_client_headers(wire_profile):
     assert built[-1]["default_headers"]["X-Wire"] == "1"
 
 
-def _payload(agent):
+def _payload(agent, api_call_count=1):
     from agent.turn_api_request import build_request_payload
 
     agent._empty_content_retries = 0
     build = build_request_payload(
         agent, api_messages=[{"role": "user", "content": "hi"}], _moa_prepared_request=None,
-        tools_for_api=agent.tools, system_message="", api_call_count=1, api_request_id="r",
+        tools_for_api=agent.tools, system_message="", api_call_count=api_call_count, api_request_id="r",
         effective_task_id="t", turn_id="u")
     return build.api_kwargs
 
@@ -127,11 +127,32 @@ def test_user_turn_attribution_is_consumed_by_the_first_call_only(wire_profile):
     agent._is_user_initiated_turn = True
     assert _payload(agent).get("extra_headers") == {"X-Attr": "user"}
     assert agent._is_user_initiated_turn is False
-    assert "X-Attr" not in (_payload(agent).get("extra_headers") or {})  # tool-loop follow-up
+    assert "X-Attr" not in (_payload(agent, api_call_count=2).get("extra_headers") or {})  # tool-loop follow-up
 
 
-def test_route_without_attribution_leaves_the_turn_flag_set(wire_profile):
+def test_every_attempt_of_the_first_call_keeps_user_turn_attribution(wire_profile):
+    # A 429/5xx retry (or a fallback activation) rebuilds the payload for the SAME API call: still the user's.
+    agent = _agent("wirefake", HOST)
+    agent._is_user_initiated_turn = True
+    assert _payload(agent, api_call_count=1).get("extra_headers") == {"X-Attr": "user"}
+    assert _payload(agent, api_call_count=1).get("extra_headers") == {"X-Attr": "user"}  # retried attempt
+    assert "X-Attr" not in (_payload(agent, api_call_count=2).get("extra_headers") or {})
+    agent.reset_session_state()
+    assert "X-Attr" not in (_payload(agent, api_call_count=1).get("extra_headers") or {})
+
+
+def test_route_without_attribution_still_consumes_the_turn_flag(wire_profile):
     agent = _agent("custom", "https://llm.example.test/v1")
     agent._is_user_initiated_turn = True
     assert "X-Attr" not in (_payload(agent).get("extra_headers") or {})
-    assert agent._is_user_initiated_turn is True
+    assert agent._is_user_initiated_turn is False
+
+
+def test_failing_request_headers_hook_never_bills_a_follow_up_as_the_user(wire_profile):
+    profile, _ = wire_profile
+    agent = _agent("wirefake", HOST)
+    agent._is_user_initiated_turn = True
+    with patch.object(type(profile), "request_headers", side_effect=RuntimeError("plugin bug")):
+        assert "X-Attr" not in (_payload(agent, api_call_count=1).get("extra_headers") or {})
+    assert agent._is_user_initiated_turn is False
+    assert "X-Attr" not in (_payload(agent, api_call_count=2).get("extra_headers") or {})  # tool-loop follow-up