Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
Expand Up @@ -557,7 +557,7 @@ def _wrap(
kwargs = _apply_prompt_safety(span, kwargs, name)
_handle_input(span, event_logger, kwargs)
start_time = time.perf_counter()
# ST-10.4 (review-driven 2026-05-17): make the anthropic.chat span
# Make the anthropic.chat span
# the AMBIENT OTel context for the duration of the SDK call so the
# FortifyRoot retry handler (wrapping
# ``anthropic._base_client.SyncHttpxClientWrapper.send``) can
Expand Down Expand Up @@ -698,7 +698,7 @@ async def _awrap(
kwargs = await asyncio.to_thread(_apply_prompt_safety, span, kwargs, name) # FR: async safety
await _ahandle_input(span, event_logger, kwargs)
start_time = time.perf_counter()
# ST-10.4 (review-driven 2026-05-17): see sync _wrap above for
# See sync _wrap above for
# rationale on use_span(end_on_exit=False) around the wrapped call.
try:
with trace.use_span(span, end_on_exit=False):
Expand Down Expand Up @@ -898,16 +898,16 @@ def _instrument(self, **kwargs):
except Exception:
pass # that's ok, we don't want to fail if some methods do not exist

# ST-10.4: per-attempt retry_attempt emission via private
# Per-attempt retry_attempt emission via private
# ``anthropic._base_client`` httpx wrapper classes. Guarded
# against missing private symbols (logs warning + skips emission).
# Pass the same tracer_provider the rest of the instrumentor
# uses so retry_attempt spans land in the same exporter as the
# anthropic logical span (review-driven 2026-05-16 fix).
# anthropic logical span.
instrument_retry_emitter(tracer_provider=tracer_provider)

def _uninstrument(self, **kwargs):
uninstrument_retry_emitter() # ST-10.4 symmetry
uninstrument_retry_emitter() # retry-emitter symmetry
for wrapped_method in WRAPPED_METHODS:
wrap_package = wrapped_method.get("package")
wrap_object = wrapped_method.get("object")
Expand Down
Original file line number Diff line number Diff line change
@@ -1,7 +1,6 @@
"""ST-10.4 retry-aware emission for the Anthropic direct SDK.
"""Retry-aware emission for the Anthropic direct SDK.

Per RETRY_LOOP.md §4.4 Anthropic row + §4.7 suppression discipline +
ST-10.0 hook-table addendum (in phase_st10_retryloop.txt):
Design contract:

- Hook ``anthropic._base_client.SyncHttpxClientWrapper.send`` and
``AsyncHttpxClientWrapper.send`` — fires once per HTTP attempt
Expand All @@ -16,17 +15,16 @@
- Endpoint allow-list: only LLM endpoints emit retry_attempt spans
(``/v1/messages`` for the modern Messages API,
``/v1/complete`` for legacy completions). Anything else (token
refresh, model listing, etc.) does NOT emit retry_attempt — per
§4.4.1 allow-listing requirement.
- Suppression discipline (§4.7): check BOTH the OTel context
refresh, model listing, etc.) does NOT emit retry_attempt.
- Suppression discipline: check BOTH the OTel context
``SUPPRESS_LANGUAGE_MODEL_INSTRUMENTATION_KEY`` AND the shared
``is_framework_owned()`` registry. If either says "suppress",
skip emission. Prevents framework retries (LiteLLM / LangChain /
LlamaIndex) from DOUBLE-emitting when both framework wrappers
and direct-SDK wrappers are active.
- Parent resolution: use the current OTel ambient span. If invalid
or absent, skip gracefully (no orphan retry_attempt span). The
§4.5 backend dedup degrades gracefully.
or absent, skip gracefully (no orphan retry_attempt span). Backend
dedup degrades gracefully.

Tests: see ``tests/test_retry_attempt_emission.py``.
"""
Expand Down Expand Up @@ -54,7 +52,7 @@
logger = logging.getLogger(__name__)


# ST-10 §4.4 / §4.5 constants.
# Retry-attempt constants.
_FR_SPAN_ROLE_KEY = "fortifyroot.span.role"
_FR_LLM_ATTEMPT_SPAN_NAME_PREFIX = "fortifyroot.anthropic"
_FR_SPAN_ROLE_LLM_ATTEMPT = "llm_attempt"
Expand Down Expand Up @@ -199,9 +197,7 @@ def _start_attempt_span(request: Any, parent_span: "trace.Span") -> "trace.Span"
# ``anthropic/__init__.py`` ``_wrap``). The earlier draft of
# this handler used lower-case "anthropic" for cross-handler
# consistency, but that mismatched the upstream Anthropic
# instrumentor and the fr-system-tests ProviderModel
# ``gen_ai_system="Anthropic"`` assertion — ST-10.4
# review-driven fix 2026-05-17. Backend provider grouping is
# instrumentor and provider-model assertions. Backend provider grouping is
# case-sensitive in practice; canonicalisation happens via
# ``event_provider`` (which is always lower-case in tests).
"gen_ai.system": "Anthropic",
Expand Down Expand Up @@ -229,11 +225,11 @@ def _finalize_success(span: "trace.Span", response: Any, *, is_streaming: bool =

For non-streaming responses (regardless of 2xx vs non-2xx), parse
the JSON body and extract usage tokens + response id + response
model. Per RETRY_LOOP.md §4.4 token-usage rule (around line 164):
model. Per the retry-attempt token-usage rule:
wrappers MUST extract usage from the response body whenever it's
present, regardless of whether the attempt succeeded — some
failures DO consume tokens and the provider returns usage in the
error body. §4.5 backend dedup makes the qualifying retry_attempt
error body. Backend dedup makes the qualifying retry_attempt
canonical (even single-attempt), so usage must live on this span.

Streaming responses (``stream=True`` passed to ``send()``) skip
Expand All @@ -257,7 +253,7 @@ def _finalize_success(span: "trace.Span", response: Any, *, is_streaming: bool =
pass

# Non-streaming body parse — applies to BOTH 2xx and non-2xx per the
# §4.4 token-usage rule. ``_extract_usage_from_body`` is fully
# retry-attempt token-usage rule. ``_extract_usage_from_body`` is fully
# defensive (missing fields / parse failure / no ``usage`` block all
# degrade silently).
if not is_streaming:
Expand Down Expand Up @@ -348,22 +344,21 @@ def _should_emit_for(request: Any) -> bool:
def _sync_send_wrapper(wrapped, instance, args, kwargs):
"""Wraps ``anthropic._base_client.SyncHttpxClientWrapper.send``.

Direct-SDK wrappers do NOT register tokens in the §4.7.1 framework
Direct-SDK wrappers do NOT register tokens in the framework-attempt
registry (the registry's contract reserves registration for FRAMEWORK
wrappers — LiteLLM / LangChain / LlamaIndex). They only consult via
``is_framework_owned()``. Self-registering would falsely suppress
concurrent direct-SDK calls on the same OS thread, which manifests
most visibly under asyncio (multiple tasks sharing one thread).

Streaming skip (ST-10.4 review-driven 2026-05-17): when
Streaming skip: when
``stream=True`` is passed to send, this wrapper SKIPS retry_attempt
emission. Streaming retry_attempts cannot carry usage (SSE stream
can't be peeked) but §4.5 dedup would still promote them to the
can't be peeked) but backend dedup would still promote them to the
canonical LLMUsageEvent, producing zero-token events. Leaving the
parent ``anthropic.chat`` span as the canonical (it gets full
usage from the Anthropic streaming wrapper). Streaming retry-loop
detection is the deferred follow-up
``ST-10.4-FOLLOWUP-streaming-usage``.
detection is a deferred streaming-usage follow-up.
"""
request = args[0] if args else kwargs.get("request")
if request is None or not _should_emit_for(request):
Expand Down Expand Up @@ -490,7 +485,7 @@ def instrument_retry_emitter(tracer_provider=None) -> None:
)
if not sync_ok:
logger.warning(
"ST-10.4: anthropic._base_client.%s.%s missing/incompatible; "
"Anthropic retry emitter: anthropic._base_client.%s.%s missing/incompatible; "
"skipping sync retry_attempt emission. Normal anthropic "
"instrumentation is unaffected.",
_SYNC_WRAPPER_CLASS, _WRAPPED_METHOD,
Expand All @@ -504,13 +499,13 @@ def instrument_retry_emitter(tracer_provider=None) -> None:
)
except Exception as e:
logger.warning(
"ST-10.4: failed to wrap anthropic sync httpx send (%s); "
"Anthropic retry emitter: failed to wrap anthropic sync httpx send (%s); "
"retry_attempt emission disabled for sync path",
e,
)
if not async_ok:
logger.warning(
"ST-10.4: anthropic._base_client.%s.%s missing/incompatible; "
"Anthropic retry emitter: anthropic._base_client.%s.%s missing/incompatible; "
"skipping async retry_attempt emission. Normal anthropic "
"instrumentation is unaffected.",
_ASYNC_WRAPPER_CLASS, _WRAPPED_METHOD,
Expand All @@ -524,7 +519,7 @@ def instrument_retry_emitter(tracer_provider=None) -> None:
)
except Exception as e:
logger.warning(
"ST-10.4: failed to wrap anthropic async httpx send (%s); "
"Anthropic retry emitter: failed to wrap anthropic async httpx send (%s); "
"retry_attempt emission disabled for async path",
e,
)
Expand All @@ -543,7 +538,7 @@ def uninstrument_retry_emitter() -> None:
unwrap(f"{_ANTHROPIC_BASE_CLIENT_MODULE}.{cls}", _WRAPPED_METHOD)
except Exception:
logger.debug(
"ST-10.4: anthropic unwrap of %s.%s failed (likely "
"Anthropic retry emitter: unwrap of %s.%s failed (likely "
"wrap was never installed for this variant)",
cls, _WRAPPED_METHOD,
exc_info=True,
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -29,10 +29,10 @@


class _NoFortifyRootSpanExporter(InMemorySpanExporter):
"""ST-10.4 (2026-05-17): filter out any span whose name starts
"""FortifyRoot retry-attempt (2026-05-17): filter out any span whose name starts
with ``fortifyroot.`` from the upstream-test span exporter.

ST-10.4 added per-attempt ``fortifyroot.anthropic.attempt_1``
FortifyRoot retry-attempt added per-attempt ``fortifyroot.anthropic.attempt_1``
sibling spans under every anthropic logical call (once the
Anthropic ``_wrap`` started using ``trace.use_span`` so the retry
handler can find the parent span). Upstream / legacy Anthropic
Expand All @@ -42,7 +42,7 @@ class _NoFortifyRootSpanExporter(InMemorySpanExporter):
retry_attempt sibling is also exported.

Mirrors the OpenAI test conftest pattern (added 2026-05-16) and
the LangChain CI-hardening pattern (2026-05-15). ST-10.4 unit
the LangChain CI-hardening pattern (2026-05-15). FortifyRoot retry-attempt unit
tests in ``tests/test_retry_attempt_emission.py`` use their own
``fresh_tracer`` fixture (not this one), so they continue to see
retry_attempt spans and aren't affected by the filter.
Expand Down
Original file line number Diff line number Diff line change
@@ -1,6 +1,6 @@
"""Tests for ST-10.4 Anthropic direct-SDK retry-attempt emission.
"""Tests for FortifyRoot retry-attempt Anthropic direct-SDK retry-attempt emission.

Covers (per RETRY_LOOP.md §4.4 Anthropic row + §4.7 suppression):
Covers (per retry-loop design notes §4.4 Anthropic row + §4.7 suppression):
- Instrumentor symmetry: install/uninstall flips state cleanly and is
idempotent.
- Single-attempt happy path: ONE llm_attempt span under the active
Expand Down Expand Up @@ -101,7 +101,7 @@ def _make_response(status_code: int = 200, request_id: str = "req-abc",
body: Optional[dict] = None) -> SimpleNamespace:
"""Fake ``httpx.Response``. Non-streaming retry_attempt path now
parses the body for usage / response.id / response.model — see the
2026-05-13 review-driven C2 fix.
2026-05-13 retry-parent resolution fix.
"""
if body is None:
body = {
Expand Down Expand Up @@ -382,7 +382,7 @@ def test_framework_registry_suppression_skips_emission(fresh_tracer):


def test_direct_sdk_wrapper_does_not_self_register_in_framework_registry(fresh_tracer):
"""REGRESSION GUARD (review-driven fix 2026-05-13): direct-SDK
"""REGRESSION GUARD: direct-SDK
wrappers MUST NOT register tokens in the §4.7.1 framework registry.
Self-registration causes false suppression of concurrent direct-SDK
calls sharing the same thread (asyncio repro in
Expand All @@ -409,7 +409,7 @@ def wrapped(*a, **kw):


def test_two_concurrent_async_sends_each_emit_a_retry_attempt(fresh_tracer):
"""REGRESSION GUARD (review-driven fix 2026-05-13): two asyncio
"""REGRESSION GUARD: two asyncio
tasks sharing one OS thread MUST each emit their own retry_attempt."""
tracer, exporter, _ = fresh_tracer

Expand Down Expand Up @@ -477,12 +477,12 @@ def test_llm_endpoints_emit_with_correct_operation(fresh_tracer, path, expected_


# ---------------------------------------------------------------------------
# §4.5-driven usage-extraction policy (review-driven fix 2026-05-13).
# §4.5-driven usage-extraction policy.
# ---------------------------------------------------------------------------

def test_non_streaming_response_body_populates_usage_id_and_model(fresh_tracer):
"""Backend dedup makes the retry_attempt span the canonical
LLMUsageEvent (see ``proc_llm_extractor.go``). Anthropic non-streaming
LLMUsageEvent (see ``backend LLM extractor``). Anthropic non-streaming
responses MUST carry usage on the retry_attempt span."""
tracer, exporter, _ = fresh_tracer
request = _make_request(model="claude-haiku-4-5")
Expand Down Expand Up @@ -510,7 +510,7 @@ def test_non_streaming_response_body_populates_usage_id_and_model(fresh_tracer):


def test_streaming_request_skips_emission_entirely(fresh_tracer):
"""ST-10.4 (review-driven 2026-05-17): when ``stream=True`` is
"""FortifyRoot retry-attempt: when ``stream=True`` is
passed to send, the wrap SKIPS retry_attempt emission entirely.
See the openai analog for full rationale. The parent
``anthropic.chat`` span stays the canonical LLMUsageEvent.
Expand Down Expand Up @@ -551,8 +551,8 @@ def explode():


def test_non_2xx_response_with_usage_in_body_populates_usage_tokens(fresh_tracer):
"""REGRESSION GUARD (review-driven follow-up 2026-05-13): per
RETRY_LOOP.md §4.4 token-usage rule, wrappers MUST extract usage
"""REGRESSION GUARD: per
retry-loop design notes §4.4 token-usage rule, wrappers MUST extract usage
from the response body whenever present, regardless of success.
Some failures consume tokens and the provider returns usage in
the error body."""
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -131,8 +131,7 @@ def is_metrics_enabled() -> bool:
def _with_tracer_wrapper(func):
"""Helper for providing tracer for wrapper functions.

``tracer_provider`` (added 2026-05-16 for the ST-10.4 review-driven
fix) is plumbed alongside the tracer so the bedrock-runtime client
``tracer_provider`` is plumbed alongside the tracer so the bedrock-runtime client
wrap can pass it to ``install_event_hooks_on_client`` — letting the
retry handler's event hooks emit spans through the same provider
the rest of the instrumentor uses.
Expand Down Expand Up @@ -201,12 +200,12 @@ def _wrap(
client.converse_stream = _instrumented_converse_stream(
client.converse_stream, tracer, metric_params, event_logger
)
# ST-10.4: register per-attempt botocore event hooks on the
# Register per-attempt botocore event hooks on the
# bedrock-runtime client. Public botocore API; emits one
# retry_attempt sibling span per HTTP attempt under the outer
# bedrock.completion / bedrock.converse span. ``tracer_provider``
# is passed so retry_attempt spans go to the same provider as
# the bedrock logical span — review-driven 2026-05-16 fix.
# the bedrock logical span.
install_event_hooks_on_client(client, tracer_provider=tracer_provider)
return client
except Exception as e:
Expand Down Expand Up @@ -254,7 +253,7 @@ def with_instrumentation(*args, **kwargs):

span = tracer.start_span(_BEDROCK_INVOKE_SPAN_NAME, kind=SpanKind.CLIENT)

# ST-10.4: make the streaming span the AMBIENT OTel context for
# Make the streaming span the AMBIENT OTel context for
# the duration of the underlying boto3 call so per-attempt
# botocore event hooks (before-send.bedrock-runtime.*) can
# resolve this span as the retry_attempt parent. The span is
Expand Down Expand Up @@ -304,7 +303,7 @@ def with_instrumentation(*args, **kwargs):

span = tracer.start_span(_BEDROCK_CONVERSE_SPAN_NAME, kind=SpanKind.CLIENT)
kwargs = _apply_converse_prompt_safety(span, kwargs, _BEDROCK_CONVERSE_SPAN_NAME)
# ST-10.4: see _instrumented_model_invoke_with_response_stream
# See _instrumented_model_invoke_with_response_stream
# for the rationale on use_span(end_on_exit=False).
stream_start_time = time.perf_counter()
with trace.use_span(span, end_on_exit=False):
Expand Down
Loading
Loading