fix: correct per-operation tracing and tidy HTTP transport edge cases

Pipeline tracing:
- The per-operation HttpTracer lifecycle (operation_started /
  operation_succeeded / operation_failed) was emitted from TracingPolicy,
  which runs inside the retry and redirect wrappers and is re-entered once per
  attempt/hop. Gated on the first attempt, a call that failed once and then
  succeeded on a retry reported operation_failed and never operation_succeeded,
  and a redirect whose later hop failed reported operation_succeeded.
- Add OperationTracingPolicy at a new outermost Stage.OPERATION that brackets
  the whole call from outside the wrappers, emitting operation_started before
  the chain and exactly one operation_succeeded / operation_failed on the final
  outcome. TracingPolicy keeps the per-attempt span and the per-request events
  (request_sent / response_*); default_pipeline wires both.

Transports and utilities:
- urllib: stop dropping a valid Content-Length under Content-Encoding.
  http.client does not decode content encodings, so the served body is the
  wire payload whose length is the upstream Content-Length; the header is
  accurate and is now surfaced (the decompressing requests/httpx/aiohttp
  adapters still drop it).
- urllib and asyncio: report "Invalid status code" for an out-of-range status,
  matching the other adapters.
- asyncio: match the chunked transfer-coding by token rather than substring so
  a coding such as "x-chunked" is not mistaken for chunked framing.
- Share one cross-origin helper (http.common.url._origin) between the redirect
  and auth policies instead of duplicating it.

Docs and the committed API surface baseline are updated accordingly.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

Author: OmarAlJarrah

docs/pipelines.md

@@ -37,31 +37,44 @@ two parallel variants:
 ```python
 from dexpace.sdk.http.stdlib import UrllibHttpClient
 from dexpace.sdk.core.pipeline import Pipeline
-from dexpace.sdk.core.pipeline.policies import LoggingPolicy, RetryPolicy, TracingPolicy
+from dexpace.sdk.core.pipeline.policies import (
+    LoggingPolicy,
+    OperationTracingPolicy,
+    RetryPolicy,
+    TracingPolicy,
+)
 
 with Pipeline(
     UrllibHttpClient(),
-    policies=[TracingPolicy(), LoggingPolicy(), RetryPolicy()],
+    policies=[OperationTracingPolicy(), TracingPolicy(), LoggingPolicy(), RetryPolicy()],
 ) as pipeline:
     response = pipeline.run(request, dispatch_ctx)
 ```
 
 The chain runs in declaration order. For the example above:
 
-1. `TracingPolicy` opens a span.
-2. `LoggingPolicy` logs the request.
-3. `RetryPolicy` invokes the transport; retries on transient failures.
-4. The transport runner calls `UrllibHttpClient.execute(request)`.
-5. The unwinding mirror order: logging emits the response, tracing
-   closes the span.
+1. `OperationTracingPolicy` emits `operation_started` for the whole call.
+2. `TracingPolicy` opens a span.
+3. `LoggingPolicy` logs the request.
+4. `RetryPolicy` invokes the transport; retries on transient failures.
+5. The transport runner calls `UrllibHttpClient.execute(request)`.
+6. The unwinding mirrors that order: logging emits the response, tracing
+   closes the span, and operation-tracing emits the single
+   `operation_succeeded` / `operation_failed` once the call settles.
+
+In the canonical `default_pipeline`, `OperationTracingPolicy` sits *outside*
+the redirect and retry wrappers (so the per-operation lifecycle fires once and
+reflects the final outcome), while `TracingPolicy` sits *inside* them and opens
+one span per attempt / hop.
 
 ## Built-in policies
 
 | Policy                              | Purpose                                                  |
 |-------------------------------------|----------------------------------------------------------|
 | `RetryPolicy` / `AsyncRetryPolicy`  | Retry transient failures with backoff + `Retry-After`. Auto-buffers single-use request bodies when `total_retries > 0`.    |
 | `LoggingPolicy`                     | Structured request/response logs with URL redaction.      |
-| `TracingPolicy`                     | Open a span per request; OTel semantic-conv attributes.   |
+| `OperationTracingPolicy`            | Emit the per-operation lifecycle (`operation_started`, then one `operation_succeeded`/`operation_failed`) from outside the retry/redirect loop. |
+| `TracingPolicy`                     | Open a span per attempt and emit per-request tracer events; OTel semantic-conv attributes. |
 | `BearerTokenPolicy` (auth)          | Acquire + cache + apply OAuth bearer tokens.              |
 | `KeyCredentialPolicy` (auth)        | Stamp an API key into a configurable header.              |
 | `BasicAuthPolicy` (auth)            | `Authorization: Basic <base64>`.                          |
@@ -88,16 +101,19 @@ Per-call opt-outs follow a convention:
 second `iter_bytes()` call raises `RuntimeError`. To keep retries safe
 without forcing every caller to remember `to_replayable()`, both
 `RetryPolicy.send` and `AsyncRetryPolicy.send` inspect the body up front
-and, when `total_retries > 0`, swap in a buffered replayable copy before
-the first attempt:
+and, when the *effective* retry total for the call is positive, swap in a
+buffered replayable copy before the first attempt:
 
 ```python
-if total_retries > 0 and request.body is not None and not request.body.is_replayable():
+settings = self._configure_settings(ctx.options)
+if settings["total"] > 0 and request.body is not None and not request.body.is_replayable():
     request = request.with_body(request.body.to_replayable())
 ```
 
-`total_retries == 0` (e.g. `RetryPolicy.no_retries()`) skips the buffering
-step so callers who explicitly opt out of retries pay no memory cost for a
-copy they will never use. Already-replayable bodies (`from_bytes`,
-`from_string`, `from_form`) flow through untouched because
-`to_replayable()` returns `self`.
+The decision reads the effective total *after* merging any per-call
+`retry_total` override (from `ctx.options`) with the constructor default, so a
+per-call `retry_total=3` over an instance built with `total_retries=0` still
+buffers, and a per-call `retry_total=0` (or `RetryPolicy.no_retries()`) skips
+the buffering so callers who opt out of retries pay no memory cost for a copy
+they will never use. Already-replayable bodies (`from_bytes`, `from_string`,
+`from_form`) flow through untouched because `to_replayable()` returns `self`.

packages/dexpace-sdk-core/src/dexpace/sdk/core/http/auth/policies.py

@@ -14,6 +14,7 @@
 from ...pipeline.policy import Policy
 from ...pipeline.stage import Stage
 from ...util.clock import ASYNC_SYSTEM_CLOCK, SYSTEM_CLOCK, AsyncClock, Clock
+from ..common.url import _origin
 from .access_token import AccessTokenInfo, TokenRequestOptions
 from .challenge import parse_challenges
 from .challenge_handler import ChallengeHandler
@@ -439,29 +440,9 @@ async def _authorize(
         return request.with_header("Authorization", f"{token.token_type} {token.token}")
 
 
-_DEFAULT_PORTS: dict[str, int] = {"https": 443, "http": 80}
 _AUTH_ORIGIN_KEY: str = "_auth_origin"
 
 
-def _origin(url: Url) -> tuple[str, str, int | None]:
-    """Return the ``(scheme, host, port)`` origin tuple for ``url``.
-
-    The scheme and host are lower-cased and the port is resolved to its
-    scheme default (443 for https, 80 for http) when not explicit, so two
-    URLs that differ only in an implied/explicit default port compare equal.
-
-    Args:
-        url: The URL to derive an origin from.
-
-    Returns:
-        A ``(scheme, host, effective_port)`` tuple suitable for equality
-        comparison.
-    """
-    scheme = url.scheme.lower()
-    port = url.port if url.port is not None else _DEFAULT_PORTS.get(scheme)
-    return scheme, url.host.lower(), port
-
-
 def _crosses_recorded_origin(request: Request, ctx: PipelineContext) -> bool:
     """Report whether ``request`` left the origin recorded for this operation.
 

packages/dexpace-sdk-core/src/dexpace/sdk/core/http/common/url.py

@@ -320,4 +320,28 @@ def parse(cls, raw: str) -> Self:
         )
 
 
+_DEFAULT_PORTS: dict[str, int] = {"https": 443, "http": 80}
+
+
+def _origin(url: Url) -> tuple[str, str, int | None]:
+    """Return the ``(scheme, host, effective_port)`` origin tuple for ``url``.
+
+    The scheme and host are lower-cased and the port is resolved to its scheme
+    default (443 for https, 80 for http) when not explicit, so two URLs that
+    differ only in an implied/explicit default port compare equal. Shared by
+    the redirect and auth policies for their cross-origin checks so the origin
+    definition stays in one place.
+
+    Args:
+        url: The URL to derive an origin from.
+
+    Returns:
+        A ``(scheme, host, effective_port)`` tuple suitable for equality
+        comparison.
+    """
+    scheme = url.scheme.lower()
+    port = url.port if url.port is not None else _DEFAULT_PORTS.get(scheme)
+    return scheme, url.host.lower(), port
+
+
 __all__ = ["QueryParams", "Url"]

packages/dexpace-sdk-core/src/dexpace/sdk/core/pipeline/defaults.py

@@ -19,7 +19,7 @@
 from .policies.redirect import RedirectPolicy
 from .policies.retry import RetryPolicy
 from .policies.set_date import SetDatePolicy
-from .policies.tracing_policy import TracingPolicy
+from .policies.tracing_policy import OperationTracingPolicy, TracingPolicy
 from .staged_builder import StagedPipelineBuilder
 
 if TYPE_CHECKING:
@@ -44,15 +44,21 @@ def default_pipeline(
     """Pre-configured `StagedPipelineBuilder` with the canonical stack.
 
     Wires the policies that most consumers want by default in the order their
-    stages dictate: redirect → idempotency → retry → set-date →
-    client-identity → auth → logging → tracing. Each policy is opt-out (pass
-    ``None``) or opt-in-with-override (pass a pre-configured instance to
-    replace the default).
+    stages dictate: operation-tracing → redirect → idempotency → retry →
+    set-date → client-identity → auth → logging → tracing. Each policy is
+    opt-out (pass ``None``) or opt-in-with-override (pass a pre-configured
+    instance to replace the default).
 
     Idempotency sits before retry so a write request's ``Idempotency-Key`` is
     minted once and reused across every retry; ``set-date`` and
     ``client-identity`` sit just inside the retry wrapper.
 
+    Two tracing policies cooperate: `OperationTracingPolicy` brackets the whole
+    operation from outside the redirect / retry wrappers (so the per-operation
+    lifecycle fires once and reflects the final outcome), while `TracingPolicy`
+    opens a span and emits per-request events inside the wrappers, once per hop
+    / attempt.
+
     Args:
         client: Terminal HTTP transport.
         redirect: Override for `RedirectPolicy`. ``None`` uses defaults.
@@ -73,6 +79,9 @@ def default_pipeline(
         immediate ``.build()``.
     """
     builder = StagedPipelineBuilder(client)
+    # Sorts to Stage.OPERATION (outermost), bracketing every hop / attempt so
+    # the per-operation lifecycle fires once on the final outcome.
+    builder.append(OperationTracingPolicy())
     builder.append(redirect or RedirectPolicy())
     builder.append(idempotency or IdempotencyPolicy())
     builder.append(retry or RetryPolicy())

packages/dexpace-sdk-core/src/dexpace/sdk/core/pipeline/policies/__init__.py

@@ -17,7 +17,7 @@
 from .redirect import RedirectPolicy
 from .retry import RetryMode, RetryPolicy
 from .set_date import SetDatePolicy
-from .tracing_policy import TracingPolicy
+from .tracing_policy import OperationTracingPolicy, TracingPolicy
 
 __all__ = [
     "AsyncClientIdentityPolicy",
@@ -28,6 +28,7 @@
     "ClientIdentityPolicy",
     "IdempotencyPolicy",
     "LoggingPolicy",
+    "OperationTracingPolicy",
     "RedirectPolicy",
     "RequestHistory",
     "RetryMode",

packages/dexpace-sdk-core/src/dexpace/sdk/core/pipeline/policies/redirect.py

@@ -29,7 +29,7 @@
 from typing import TYPE_CHECKING, ClassVar, Literal, cast
 from urllib.parse import urljoin
 
-from ...http.common.url import Url
+from ...http.common.url import Url, _origin
 from ...http.request.method import Method
 from ..policy import Policy
 from ..stage import Stage
@@ -43,26 +43,6 @@
 
 _REDIRECT_STATUSES: frozenset[int] = frozenset({301, 302, 303, 307, 308})
 _CONTENT_HEADER_PREFIX: str = "content-"
-_DEFAULT_PORTS: dict[str, int] = {"https": 443, "http": 80}
-
-
-def _origin(url: Url) -> tuple[str, str, int | None]:
-    """Return the ``(scheme, host, port)`` origin tuple for ``url``.
-
-    The scheme and host are lower-cased and the port is resolved to its
-    scheme default (443 for https, 80 for http) when not explicit, so two
-    URLs that differ only in an implied/explicit default port compare equal.
-
-    Args:
-        url: The URL to derive an origin from.
-
-    Returns:
-        A ``(scheme, host, effective_port)`` tuple suitable for equality
-        comparison.
-    """
-    scheme = url.scheme.lower()
-    port = url.port if url.port is not None else _DEFAULT_PORTS.get(scheme)
-    return scheme, url.host.lower(), port
 
 
 #: ``ctx.data`` key holding the per-operation ``HttpTracer``. The first policy