Reset metadata between retry attempts per spec 3.4

CHANGELOG.md

@@ -26,6 +26,10 @@ The format follows [Keep a Changelog](https://keepachangelog.com/en/1.1.0/). The
 - **Three new patterns docs.** `docs/patterns/state-migration-on-resume.md`, `docs/patterns/caller-supplied-trace-identifiers.md`, and `docs/patterns/observer-state-reconciliation.md` graduate the corresponding entries from `docs/agent/non-obvious-shapes.md` into full pattern recipes with code snippets and "when this is right / when it isn't" guidance. The programmatic patterns API (`openarmature.patterns.list()` / `get(name)`) grows from 4 to 7 entries.
 - **HyperDX OTel integration test path and "Production swap" docs in the observer-hooks example.** `examples/observer-hooks/main.py`'s module docstring grows a "Production swap" section showing how to substitute the demo's `SimpleSpanProcessor` + `ConsoleSpanExporter` for `BatchSpanProcessor` + `OTLPSpanExporter` pointed at HyperDX (or any other OTLP-HTTP collector). A new opt-in integration test (`tests/integration/test_otel_hyperdx_export.py`, gated by `HYPERDX_API_KEY` + `HYPERDX_OTLP_ENDPOINT` env vars and `@pytest.mark.integration`) drives the same production export path end-to-end against a live endpoint. `opentelemetry-exporter-otlp-proto-http` lands as a dev-only dep; not promoted to a public extras group yet.
 
+### Fixed
+
+- **`RetryMiddleware` now enforces per-attempt invocation-metadata scoping** (proposal 0048 / spec observability §3.4). Each retry attempt sees only the metadata in scope at retry-loop entry plus that attempt's own writes; writes from a prior attempt that subsequently failed are discarded. Prior to this fix the retry middleware only managed the `attempt_index` ContextVar and left the invocation-metadata ContextVar unchanged across attempts, so a `set_invocation_metadata(...)` call inside a failed attempt remained visible on the retry. The fix captures the pre-attempt baseline once at retry-loop entry, resets the metadata ContextVar to that baseline on each iteration, discards the failed attempt's writes on retry-eligible and terminal failure paths, and leaves the successful attempt's writes in place so downstream nodes see them. Closes the v0.12.0 cycle's `partial` claim on proposal 0048; manifest entry flips back to `implemented`.
+
 ### Changed (breaking)
 
 - **`OpenAIProvider.ready()` default probe flipped to `chat_completions`.** A new constructor kwarg `readiness_probe: Literal["models", "chat_completions", "both"]` selects which wire path `ready()` exercises; the default is now the chat-completions path (`POST /v1/chat/completions` with `max_tokens=1`), which actually exercises the inference path. The previous catalog-only behavior is still available as `readiness_probe="models"`, and `readiness_probe="both"` runs catalog then chat for the strongest signal. Motivation: OpenAI-compatible proxies (Bifrost and similar) can return 200 on `GET /v1/models` while rejecting `POST /v1/chat/completions`, leaving the catalog probe green while every real call fails. The new default surfaces that class of failure at preflight rather than at first inference. Non-200 chat-probe responses route through `classify_http_error`, so the canonical error categories (`provider_authentication`, `provider_unavailable`, `provider_invalid_model`, etc.) surface consistently. Callers that depended on the catalog-only behavior (cost-sensitive cloud setups where every `ready()` would now bill prompt tokens) can opt back in by passing `readiness_probe="models"`.

conformance.toml

@@ -251,11 +251,14 @@ status = "not-yet"
 # Spec v0.40.0 (proposal 0048).  Read-symmetric invocation metadata.
 # Adds ``get_invocation_metadata()`` symmetric to the existing
 # ``set_invocation_metadata()`` write API.  The python implementation
-# satisfies most of the §3.4 read contract via the pre-existing
+# satisfies the §3.4 read contract via the pre-existing
 # ``current_invocation_metadata`` machinery: returns ``MappingProxyType``
 # snapshot of the current async context's view, silent no-op outside
 # an invocation, no observer emission, per-async-context scoping under
-# fan-out and parallel-branches (inherited from 0034/0040/0045).
+# fan-out and parallel-branches (inherited from 0034/0040/0045),
+# per-attempt scoping under retry (engine-side reset of
+# ``_invocation_metadata_var`` around each ``RetryMiddleware`` iteration,
+# landed in the same release cycle as 0048).
 # ``get_invocation_metadata`` lands as the canonical spec-idiomatic
 # public name; ``current_invocation_metadata`` stays as a stable
 # alias (Option A per the proposal-0048-implementation coord thread).
@@ -265,24 +268,15 @@ status = "not-yet"
 # lifecycle.  §9 is convention-only per spec — no new abstract
 # surface on ``Observer``.
 #
-# Open gap: §3.4 *per-attempt scoping under retry middleware* is NOT
-# enforced in v0.12.0 — the retry middleware
-# (``src/openarmature/graph/middleware/retry.py``) manages the
-# ``attempt_index`` ContextVar but does NOT reset the
-# invocation-metadata ContextVar between attempts.  A write from a
-# failed attempt remains visible on the next attempt.  Closing this
-# gap is a follow-on PR (engine-side reset of ``_invocation_metadata_var``
-# around each retry iteration + a pinning unit test against spec
-# fixture 045's contract).
-#
 # Conformance fixtures 043-049 introduce new directive shapes
 # (augment_metadata / capture_invocation_metadata_into /
 # capture_queryable_observer_read_into / per_attempt_behavior +
 # queryable_observers / inner_subgraphs / caller_metadata /
 # direct_call / sequential_invocations top-level keys) that the
 # cross-capability parser doesn't model; fixture-shape activation is
-# queued for a future PR.  The shipped portion of the contract is
-# pinned by explicit unit tests:
+# queued for a future PR slotted after the upcoming spec
+# conformance-adapter capability ratifies the directive vocabulary.
+# The shipped contract is pinned by explicit unit tests:
 #   - alias identity + roundtrip + immutable-mapping return:
 #     ``tests/unit/test_observability_metadata.py::
 #     test_get_invocation_metadata_is_same_callable_as_current``,
@@ -291,11 +285,14 @@ status = "not-yet"
 #   - mid-invocation augmentation visible to subsequent reads:
 #     ``::test_mid_invocation_augmentation_persists_to_next_node``;
 #   - outside-invocation empty:
-#     ``::test_current_invocation_metadata_empty_outside_invocation``.
+#     ``::test_current_invocation_metadata_empty_outside_invocation``;
+#   - per-attempt scoping under retry (mirrors spec fixture 045):
+#     ``::test_per_attempt_scoping_under_retry_discards_failed_attempt_writes``,
+#     ``::test_terminal_failure_discards_final_failed_attempt_writes``,
+#     ``::test_cancellation_discards_in_flight_attempt_writes``.
 [proposals."0048"]
-status = "partial"
+status = "implemented"
 since = "0.12.0"
-note = "Read API + §9 queryable observer pattern docs ship in v0.12.0. Spec §3.4 per-attempt scoping under retry middleware is NOT yet enforced — the python retry middleware does not reset the invocation-metadata ContextVar between attempts, so a failed attempt's writes remain visible on retry. Per-async-context scoping under fan-out and parallel-branches is fully implemented. Closing the retry-side gap is tracked for a follow-on PR."
 
 # Spec v0.41.0 (proposal 0049).  Typed LLM Completion Event — first
 # typed event variant on the observer event union.  Queued for

docs/concepts/observability.md

@@ -391,22 +391,10 @@ that context, per normal `ContextVar` semantics.
 `openarmature.observability.get_invocation_metadata()` returns an
 immutable `MappingProxyType` snapshot of the entries visible in the
 current async context's view, or an empty mapping outside any active
-invocation. Reads do NOT emit a metadata-augmentation event; the
-augmentation event signals mutations to backends, not consumer
-reads.
-
-!!! info "Per-attempt retry scoping is partial in v0.12.0"
-    Spec §3.4 specifies that values written during a failed retry
-    attempt MUST NOT carry over to subsequent attempts. The v0.12.0
-    python retry middleware manages the `attempt_index` ContextVar
-    but does NOT reset the invocation-metadata ContextVar between
-    attempts; a write from a failed attempt remains visible on the
-    next attempt. Closing this gap is tracked for a follow-on PR
-    (engine-side reset of the metadata ContextVar around each retry
-    iteration, with the per-attempt scoping unit test pinned by
-    spec fixture 045). Per-async-context scoping under fan-out and
-    parallel-branches is fully implemented; only the retry-side
-    reset is the open gap.
+invocation. The read is per-attempt scoped under retry middleware:
+values written in a prior failed attempt are not visible. Reads do
+NOT emit a metadata-augmentation event; the augmentation event
+signals mutations to backends, not consumer reads.
 
 The existing `current_invocation_metadata()` is a stable alias
 pointing at the same function; both names live in `__all__`. Pick

src/openarmature/graph/middleware/retry.py

@@ -24,6 +24,11 @@
 
 from openarmature.llm.errors import TRANSIENT_CATEGORIES
 from openarmature.observability.correlation import _reset_attempt_index, _set_attempt_index
+from openarmature.observability.metadata import (
+    _invocation_metadata_var,
+    _reset_invocation_metadata,
+    _set_invocation_metadata,
+)
 
 from ._core import NextCall
 
@@ -128,6 +133,14 @@ def __init__(
 
     async def __call__(self, state: Any, next_: NextCall) -> Mapping[str, Any]:
         attempt = 0
+        # Spec observability §3.4 per-attempt scoping: each retry
+        # attempt sees only the metadata in scope at retry-loop entry
+        # ("pre-attempt baseline") plus that attempt's own writes;
+        # writes from a prior attempt that subsequently failed do NOT
+        # carry over. Captured once outside the loop because the
+        # baseline is the metadata view at retry-middleware entry, not
+        # at each iteration.
+        pre_attempt_baseline = _invocation_metadata_var.get()
         while True:
             # Spec graph-engine §6 (clarified in v0.16.1): the wrapping
             # retry's attempt counter MUST propagate to events emitted
@@ -137,22 +150,52 @@ async def __call__(self, state: Any, next_: NextCall) -> Mapping[str, Any]:
             # reset on exit; Python's ContextVar token stack gives
             # innermost-wins precedence for free when retry middlewares
             # nest.
-            token = _set_attempt_index(attempt)
+            attempt_token = _set_attempt_index(attempt)
+            # Reset the metadata ContextVar to the pre-attempt baseline.
+            # The token captures the var's state at the moment of the
+            # set call — on failure we reset against this token to
+            # discard any writes the attempt's node body issued.
+            metadata_token = _set_invocation_metadata(pre_attempt_baseline)
             try:
                 try:
-                    return await next_(state)
+                    result = await next_(state)
+                    # Success path: keep the successful attempt's
+                    # metadata writes in scope so downstream nodes see
+                    # them. Do NOT reset metadata_token here — the
+                    # engine's outer reset (around the whole invoke)
+                    # pops the stack at invocation exit.
+                    return result
                 except Exception as exc:
                     # Spec §6.1: cancellation propagates by virtue of
                     # `CancelledError` extending `BaseException`, not
                     # `Exception` — it never enters this branch in Python.
+                    # Failure path (retry-eligible OR terminal):
+                    # discard the failed attempt's metadata writes per
+                    # §3.4. Reset BEFORE the re-raise so the caller's
+                    # error-handling path (e.g., observer hooks reading
+                    # metadata for the error span) sees the baseline,
+                    # not the failed attempt's transient state.
+                    _reset_invocation_metadata(metadata_token)
                     if attempt + 1 >= self.max_attempts or not self.classifier(exc, state):
                         raise
                     if self.on_retry is not None:
                         await self.on_retry(exc, attempt)
                     await asyncio.sleep(self.backoff(attempt))
                     attempt += 1
+                except BaseException:
+                    # Cancellation path. `CancelledError` (or other
+                    # `BaseException`) ends the attempt without retry —
+                    # spec §6.1 cancellation MUST propagate, never get
+                    # swallowed or retried. But spec §3.4 per-attempt
+                    # scoping still applies: cancellation IS a failed
+                    # attempt from the metadata-scoping perspective, so
+                    # its writes must be discarded too. Reset the token,
+                    # then re-raise. NO on_retry, NO sleep — straight
+                    # propagation.
+                    _reset_invocation_metadata(metadata_token)
+                    raise
             finally:
-                _reset_attempt_index(token)
+                _reset_attempt_index(attempt_token)
 
 
 __all__ = [

tests/unit/test_observability_metadata.py

@@ -13,6 +13,7 @@
 from __future__ import annotations
 
 import asyncio
+from collections.abc import Mapping
 from typing import Any
 
 import pytest
@@ -555,3 +556,151 @@ async def _read_after_write(_s: _SimpleState) -> dict[str, Any]:
     # Caller baseline + in-node write, both visible to the read.
     assert captured == {"tenantId": "T1", "audit_kind": "fraud"}
     assert captured_type == [MappingProxyType]
+
+
+# Spec observability §3.4 *Per-attempt scoping*: under retry
+# middleware, each attempt sees only the metadata in scope at
+# retry-entry plus that attempt's own writes; failed-attempt
+# writes are discarded along with the attempt itself. The pin
+# below mirrors the spec's fixture 045 case shape (attempt 0
+# writes + fails, attempt 1 asserts marker absent + writes +
+# succeeds, downstream reads successful attempt's marker).
+# Companion test verifies the same discard discipline on
+# terminal failure (all retries exhausted).
+
+
+class _RetryTransient(Exception):
+    """Carries a transient category so the default classifier
+    treats it as retryable. Matches the ``provider_rate_limit``
+    category used in ``tests/unit/test_middleware.py``."""
+
+    category = "provider_rate_limit"
+
+
+async def test_per_attempt_scoping_under_retry_discards_failed_attempt_writes() -> None:
+    from openarmature.graph.middleware import RetryMiddleware
+
+    captured_attempt_1_read: dict[str, Any] = {}
+    captured_downstream_read: dict[str, Any] = {}
+    attempts: list[int] = []
+
+    async def _retried(_s: _SimpleState) -> dict[str, Any]:
+        attempt_n = len(attempts)
+        attempts.append(attempt_n)
+        if attempt_n == 0:
+            # First attempt: write a marker, then raise transient.
+            set_invocation_metadata(attempt_marker="first")
+            raise _RetryTransient()
+        # Second attempt: read first — assert the failed-attempt's
+        # marker is NOT visible — then write a new marker and succeed.
+        captured_attempt_1_read.update(dict(get_invocation_metadata()))
+        set_invocation_metadata(attempt_marker="second")
+        return {"counter": 1}
+
+    async def _downstream(_s: _SimpleState) -> dict[str, Any]:
+        captured_downstream_read.update(dict(get_invocation_metadata()))
+        return {"counter": 2}
+
+    graph = (
+        GraphBuilder(_SimpleState)
+        .add_node(
+            "retried",
+            _retried,
+            middleware=[RetryMiddleware(max_attempts=2, backoff=lambda _i: 0.0)],
+        )
+        .add_node("downstream", _downstream)
+        .add_edge("retried", "downstream")
+        .add_edge("downstream", END)
+        .set_entry("retried")
+        .compile()
+    )
+    await graph.invoke(_SimpleState(), metadata={"tenantId": "T1"})
+
+    assert attempts == [0, 1]
+    # Attempt 1's read: baseline only — attempt 0's transient
+    # ``attempt_marker=first`` write was discarded on failure.
+    assert captured_attempt_1_read == {"tenantId": "T1"}
+    # Downstream node: baseline + the successful attempt's write
+    # persists past the retry boundary.
+    assert captured_downstream_read == {"tenantId": "T1", "attempt_marker": "second"}
+
+
+async def test_terminal_failure_discards_final_failed_attempt_writes() -> None:
+    # Exercises the middleware directly via ``compose_chain`` so the
+    # post-retry metadata view is readable in the test scope (the
+    # engine's outer invoke() reset would otherwise pop the var back
+    # to empty before control returns to the test, masking the
+    # middleware's own discard). The contract pinned here is that
+    # AFTER the retry middleware re-raises a terminal failure, the
+    # metadata ContextVar is back at the pre-attempt baseline — no
+    # leak of the final failed attempt's writes.
+    from openarmature.graph.middleware import RetryMiddleware, compose_chain
+    from openarmature.observability.metadata import (
+        _reset_invocation_metadata,
+        _set_invocation_metadata,
+        validate_invocation_metadata,
+    )
+
+    attempts: list[int] = []
+
+    async def _always_fails(_state: Any) -> Mapping[str, Any]:
+        attempts.append(len(attempts))
+        set_invocation_metadata(attempt_marker=f"attempt_{len(attempts) - 1}")
+        raise _RetryTransient()
+
+    retry = RetryMiddleware(max_attempts=2, backoff=lambda _i: 0.0)
+    chain = compose_chain([retry], _always_fails)
+
+    # Establish a baseline outside the middleware so we can read it
+    # back post-failure. Mirrors how the engine sets the baseline
+    # at the invoke() boundary.
+    baseline_token = _set_invocation_metadata(validate_invocation_metadata({"tenantId": "T1"}))
+    try:
+        with pytest.raises(_RetryTransient):
+            await chain(_SimpleState())
+        # Both attempts ran.
+        assert attempts == [0, 1]
+        # Post-failure view: the pre-attempt baseline, with NO
+        # ``attempt_marker`` leaked from the final failed attempt.
+        assert dict(get_invocation_metadata()) == {"tenantId": "T1"}
+    finally:
+        _reset_invocation_metadata(baseline_token)
+
+
+async def test_cancellation_discards_in_flight_attempt_writes() -> None:
+    # Spec §3.4: failed-attempt metadata writes are discarded along
+    # with the attempt. When ``CancelledError`` (or any other
+    # ``BaseException``) ends the attempt, the same discard discipline
+    # applies — cancellation IS a failed attempt from the
+    # metadata-scoping perspective. Spec §6.1: cancellation MUST
+    # propagate (no retry, no swallow), so the reset must happen IN
+    # ADDITION to, not instead of, propagating ``CancelledError``.
+    from openarmature.graph.middleware import RetryMiddleware, compose_chain
+    from openarmature.observability.metadata import (
+        _reset_invocation_metadata,
+        _set_invocation_metadata,
+        validate_invocation_metadata,
+    )
+
+    attempts: list[int] = []
+
+    async def _writes_then_cancels(_state: Any) -> Mapping[str, Any]:
+        attempts.append(len(attempts))
+        set_invocation_metadata(attempt_marker="leaked")
+        raise asyncio.CancelledError("aborted")
+
+    retry = RetryMiddleware(max_attempts=3, backoff=lambda _i: 0.0)
+    chain = compose_chain([retry], _writes_then_cancels)
+
+    baseline_token = _set_invocation_metadata(validate_invocation_metadata({"tenantId": "T1"}))
+    try:
+        with pytest.raises(asyncio.CancelledError):
+            await chain(_SimpleState())
+        # Cancellation propagated — exactly ONE attempt ran (retry
+        # MUST NOT swallow ``CancelledError`` per spec §6.1).
+        assert attempts == [0]
+        # The cancelled attempt's metadata write was discarded per
+        # §3.4 — post-failure view is the pre-attempt baseline.
+        assert dict(get_invocation_metadata()) == {"tenantId": "T1"}
+    finally:
+        _reset_invocation_metadata(baseline_token)