diff --git a/CHANGELOG.md b/CHANGELOG.md index 0c57cac3..827959e5 100644 --- a/CHANGELOG.md +++ b/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"`. diff --git a/conformance.toml b/conformance.toml index 111e6071..61264263 100644 --- a/conformance.toml +++ b/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 diff --git a/docs/concepts/observability.md b/docs/concepts/observability.md index 2a0b1b99..56c5e6b6 100644 --- a/docs/concepts/observability.md +++ b/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 diff --git a/src/openarmature/graph/middleware/retry.py b/src/openarmature/graph/middleware/retry.py index 753b7f37..f5463923 100644 --- a/src/openarmature/graph/middleware/retry.py +++ b/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__ = [ diff --git a/tests/unit/test_observability_metadata.py b/tests/unit/test_observability_metadata.py index 28f6f225..90bc8336 100644 --- a/tests/unit/test_observability_metadata.py +++ b/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)