diff --git a/.github/codeql/codeql-config.yml b/.github/codeql/codeql-config.yml new file mode 100644 index 00000000..6747e292 --- /dev/null +++ b/.github/codeql/codeql-config.yml @@ -0,0 +1,50 @@ +name: "openarmature-python CodeQL config" + +# CodeQL's Python ``code-quality`` suite includes +# ``py/ineffectual-statement``, which produces a fixed false-positive +# stream against this codebase: +# +# - PEP 544 ``Protocol`` method bodies use ``...`` (the canonical +# Python idiom — never executed because Protocols are structural, +# not inheritance-based). Replacing with ``raise NotImplementedError`` +# would imply ABC semantics that the protocol classes +# intentionally don't have. Cited sites: +# - ``llm/provider.py`` (Provider.ready / complete) +# - ``checkpoint/protocol.py`` (Checkpointer.save / load / list / +# delete) +# - ``graph/middleware/_core.py`` (Middleware.__call__ + chain +# callable) +# - ``graph/observer.py`` (Observer.__call__) +# - ``await worker`` after a queue sentinel in +# ``tests/unit/test_observer.py`` is flagged as no-effect, but +# ``await`` on a Task waits for completion — the entire reason +# the line exists. CodeQL bug for async patterns. +# +# Each finding has been individually triaged across PR review rounds; +# the rule has produced zero true positives. Suppressing the rule +# repo-wide stops the regenerating noise without leaving real +# unused-statement bugs un-caught (pyright's ``reportUnusedExpression`` +# already covers the genuine cases at type-check time). +query-filters: + - exclude: + id: py/ineffectual-statement + # ``py/unused-import`` produces false positives on three patterns + # this codebase relies on: + # + # - Forward-reference casts: ``cast("FinishReason", x)`` / + # ``cast("Checkpointer", capturing)``. CodeQL doesn't look + # inside the string argument; pyright's strict mode does, AND + # raises ``reportUndefinedVariable`` if the name isn't in scope + # (verified empirically — removing ``FinishReason`` from + # ``openai.py``'s import yields the pyright error). + # - Subscripted base classes: ``class _TracingFanOutNode( + # FanOutNode[State, State]):``. The base class IS used; the + # generic subscription happens at class-definition time. + # - Re-exports for the public API surface (some submodule + # ``__init__`` files import names solely to re-expose). + # + # Pyright's ``reportUnusedImport`` (default in strict mode) catches + # the genuine unused-import cases without these false positives, + # so dropping the CodeQL rule doesn't lose signal. + - exclude: + id: py/unused-import diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 744a57ca..d1856183 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -36,7 +36,14 @@ jobs: # version developers run locally. - name: Sync deps - run: uv sync --frozen + # ``--all-extras`` installs every optional-dependency group + # (currently ``[otel]``) so pyright type-checks the + # OTel-using modules with real types and tests covering + # extras-gated code paths run end-to-end. Without this, + # ``opentelemetry.*`` symbols come back Unknown and pyright + # produces hundreds of false-positive errors on + # ``tests/unit/test_observability_otel.py``. + run: uv sync --frozen --all-extras - name: Lint (ruff check) run: uv run ruff check . diff --git a/.github/workflows/codeql.yml b/.github/workflows/codeql.yml new file mode 100644 index 00000000..fba94227 --- /dev/null +++ b/.github/workflows/codeql.yml @@ -0,0 +1,61 @@ +# CodeQL advanced setup — replaces the prior default setup so we can +# reference ``.github/codeql/codeql-config.yml`` (default setup's UI +# doesn't expose a config-file field). Config excludes +# ``py/ineffectual-statement`` which produced a regenerating stream +# of false positives against PEP 544 Protocol method bodies; see the +# config file for the full rationale. +name: CodeQL + +on: + push: + branches: [main] + pull_request: + branches: [main] + schedule: + # Weekly scan to surface new issues that didn't change in any + # PR. Matches default setup's cadence. + - cron: "25 16 * * 5" + +# Least-privilege per the CI workflow next door. CodeQL needs: +# - ``actions: read`` to fetch action metadata. +# - ``contents: read`` to check out code. +# - ``security-events: write`` to upload SARIF results to the Security +# tab. +permissions: + actions: read + contents: read + security-events: write + +concurrency: + group: ${{ github.workflow }}-${{ github.ref }} + cancel-in-progress: true + +jobs: + analyze: + name: Analyze (${{ matrix.language }}) + runs-on: ubuntu-latest + strategy: + fail-fast: false + matrix: + language: [actions, python] + steps: + - name: Checkout + uses: actions/checkout@v6 + + - name: Initialize CodeQL + uses: github/codeql-action/init@v3 + with: + languages: ${{ matrix.language }} + # Reference the in-tree config so the rule-suppressions land. + config-file: ./.github/codeql/codeql-config.yml + # Match the prior default setup's query coverage by including + # the security-and-quality suite explicitly. Code-quality + # findings still surface; only the per-rule + # ``py/ineffectual-statement`` exclusion in the config file + # is suppressed. + queries: security-and-quality + + - name: Perform CodeQL Analysis + uses: github/codeql-action/analyze@v3 + with: + category: "/language:${{ matrix.language }}" diff --git a/docs/RELEASING.md b/docs/RELEASING.md index adbf3db1..35d61a07 100644 --- a/docs/RELEASING.md +++ b/docs/RELEASING.md @@ -43,9 +43,9 @@ vim src/openarmature/__init__.py # __version__ = "0.5.0rc1" vim tests/test_smoke.py # match # uv.lock auto-updates via pre-commit when you stage pyproject.toml git add pyproject.toml src/openarmature/__init__.py tests/test_smoke.py uv.lock -git commit -m "release: prep 0.5.0-rc1" +git commit -m "release: prep 0.5.0rc1" git push origin -gh pr create --title "release: prep 0.5.0-rc1" ... +gh pr create --title "release: prep 0.5.0rc1" ... ``` If this is the same release cycle as a previous RC, also bump the spec pins @@ -56,6 +56,14 @@ Merge the PR. ### 2. Tag the RC +Two version forms appear in this guide and they're not interchangeable +in writing even though PEP 440 normalizes them: `pyproject.toml` / +`__init__.py` use `0.5.0rc1` (no hyphen — the canonical PEP 440 +form); git tags + the release-workflow's tag-shape regex use +`v0.5.0-rc1` (with hyphen). Mixing them in commit messages or PR +titles muddies the audit trail; stick to the no-hyphen form for +version-string fields and the hyphenated form for git tags. + ```bash git checkout main && git pull --ff-only git tag v0.5.0-rc1 diff --git a/pyproject.toml b/pyproject.toml index 7741f483..629baffc 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -16,6 +16,21 @@ dependencies = [ "httpx>=0.27", ] +[project.optional-dependencies] +# Spec observability §6 OTel mapping. Optional per charter §3.1 +# principle 5 — backend mappings are pluggable rather than core +# dependencies. Plan to lift into a sibling ``openarmature-otel`` +# package at v1.0 launch alongside ``openarmature-eval``. +otel = [ + # Upper bound guards against the SDK 2.x release that removes + # ``opentelemetry.sdk._logs.LoggingHandler`` (currently emits a + # DeprecationWarning). Migration to + # ``opentelemetry-instrumentation-logging`` lands in Phase 6.1 + # before bumping the upper bound. + "opentelemetry-api>=1.27,<2", + "opentelemetry-sdk>=1.27,<2", +] + [project.urls] Repository = "https://github.com/LunarCommand/openarmature-python" Specification = "https://github.com/LunarCommand/openarmature-spec" diff --git a/src/openarmature/graph/compiled.py b/src/openarmature/graph/compiled.py index a3c65abf..44d7463e 100644 --- a/src/openarmature/graph/compiled.py +++ b/src/openarmature/graph/compiled.py @@ -41,6 +41,16 @@ CheckpointRecord, NodePosition, ) +from openarmature.observability.correlation import ( + _reset_active_dispatch, + _reset_active_observers, + _reset_correlation_id, + _reset_invocation_id, + _set_active_dispatch, + _set_active_observers, + _set_correlation_id, + _set_invocation_id, +) from .edges import END, ConditionalEdge, EndSentinel, StaticEdge from .errors import ( @@ -369,6 +379,18 @@ async def invoke( pending_resume_states=pending_resume_states, resume_invocation=resume_invocation, ) + # Spec observability §3.1: the correlation_id MUST be readable + # from anywhere within the invocation's async call tree via the + # language's idiomatic context primitive. Set the ContextVar + # BEFORE creating the delivery worker so the worker's captured + # context sees the correlation_id (asyncio.create_task snapshots + # the current Context at creation time). Reset on return so + # subsequent invocations get a fresh slate. Nested ``invoke()`` + # calls (subgraph-as-node uses ``_invoke`` directly, not the + # public ``invoke``, so they don't re-set; see §3.1's + # "per-invocation is OUTERMOST invoke" wording). + correlation_token = _set_correlation_id(resolved_correlation_id) + invocation_token = _set_invocation_id(invocation_id) worker = asyncio.create_task(deliver_loop(queue)) self._active_workers.add(worker) # Auto-prune: when the worker completes (after the sentinel is @@ -378,6 +400,8 @@ async def invoke( try: return await self._invoke(starting_state, context) finally: + _reset_invocation_id(invocation_token) + _reset_correlation_id(correlation_token) # Sentinel terminates the worker after it processes events # already on the queue (including any error event we just # dispatched on the failure path). Drain semantics live on @@ -585,14 +609,27 @@ async def innermost(s: Any) -> Mapping[str, Any]: innermost, ) + # Spec observability §3 / Phase 6 LLM-span hook: capability + # backends emitting from inside a node body (the + # llm-provider span instrumentation in OpenAIProvider) need + # to find the observers active for THIS invocation. Set the + # ContextVar around the chain invocation; reset in + # ``try/finally`` so an exception escaping the chain still + # restores the prior value. + observers_token = _set_active_observers(context.full_observers()) + dispatch_token = _set_active_dispatch(lambda event: _dispatch(context, event)) try: - final_partial = await chain(state) - except RuntimeGraphError: - raise - except Exception as e: - # A raw exception (node-raised or middleware-raised) escaped - # the chain unrecovered. Wrap as NodeException per §4. - raise NodeException(node_name=current, cause=e, recoverable_state=state) from e + try: + final_partial = await chain(state) + except RuntimeGraphError: + raise + except Exception as e: + # A raw exception (node-raised or middleware-raised) escaped + # the chain unrecovered. Wrap as NodeException per §4. + raise NodeException(node_name=current, cause=e, recoverable_state=state) from e + finally: + _reset_active_dispatch(dispatch_token) + _reset_active_observers(observers_token) # Engine's canonical merge uses the ORIGINAL state per §2: "the # transformed state is passed to ``next``, NOT to the engine's # merge step." If middleware transformed state mid-chain, the @@ -649,18 +686,29 @@ async def innermost(s: Any) -> Mapping[str, Any]: list(self.middleware) + list(node.middleware), innermost, ) + # Same active-observers scope as _step_function_node — parent + # middleware running before the descent should see the parent's + # observer set; the inner _invoke (called via ``node.run``) + # descends into its own context and sets a new scope from + # there. + observers_token = _set_active_observers(context.full_observers()) + dispatch_token = _set_active_dispatch(lambda event: _dispatch(context, event)) try: - final_partial = await chain(state) - except RuntimeGraphError: - raise - except Exception as e: - # Same wrap as _step_function_node: a raw exception escaping - # the parent's middleware chain (e.g., a middleware bug or a - # projection error) becomes NodeException tagged with the - # SubgraphNode's wrapper name so §4 recoverable_state is - # preserved. - raise NodeException(node_name=current, cause=e, recoverable_state=state) from e + try: + final_partial = await chain(state) + except RuntimeGraphError: + raise + except Exception as e: + # Same wrap as _step_function_node: a raw exception escaping + # the parent's middleware chain (e.g., a middleware bug or a + # projection error) becomes NodeException tagged with the + # SubgraphNode's wrapper name so §4 recoverable_state is + # preserved. + raise NodeException(node_name=current, cause=e, recoverable_state=state) from e + finally: + _reset_active_dispatch(dispatch_token) + _reset_active_observers(observers_token) return _merge_partial(state, final_partial, self.reducers, current) async def _step_fan_out_node( @@ -740,12 +788,23 @@ async def innermost(s: Any) -> Mapping[str, Any]: innermost, ) + # Same observability §3 / LLM-span hook contract as + # _step_function_node: set the active observer set in scope + # around the chain invocation so capability backends emitting + # from inside the fan-out's parent dispatch (or any code + # running on its call stack) can find the observers. + observers_token = _set_active_observers(context.full_observers()) + dispatch_token = _set_active_dispatch(lambda event: _dispatch(context, event)) try: - final_partial = await chain(state) - except RuntimeGraphError: - raise - except Exception as e: - raise NodeException(node_name=current, cause=e, recoverable_state=state) from e + try: + final_partial = await chain(state) + except RuntimeGraphError: + raise + except Exception as e: + raise NodeException(node_name=current, cause=e, recoverable_state=state) from e + finally: + _reset_active_dispatch(dispatch_token) + _reset_active_observers(observers_token) merged_outer = _merge_partial(state, final_partial, self.reducers, current) # Spec §10.3 + §10.7: the fan-out's own completion DOES save — # one record once the fan-out as a whole has finished and diff --git a/src/openarmature/llm/providers/openai.py b/src/openarmature/llm/providers/openai.py index fc4bad05..a682603f 100644 --- a/src/openarmature/llm/providers/openai.py +++ b/src/openarmature/llm/providers/openai.py @@ -39,11 +39,15 @@ import json from collections.abc import Sequence -from typing import Any, cast +from typing import Any, Literal, cast import httpx from pydantic import ValidationError +from openarmature.graph.events import NodeEvent +from openarmature.graph.state import State +from openarmature.observability.correlation import current_dispatch + from ..errors import ( ProviderAuthentication, ProviderInvalidModel, @@ -190,6 +194,41 @@ async def complete( validate_tools(tools) body = self._build_request_body(messages, tools, config) + # Spec observability §5.5 LLM provider span: when an + # observability backend is active in the current invocation, + # emit a started/completed event pair around the wire call so + # the backend can build a span. Queue-mediated dispatch + # preserves spec §6 serial event ordering across all event + # sources within an invocation. ``current_dispatch()`` returns + # ``None`` outside an openarmature invocation (direct + # provider use in scripts/tests), in which case the call + # proceeds without span emission. + dispatch = current_dispatch() + if dispatch is not None: + dispatch(_make_llm_event("started", model=self.model)) + + try: + response = await self._do_complete(body) + except Exception as exc: + if dispatch is not None: + dispatch(_make_llm_event("completed", model=self.model, error=exc)) + raise + + if dispatch is not None: + dispatch( + _make_llm_event( + "completed", + model=self.model, + finish_reason=response.finish_reason, + usage=response.usage, + ) + ) + return response + + async def _do_complete(self, body: dict[str, Any]) -> Response: + """Wire-call helper: separated from ``complete()`` so the + LLM-provider span hook in ``complete()`` can wrap success and + failure paths uniformly.""" try: resp = await self._client.post("/v1/chat/completions", json=body) except httpx.HTTPError as exc: @@ -481,6 +520,87 @@ def _looks_like_model_not_loaded(message: object) -> bool: return "not loaded" in lower or "loading" in lower +# --------------------------------------------------------------------------- +# Observability §5.5 LLM provider span event helpers +# --------------------------------------------------------------------------- + + +class _LlmEventState(State): + """Typed payload for LLM-provider span events. Subclasses + :class:`openarmature.graph.state.State` so the + ``NodeEvent.pre_state: State`` contract holds — observers + calling ``event.pre_state.model_dump()`` (or any other + Pydantic-on-State method) work without the raw-dict overload + that previously violated the schema. + + Backend mappings (the OTel observer in this repo, future + Langfuse / Datadog adapters) recognize the + ``("openarmature.llm.complete",)`` namespace sentinel and read + these fields directly via attribute access. + """ + + model: str + finish_reason: str | None = None + prompt_tokens: int | None = None + completion_tokens: int | None = None + total_tokens: int | None = None + # On error responses the provider caller doesn't have a + # graph-engine §4 ``RuntimeGraphError`` to put in + # ``NodeEvent.error``, so we surface the failure detail through + # these fields instead. ``error_category`` is the canonical §7 + # llm-provider category (``provider_unavailable``, etc.) when + # the failed exception carries one. + error_type: str | None = None + error_message: str | None = None + error_category: str | None = None + + +def _make_llm_event( + phase: Literal["started", "completed"], + *, + model: str, + finish_reason: FinishReason | None = None, + usage: Usage | None = None, + error: BaseException | None = None, +) -> NodeEvent: + """Build a NodeEvent-shaped record for the engine's delivery + queue. The OTel observer (or any backend mapping) recognises the + sentinel ``node_name`` and ``namespace`` and emits an LLM-specific + span instead of a node span. Backend-specific attribute extraction + reads ``model``, ``finish_reason``, and ``usage`` from + ``pre_state`` directly via attribute access. + """ + error_type: str | None = None + error_message: str | None = None + error_category: str | None = None + if error is not None: + error_type = type(error).__name__ + error_message = str(error) + category = getattr(error, "category", None) + if isinstance(category, str): + error_category = category + payload = _LlmEventState( + model=model, + finish_reason=finish_reason, + prompt_tokens=usage.prompt_tokens if usage is not None else None, + completion_tokens=usage.completion_tokens if usage is not None else None, + total_tokens=usage.total_tokens if usage is not None else None, + error_type=error_type, + error_message=error_message, + error_category=error_category, + ) + return NodeEvent( + node_name="openarmature.llm.complete", + namespace=("openarmature.llm.complete",), + step=-1, + phase=phase, + pre_state=payload, + post_state=None, + error=None, + parent_states=(), + ) + + __all__ = [ "OpenAIProvider", ] diff --git a/src/openarmature/observability/__init__.py b/src/openarmature/observability/__init__.py new file mode 100644 index 00000000..5d22f2f8 --- /dev/null +++ b/src/openarmature/observability/__init__.py @@ -0,0 +1,35 @@ +"""openarmature.observability — cross-backend observability surface. + +Two layers: + +- **Core** (this module + ``correlation.py``): always available, no + extra dependencies. Exposes :func:`current_correlation_id` and + :func:`current_active_observers` — the spec observability §3 + ``ContextVar`` primitives that every backend mapping consumes. +- **Backend mappings** (under ``observability.otel`` and future + ``observability.langfuse`` etc.): gated behind optional + dependencies (``pip install openarmature[otel]``). Importing the + subpackage without the extras installed raises an informative + ``ImportError`` pointing the caller at the install command. + +The split mirrors charter §3.1 principle 5: core defines the +contracts; specific backends implement them. At v1.0 launch the +backend mappings will lift into sibling packages +(``openarmature-otel``, ``openarmature-langfuse``) — until then +they live here under per-backend subpackages so the layering is +established up front. +""" + +from .correlation import ( + current_active_observers, + current_correlation_id, + current_dispatch, + current_invocation_id, +) + +__all__ = [ + "current_active_observers", + "current_correlation_id", + "current_dispatch", + "current_invocation_id", +] diff --git a/src/openarmature/observability/correlation.py b/src/openarmature/observability/correlation.py new file mode 100644 index 00000000..45f87091 --- /dev/null +++ b/src/openarmature/observability/correlation.py @@ -0,0 +1,226 @@ +"""Cross-backend correlation primitives (spec observability §3). + +Two ``ContextVar``-backed primitives that any observability backend +mapping (OTel here, Langfuse / Datadog / custom in the future) consumes +through a uniform user-readable surface: + +- :data:`current_correlation_id` — the per-invocation cross-backend + join key. Set on every outermost ``invoke()`` call (caller-supplied + or auto-generated UUIDv4 per spec §3.1) and reset on return. User + code in node bodies, middleware, and observers reads it via + :func:`current_correlation_id`. +- :data:`_active_observers` — the observer set in scope for any code + running INSIDE a node body. Read by capability backends that need + to emit observer events from outside the engine's per-step machinery + (e.g., the llm-provider span hook puts a NodeEvent-shaped record on + the engine's delivery queue, then those observers receive it). The + engine sets this around each ``chain(state)`` invocation via + ``try/finally`` so reset is guaranteed even on exception. + +These primitives live in the core package — no OpenTelemetry +dependency — because the spec §3.1 contract ("MUST propagate via the +language's idiomatic context primitive — Python ``ContextVar``") is +backend-agnostic. The OTel-specific surfacing lives under +``openarmature.observability.otel`` and is gated behind the +``[otel]`` extras. +""" + +from __future__ import annotations + +from collections.abc import Callable +from contextvars import ContextVar, Token +from typing import TYPE_CHECKING + +if TYPE_CHECKING: + from openarmature.graph.events import NodeEvent + from openarmature.graph.observer import SubscribedObserver + + +# --------------------------------------------------------------------------- +# Correlation ID — spec observability §3.1 +# --------------------------------------------------------------------------- + + +_correlation_id_var: ContextVar[str | None] = ContextVar("openarmature.correlation_id", default=None) + + +def current_correlation_id() -> str | None: + """Return the correlation ID for the current invocation, or + ``None`` if no openarmature invocation is in scope. + + Per spec §3.1 the correlation ID MUST be readable from anywhere + within an invocation's async call tree — node bodies, middleware, + observers — without explicit threading through function arguments. + This is the public reader. + + Returns ``None`` outside an invocation (e.g., at module import + time, inside a test that runs without going through ``invoke()``). + Callers MUST handle the ``None`` case rather than asserting a + string is always present. + """ + return _correlation_id_var.get() + + +def _set_correlation_id(value: str) -> Token[str | None]: + """Set the correlation ID for the current invocation. Internal — + callers OUTSIDE the engine should not touch this; the engine + paves the lifecycle in ``CompiledGraph.invoke``. + + Returns the ``Token`` the caller MUST hand back to + :func:`_reset_correlation_id` so the prior value is restored + cleanly under nesting. Use ``try/finally``.""" + return _correlation_id_var.set(value) + + +def _reset_correlation_id(token: Token[str | None]) -> None: + _correlation_id_var.reset(token) + + +# --------------------------------------------------------------------------- +# Invocation ID — spec observability §5.1 +# +# The framework-generated UUIDv4 that ties spans of one invocation +# together within a single backend. Distinct from ``correlation_id`` +# (which is the cross-backend join key, caller-supplied or auto- +# generated). Engine sets this ContextVar in ``invoke()`` BEFORE the +# delivery worker is created so the worker's captured context sees +# the right value. +# --------------------------------------------------------------------------- + + +_invocation_id_var: ContextVar[str | None] = ContextVar("openarmature.invocation_id", default=None) + + +def current_invocation_id() -> str | None: + """Return the engine-minted invocation ID for the current + invocation, or ``None`` if no openarmature invocation is in + scope. + + Per spec observability §5.1 every invocation produces a unique + UUIDv4 ``invocation_id``, framework-generated, surfaced as the + ``openarmature.invocation_id`` attribute on the invocation span + + on every per-backend record. This is the public reader for + backend mappings (OTel, future Langfuse) that need to populate + that attribute. + """ + return _invocation_id_var.get() + + +def _set_invocation_id(value: str) -> Token[str | None]: + """Set the invocation ID for the current invocation. Internal — + engine-only.""" + return _invocation_id_var.set(value) + + +def _reset_invocation_id(token: Token[str | None]) -> None: + _invocation_id_var.reset(token) + + +# --------------------------------------------------------------------------- +# Active observer set — for capability backends emitting from outside the +# engine's per-step path (llm-provider span hook in Phase 6, future +# Langfuse/Datadog backends, user-written instrumented capabilities). +# --------------------------------------------------------------------------- + + +_active_observers_var: ContextVar[tuple[SubscribedObserver, ...]] = ContextVar( + "openarmature.active_observers", default=() +) + + +def current_active_observers() -> tuple[SubscribedObserver, ...]: + """Return the observer tuple in scope for the current node body + (or empty tuple outside any invocation). + + Capability code that needs to emit observer events from outside + the engine's per-step machinery (e.g., the llm-provider span hook + inside ``OpenAIProvider.complete``) reads this to find which + observers should receive the event. Combined with the engine's + delivery queue, this preserves spec §6's strict serial ordering + across all event sources within an invocation. + + Returns an empty tuple when no invocation is active, by design — + callers can iterate without a None check. + """ + return _active_observers_var.get() + + +def _set_active_observers( + observers: tuple[SubscribedObserver, ...], +) -> Token[tuple[SubscribedObserver, ...]]: + """Set the observer tuple in scope. Internal — engine-only. + Returns a Token to hand to :func:`_reset_active_observers`.""" + return _active_observers_var.set(observers) + + +def _reset_active_observers(token: Token[tuple[SubscribedObserver, ...]]) -> None: + _active_observers_var.reset(token) + + +# --------------------------------------------------------------------------- +# Active dispatch hook — queue-mediated event emission from outside the +# engine's per-step path. The engine sets this ContextVar to a closure +# over the current invocation's delivery queue + observer chain; +# capability backends (the LLM provider span hook in Phase 6, future +# Langfuse/Datadog instrumentations) call ``current_dispatch()(event)`` +# to enqueue an event for the same delivery worker the engine uses. +# +# Spec §6 mandates strictly serial per-invocation event delivery. By +# routing capability events through the same queue (rather than calling +# observers directly), the engine's ordering guarantees extend +# automatically to LLM events, future backend events, etc. without each +# capability re-deriving the locking story. +# --------------------------------------------------------------------------- + + +_active_dispatch_var: ContextVar[Callable[[NodeEvent], None] | None] = ContextVar( + "openarmature.active_dispatch", default=None +) + + +def current_dispatch() -> Callable[[NodeEvent], None] | None: + """Return the engine's dispatch callable for the current invocation, + or ``None`` outside any invocation. + + Capability code emitting observer events from inside a node body + calls this to put a ``NodeEvent``-shaped record on the engine's + delivery queue. The queue's serial worker preserves spec §6's + per-invocation event ordering across all event sources (engine, + checkpoint, LLM provider, future backends). + """ + return _active_dispatch_var.get() + + +def _set_active_dispatch( + dispatch: Callable[[NodeEvent], None], +) -> Token[Callable[[NodeEvent], None] | None]: + """Set the engine's dispatch callable in scope. Internal — + engine-only.""" + return _active_dispatch_var.set(dispatch) + + +def _reset_active_dispatch( + token: Token[Callable[[NodeEvent], None] | None], +) -> None: + _active_dispatch_var.reset(token) + + +__all__ = [ + # Public surface — readable from anywhere within an invocation. + "current_active_observers", + "current_correlation_id", + "current_dispatch", + "current_invocation_id", + # Engine-internal lifecycle helpers — exported so the engine in + # ``openarmature.graph.compiled`` can drive set/reset without + # pyright's strict ``reportUnusedFunction`` flagging them as + # dead. Underscore-prefixed; not part of the user-facing API. + "_reset_active_dispatch", + "_reset_active_observers", + "_reset_correlation_id", + "_reset_invocation_id", + "_set_active_dispatch", + "_set_active_observers", + "_set_correlation_id", + "_set_invocation_id", +] diff --git a/src/openarmature/observability/otel/__init__.py b/src/openarmature/observability/otel/__init__.py new file mode 100644 index 00000000..e29aeb06 --- /dev/null +++ b/src/openarmature/observability/otel/__init__.py @@ -0,0 +1,44 @@ +"""OpenTelemetry backend mapping for openarmature observability. + +Per charter §3.1 principle 5: the core defines observability contracts +(``correlation_id`` ContextVar, dispatch primitives); specific +backends implement them. This subpackage is extras-gated — install +with ``pip install openarmature[otel]`` to bring in +``opentelemetry-api`` and ``opentelemetry-sdk``. + +Importing this subpackage without the extras installed raises an +informative :class:`ImportError` pointing the caller at the install +command. We do NOT do partial fallbacks (e.g., a stubbed observer +that silently no-ops) — the user opted in to OTel by importing +``openarmature.observability.otel``, so a clean failure on missing +deps is preferable to silent-broken behavior. + +Public surface: + +- :class:`OTelObserver` — the observer-driven span lifecycle implementation + per spec observability §6 RECOMMENDED path. +- :func:`install_log_bridge` — helper to wire the OTel Logs SDK to + the stdlib ``logging`` root with ``correlation_id`` injection. + +Plan to lift into a sibling ``openarmature-otel`` package at the v1.0 +launch alongside ``openarmature-eval``; until then this subpackage +holds the implementation in-tree. +""" + +from __future__ import annotations + +try: + from .logs import install_log_bridge + from .observer import OTelObserver +except ImportError as exc: # pragma: no cover - exercised by extras-not-installed path + if "opentelemetry" in str(exc): + raise ImportError( + "openarmature.observability.otel requires the optional `otel` extras. " + "Install with: pip install 'openarmature[otel]'" + ) from exc + raise + +__all__ = [ + "OTelObserver", + "install_log_bridge", +] diff --git a/src/openarmature/observability/otel/logs.py b/src/openarmature/observability/otel/logs.py new file mode 100644 index 00000000..2422edcf --- /dev/null +++ b/src/openarmature/observability/otel/logs.py @@ -0,0 +1,99 @@ +"""OTel Logs Bridge integration (spec observability §7). + +Provides :func:`install_log_bridge` — an opt-in helper that wires +the stdlib :mod:`logging` root logger through the OTel Logs SDK so +every log record emitted within an invocation carries the active +``trace_id``/``span_id`` plus ``openarmature.correlation_id``. + +Opt-in by design: users may have their own logging configuration we +shouldn't override silently. Calling ``install_log_bridge(provider)`` +explicitly attaches an OTel ``LoggingHandler`` to the root logger +and registers a filter that injects the correlation_id from the +ContextVar. +""" + +from __future__ import annotations + +import logging +from typing import TYPE_CHECKING + +if TYPE_CHECKING: + from opentelemetry.sdk._logs import LoggerProvider + + +class _CorrelationIdFilter(logging.Filter): + """Logging filter that reads the openarmature correlation_id + ContextVar and attaches it to every log record as the + ``openarmature.correlation_id`` attribute. Per spec §7 the + attribute MUST appear on every log record emitted during an + invocation.""" + + def filter(self, record: logging.LogRecord) -> bool: + from openarmature.observability.correlation import current_correlation_id + + cid = current_correlation_id() + if cid is not None: + # Stored on the log record so any formatter/handler that + # reads ``record.__dict__`` (including the OTel + # LoggingHandler) sees it. + setattr(record, "openarmature.correlation_id", cid) + return True + + +def install_log_bridge( + provider: LoggerProvider, + *, + level: int = logging.NOTSET, +) -> None: + """Wire the stdlib root logger to the supplied OTel + :class:`LoggerProvider`. Adds a + :class:`opentelemetry.sdk._logs.LoggingHandler` for OTel-native + ``trace_id`` / ``span_id`` bridging, AND attaches an + :class:`_CorrelationIdFilter` directly to the ROOT LOGGER (not + the handler) so the ``openarmature.correlation_id`` attribute + lands on every log record emitted during an invocation — + including records consumed by pre-existing stdout / file / + third-party handlers the user already had configured. + + Filter-on-the-root-logger placement matters per spec §7: + "log records emitted from anywhere within an invocation MUST + carry ``openarmature.correlation_id``." A handler-level filter + would only modify records flowing through THAT handler, so a + user's existing stdout handler would see records without the + attribute. The root-logger filter applies to every record, + regardless of which handler eventually processes it. + + Idempotent: re-calling is a no-op (we check for the existing + OA-tagged handler AND for an existing filter instance on the + root logger). + + The user retains responsibility for providing the + :class:`LoggerProvider` (typically built with their preferred + exporter — :class:`InMemoryLogExporter` for tests, + :class:`OTLPLogExporter` for production). + """ + from opentelemetry.sdk._logs import LoggingHandler # type: ignore[attr-defined] + + root = logging.getLogger() + # Idempotency #1: don't double-add the OTel LoggingHandler. + handler_already_installed = any( + isinstance(h, LoggingHandler) and getattr(h, "_openarmature_installed", False) for h in root.handlers + ) + if not handler_already_installed: + handler = LoggingHandler(level=level, logger_provider=provider) + # Direct assignment isn't typed on LoggingHandler; route + # through ``object.__setattr__`` to avoid pyright's strict + # attribute-access check without losing the idempotency- + # marker behavior. + object.__setattr__(handler, "_openarmature_installed", True) + root.addHandler(handler) + # Idempotency #2: don't double-add the correlation_id filter to + # the root logger. + filter_already_installed = any(isinstance(f, _CorrelationIdFilter) for f in root.filters) + if not filter_already_installed: + root.addFilter(_CorrelationIdFilter()) + + +__all__ = [ + "install_log_bridge", +] diff --git a/src/openarmature/observability/otel/observer.py b/src/openarmature/observability/otel/observer.py new file mode 100644 index 00000000..555ba8e1 --- /dev/null +++ b/src/openarmature/observability/otel/observer.py @@ -0,0 +1,794 @@ +"""OTelObserver — observer-driven span lifecycle (spec observability §6). + +The observer subscribes to all three §6 phases (``started``, +``completed``, ``checkpoint_saved``) plus the LLM-provider events the +``OpenAIProvider`` enqueues from inside node bodies. On a ``started`` +event it opens a leaf span and pushes it onto an in-flight map keyed +by ``(namespace, attempt_index, fan_out_index)``; on the matching +``completed`` event it pops the span, applies §4.2 status mapping, +and closes it. + +Subtree isolation lives in dedicated dicts rather than the leaf-span +key: + +- ``_subgraph_spans`` — synthetic subgraph dispatch spans (the + engine wrapper is transparent per fixture 013, but observability + §4.5 mandates a span). Keyed by namespace prefix. Open lazily on + the first deeper-namespace event, close when subsequent events + leave the prefix. +- ``_detached_roots`` — root spans for detached subgraphs (§4.4) + and per-instance detached fan-out roots. Each lives in its own + fresh ``trace_id``; the parent's dispatch span carries an OTel + :class:`Link` to the detached trace. +- ``_invocation_span`` — root invocation span keyed by + ``correlation_id``. Closed eagerly when a new correlation_id's + first event arrives (or explicitly via + :meth:`close_invocation` / :meth:`shutdown`). + +Spans are emitted through a **private** :class:`TracerProvider` +constructed by this observer — never the OTel global. Per spec §6 +TracerProvider isolation, registering globally would cause every +auto-instrumentation library that writes to the global provider +(OpenInference, opentelemetry-instrumentation-openai, LiteLLM, etc.) +to emit duplicate spans alongside ours. + +Detached trace mode (§4.4) is implemented by minting a fresh +:class:`SpanContext` with a new ``trace_id`` when entering a +configured-detached subgraph or fan-out; the parent's dispatch span +carries an OTel :class:`Link` to the detached trace. Inner-event +parent resolution checks the per-fan-out-instance key +(``namespace[:1] + (str(fan_out_index),)``) before the generic +prefix scan, so per-instance detached roots win without depending +on attach-then-resolve ordering. +""" + +from __future__ import annotations + +from dataclasses import dataclass, field +from typing import TYPE_CHECKING, Any, cast + +from opentelemetry import context as otel_context +from opentelemetry import trace as otel_trace +from opentelemetry.context import attach, detach +from opentelemetry.sdk.trace import SpanProcessor, TracerProvider +from opentelemetry.sdk.trace.id_generator import RandomIdGenerator +from opentelemetry.trace import ( + Link, + NonRecordingSpan, + Span, + SpanContext, + SpanKind, + Status, + StatusCode, + TraceFlags, +) +from opentelemetry.trace.propagation import set_span_in_context + +if TYPE_CHECKING: + from openarmature.graph.events import NodeEvent + + +# Span-stack key shape: ``(namespace, attempt_index, fan_out_index)`` +# — these three fields uniquely identify any node attempt within an +# invocation. Trace_id was previously included in the key but that +# created a registration/lookup mismatch when the OTel current-span +# context changed between a span's started and completed events +# (e.g., detached fan-out instances opening detached roots in +# between). Detached sub-tree spans live in ``_detached_roots`` / +# ``_subgraph_spans`` separately, so the namespace alone doesn't +# collide here. +_StackKey = tuple[tuple[str, ...], int, int | None] + + +# Sentinel namespace the LLM provider emits to signal "this is an LLM +# event, not a regular node event." +_LLM_NAMESPACE = ("openarmature.llm.complete",) + + +def _read_spec_version() -> str: + """Read the spec version pinned at package level. Lazy import + avoids a circular at module-load time (the package's ``__init__`` + imports submodules that may import the observability stack).""" + from openarmature import __spec_version__ + + return __spec_version__ + + +def _empty_str_frozenset() -> frozenset[str]: + """Typed empty frozenset factory for ``detached_subgraphs`` / + ``detached_fan_outs`` defaults. ``default_factory=frozenset`` + alone produces ``frozenset[Unknown]`` under pyright strict + mode; a named factory with the explicit return annotation + preserves the ``frozenset[str]`` typing without falling back to + a lambda.""" + return frozenset() + + +@dataclass +class _OpenSpan: + """An in-flight span paired with the OTel context token that pinned + its scope. The token is ``detach``ed when the span closes so the + OTel current-span context unwinds correctly.""" + + span: Span + token: object + + +@dataclass +class OTelObserver: + """Observer-driven OTel span lifecycle per spec observability §6. + + Construct with a :class:`SpanProcessor` (typically a + :class:`BatchSpanProcessor` wrapping a real exporter, or a + :class:`SimpleSpanProcessor` wrapping :class:`InMemorySpanExporter` + for tests). The observer instantiates its own private + :class:`TracerProvider` from the supplied processor — callers + MUST NOT pre-register the provider globally. + + Constructor knobs: + + - ``detached_subgraphs`` — set of subgraph wrapper node names + that should run in their own trace (§4.4). One detached trace + per such subgraph. + - ``detached_fan_outs`` — set of fan-out node names whose + INSTANCES each get their own trace. One detached trace per + instance. + - ``disable_llm_spans`` — when ``True`` the observer skips the + §5.5 LLM provider span. All other spans (node, subgraph, + fan-out, etc.) emit normally. Useful when an external + auto-instrumentation library (OpenInference, etc.) is the + canonical source of LLM spans. + - ``spec_version`` — string surfaced as + ``openarmature.graph.spec_version`` on the invocation span. + + **Concurrency model.** A single ``OTelObserver`` instance is + SAFE for sequential invocations on the same graph (one + ``invoke()`` followed by another, with the observer reused + between). It is NOT safe to share an instance across CONCURRENT + invocations — the internal span state (``_open_spans``, + ``_subgraph_spans``, ``_detached_roots``, ``_invocation_span``) + is keyed without per-invocation scoping, so overlapping + namespaces collide and the close-prior-correlation_id logic in + ``_handle_started`` would close another in-flight invocation's + span. Recommended pattern: one observer per ``CompiledGraph`` + instance for sequential workloads; for ASGI / batch / parallel + invocation services, attach a fresh observer per invocation + (via ``invoke(observers=[...])``) until the Phase 6.1 + correlation_id-scoped state lands. + """ + + span_processor: SpanProcessor + detached_subgraphs: frozenset[str] = field(default_factory=_empty_str_frozenset) + detached_fan_outs: frozenset[str] = field(default_factory=_empty_str_frozenset) + disable_llm_spans: bool = False + # Read from the package's ``__spec_version__`` (one of the three + # places the spec version is pinned per CLAUDE.md). Bumping the + # spec submodule + the two version fields automatically updates + # the value reported on every invocation span. + spec_version: str = field(default_factory=_read_spec_version) + + # Internal state, populated in __post_init__ and during invocation. + _provider: TracerProvider = field(init=False, repr=False) + _tracer: otel_trace.Tracer = field(init=False, repr=False) + _open_spans: dict[_StackKey, _OpenSpan] = field( + init=False, repr=False, default_factory=dict[_StackKey, _OpenSpan] + ) + # The invocation root span, opened on the first event of an + # invocation and closed when the matching outermost completed + # event arrives (or, in practice, when the engine's queue drains + # — the invocation span has no started/completed pair of its + # own, so we open it lazily and close it on a sentinel). + _invocation_span: dict[str, _OpenSpan] = field( + init=False, repr=False, default_factory=dict[str, _OpenSpan] + ) + # Synthetic subgraph dispatch spans: the engine wrapper for + # ``add_subgraph_node`` is transparent (graph-engine fixture 013 + # — no started/completed events of its own), but observability + # §4.5 mandates a subgraph span. The OTel observer synthesizes + # one by detecting deeper-namespace events and opening an + # ancestor span for each new prefix; closes when subsequent + # events leave that prefix. + _subgraph_spans: dict[tuple[str, ...], _OpenSpan] = field( + init=False, repr=False, default_factory=dict[tuple[str, ...], _OpenSpan] + ) + # Per-namespace-prefix detached trace tracking. When a detached + # subgraph or fan-out instance enters, we mint a fresh trace and + # store the root span here so subsequent inner events at that + # prefix find the right parent. Keyed by namespace prefix + # (subgraph) or namespace_prefix + (str(fan_out_index),) (fan-out + # instance). The fan-out node's own span (in the parent trace) + # collects Links to each detached instance trace. + _detached_roots: dict[tuple[str, ...], _OpenSpan] = field( + init=False, repr=False, default_factory=dict[tuple[str, ...], _OpenSpan] + ) + # Subset of ``_detached_roots`` keys that represent per-instance + # fan-out roots — they're closed by ``_handle_completed`` on the + # fan-out node's own completion, NOT by ``_sync_subgraph_spans``. + # Using an explicit set rather than parsing the key (e.g., + # checking ``prefix[-1].isdigit()``) so node names that happen + # to be pure digits don't get misclassified. + _fan_out_instance_root_prefixes: set[tuple[str, ...]] = field( + init=False, repr=False, default_factory=set[tuple[str, ...]] + ) + + def __post_init__(self) -> None: + # Private provider per spec §6 TracerProvider isolation — + # MUST NOT be registered globally. + self._provider = TracerProvider() + self._provider.add_span_processor(self.span_processor) + self._tracer = self._provider.get_tracer("openarmature") + + # ------------------------------------------------------------------ + # Observer protocol — async callable accepting a NodeEvent + # ------------------------------------------------------------------ + + async def __call__(self, event: NodeEvent) -> None: + # LLM provider events use a sentinel namespace so we can route + # them to the dedicated §5.5 span path. + if event.namespace == _LLM_NAMESPACE: + if not self.disable_llm_spans: + self._handle_llm_event(event) + return + if event.phase == "checkpoint_saved": + self._emit_checkpoint_save_span(event) + return + if event.phase == "started": + self._handle_started(event) + elif event.phase == "completed": + self._handle_completed(event) + + # ------------------------------------------------------------------ + # Started / completed pairing + # ------------------------------------------------------------------ + + def _handle_started(self, event: NodeEvent) -> None: + """Open a span for this attempt, push onto the in-flight map.""" + from openarmature.observability.correlation import current_correlation_id + + # Lazily open the invocation span on the first event we see + # for this invocation. Detect "first event" by matching the + # correlation_id; the invocation span lives until either + # ``shutdown()`` runs OR a new correlation_id arrives (i.e., + # a new invocation starts on the same long-lived observer). + # The latter close path matters for shared observers reused + # across many invocations — without it the + # ``_invocation_span`` dict grows unbounded. + correlation_id = current_correlation_id() + if correlation_id is not None: + # New correlation_id → close prior invocation spans. + for prior_cid in list(self._invocation_span.keys()): + if prior_cid != correlation_id: + self._close_invocation_span(prior_cid) + if correlation_id not in self._invocation_span: + self._open_invocation_span(correlation_id, event) + + # Synthesize subgraph dispatch spans for any ancestor namespace + # prefix that doesn't have one yet (per observability §4.5). + # Also closes subgraph spans we've left. + self._sync_subgraph_spans(event) + + parent_ctx = self._resolve_parent_context(event) + span = self._tracer.start_span( + name=event.node_name, + context=cast("Any", parent_ctx), + kind=SpanKind.INTERNAL, + attributes=self._node_attrs(event), + ) + token = attach(set_span_in_context(span)) + key = self._key_for(event) + self._open_spans[key] = _OpenSpan(span=span, token=token) + + def _handle_completed(self, event: NodeEvent) -> None: + """Close the matching span, applying §4.2 status mapping.""" + # If this is the fan-out node's own completion AND the + # fan-out is configured detached, close all per-instance + # detached roots that this fan-out spawned. Done BEFORE the + # regular pop so the OTel current-span context is restored + # to the fan-out span's parent (otherwise inner instance + # roots would still be attached). + if event.fan_out_index is None and event.namespace and event.namespace[0] in self.detached_fan_outs: + for key in list(self._detached_roots.keys()): + if len(key) > len(event.namespace) and key[: len(event.namespace)] == event.namespace: + self._close_detached_root(key) + key = self._key_for(event) + open_span = self._open_spans.pop(key, None) + if open_span is None: + # Started event was never delivered (e.g., observer was + # attached mid-invocation). Nothing to close. + return + span = open_span.span + if event.error is not None: + span.set_status(Status(StatusCode.ERROR, description=event.error.category)) + span.record_exception(event.error) + span.set_attribute("openarmature.error.category", event.error.category) + else: + span.set_status(Status(StatusCode.OK)) + span.end() + token = open_span.token + if token is not None: + detach(cast("Any", token)) + # If this was a detached root, drop the root entry so a + # subsequent re-entry mints a fresh trace. + self._detached_roots.pop(event.namespace, None) + + # ------------------------------------------------------------------ + # Special-event paths + # ------------------------------------------------------------------ + + def _emit_checkpoint_save_span(self, event: NodeEvent) -> None: + """Spec pipeline-utilities §10.8 + observability §4.5: emit a + zero-duration ``openarmature.checkpoint.save`` span attached + to the most-recently-opened node span (the node whose + completed event triggered the save).""" + parent_ctx = self._resolve_parent_context(event) + from openarmature.observability.correlation import current_correlation_id + + attrs: dict[str, Any] = { + "openarmature.checkpoint.save_node": event.node_name, + } + cid = current_correlation_id() + if cid is not None: + attrs["openarmature.correlation_id"] = cid + span = self._tracer.start_span( + name="openarmature.checkpoint.save", + context=cast("Any", parent_ctx), + kind=SpanKind.INTERNAL, + attributes=attrs, + ) + span.set_status(Status(StatusCode.OK)) + span.end() + + def _handle_llm_event(self, event: NodeEvent) -> None: + """LLM provider span per spec §5.5 — parented to the node + span that invoked the provider.""" + # ``pre_state`` is a typed ``_LlmEventState`` Pydantic + # subclass — see + # ``openarmature.llm.providers.openai._LlmEventState``. We + # read attributes directly rather than treating the State + # as a dict, preserving the ``NodeEvent.pre_state: State`` + # contract for any other observers that consume the event. + # Lazy import to avoid the otel→llm package dependency at + # module load time. + from openarmature.llm.providers.openai import _LlmEventState + + if not isinstance(event.pre_state, _LlmEventState): + # Defensive — callers other than the OpenAIProvider hook + # shouldn't dispatch through the LLM_NAMESPACE sentinel. + return + payload = event.pre_state + if event.phase == "started": + parent_ctx = self._current_span_context() + attrs: dict[str, Any] = {"openarmature.llm.model": payload.model} + from openarmature.observability.correlation import current_correlation_id + + cid = current_correlation_id() + if cid is not None: + attrs["openarmature.correlation_id"] = cid + span = self._tracer.start_span( + name="openarmature.llm.complete", + context=cast("Any", parent_ctx), + kind=SpanKind.CLIENT, + attributes=attrs, + ) + token = attach(set_span_in_context(span)) + self._open_spans[self._llm_key()] = _OpenSpan(span=span, token=token) + elif event.phase == "completed": + open_span = self._open_spans.pop(self._llm_key(), None) + if open_span is None: + return + span = open_span.span + if payload.finish_reason is not None: + span.set_attribute("openarmature.llm.finish_reason", payload.finish_reason) + if payload.prompt_tokens is not None: + span.set_attribute("openarmature.llm.usage.prompt_tokens", payload.prompt_tokens) + if payload.completion_tokens is not None: + span.set_attribute("openarmature.llm.usage.completion_tokens", payload.completion_tokens) + if payload.total_tokens is not None: + span.set_attribute("openarmature.llm.usage.total_tokens", payload.total_tokens) + if payload.error_type is not None: + span.set_status( + Status( + StatusCode.ERROR, + description=payload.error_category or payload.error_type, + ) + ) + if payload.error_category is not None: + span.set_attribute("openarmature.error.category", payload.error_category) + else: + span.set_status(Status(StatusCode.OK)) + span.end() + detach(cast("Any", open_span.token)) + + # ------------------------------------------------------------------ + # Helpers + # ------------------------------------------------------------------ + + def _open_invocation_span(self, correlation_id: str, event: NodeEvent) -> None: + """Open the root invocation span for a new invocation.""" + from openarmature.observability.correlation import current_invocation_id + + # The first event we receive carries the entry node's + # name; treat it as the invocation's entry_node attribute. + # ``invocation_id`` is set by the engine in ``invoke()`` via + # the ``current_invocation_id`` ContextVar; if the observer + # somehow runs outside an engine invocation (None) we omit + # the attribute rather than emitting a misleading sentinel. + attrs: dict[str, Any] = { + "openarmature.graph.entry_node": event.node_name, + "openarmature.graph.spec_version": self.spec_version, + "openarmature.correlation_id": correlation_id, + } + invocation_id = current_invocation_id() + if invocation_id is not None: + attrs["openarmature.invocation_id"] = invocation_id + span = self._tracer.start_span( + name="openarmature.invocation", + kind=SpanKind.INTERNAL, + attributes=attrs, + ) + token = attach(set_span_in_context(span)) + self._invocation_span[correlation_id] = _OpenSpan(span=span, token=token) + + def _key_for(self, event: NodeEvent) -> _StackKey: + return (event.namespace, event.attempt_index, event.fan_out_index) + + def _llm_key(self) -> _StackKey: + """LLM events are unique within a node body — only one LLM + call can be open at a time per active span scope.""" + return (_LLM_NAMESPACE, 0, None) + + def _resolve_parent_context(self, event: NodeEvent) -> object: + """Return the OTel context to use as the parent for this + event's span. Walks namespace ancestors finding the + innermost-open subgraph or detached root span.""" + # 1a. Detached fan-out instance root — keyed by + # ``namespace[:1] + (str(fan_out_index),)`` per + # ``_open_detached_fan_out_instance_root``. Checked + # explicitly before the generic prefix scan so the + # parent attribution doesn't depend on the + # attach-then-resolve ordering of the surrounding + # ``_sync_subgraph_spans`` call. + if event.fan_out_index is not None and event.namespace: + instance_key = event.namespace[:1] + (str(event.fan_out_index),) + if instance_key in self._detached_roots: + root = self._detached_roots[instance_key] + return set_span_in_context(root.span) + # 1b. Detached subgraph root at any matching prefix wins + # (highest precedence — events inside a detached subtree + # always parent under the detached root, never bleed up). + for prefix_len in range(len(event.namespace) - 1, -1, -1): + prefix = event.namespace[:prefix_len] + if prefix in self._detached_roots: + root = self._detached_roots[prefix] + return set_span_in_context(root.span) + # 2. Innermost synthetic subgraph span at any prefix. + for prefix_len in range(len(event.namespace) - 1, 0, -1): + prefix = event.namespace[:prefix_len] + if prefix in self._subgraph_spans: + sg = self._subgraph_spans[prefix] + return set_span_in_context(sg.span) + # 3. Otherwise, current OTel context (typically invocation span). + return self._current_span_context() + + def _current_span_context(self) -> object: + """Return the current OTel context.""" + return otel_context.get_current() + + def _sync_subgraph_spans(self, event: NodeEvent) -> None: + """Open any synthetic subgraph dispatch spans we need (per + observability §4.5: subgraph wrapper MUST emit a span); close + any subgraph spans whose prefix is no longer an ancestor of + the current event's namespace. + + Called from ``_handle_started`` BEFORE opening the leaf node + span. Detached-mode entries (subgraph or fan-out instance) + are registered as detached roots so their inner spans live + in a fresh trace. + """ + namespace = event.namespace + # 1. Close any open subgraph spans that aren't ancestors of + # the current namespace — we've left those subgraphs. + for prefix in list(self._subgraph_spans.keys()): + if not (len(prefix) < len(namespace) and namespace[: len(prefix)] == prefix): + self._close_subgraph_span(prefix) + # 2. Same for detached subgraph roots — close ones we've + # left. (Detached fan-out instance roots are NOT closed + # here; they close on the fan-out's own completion.) + for prefix in list(self._detached_roots.keys()): + if ( + len(prefix) < len(namespace) + and namespace[: len(prefix)] == prefix + and event.fan_out_index is None + ): + # Still inside this detached subgraph — leave open. + continue + # Detached fan-out instance roots: keyed by namespace + + # (str(fan_out_index),); leave those alone here, they're + # closed when the fan-out parent dispatch completes. + if prefix in self._fan_out_instance_root_prefixes: + # Closed by ``_handle_completed`` on the fan-out + # node's own completion, not here. + continue + if not (len(prefix) < len(namespace) and namespace[: len(prefix)] == prefix): + self._close_detached_root(prefix) + # 3. Open ancestor subgraph spans for any prefix that doesn't + # have one yet. + for depth in range(1, len(namespace)): + prefix = namespace[:depth] + if prefix in self._subgraph_spans: + continue + if prefix in self._detached_roots: + continue + # If this prefix's first segment is configured as a + # detached subgraph, mint a fresh trace. + if depth == 1 and prefix[0] in self.detached_subgraphs: + self._open_detached_subgraph_root(prefix) + continue + # If this is a fan-out instance namespace (event.fan_out_index + # populated, prefix == namespace[:1]), and the fan-out + # node is detached, open a per-instance detached root. + if depth == 1 and event.fan_out_index is not None and prefix[0] in self.detached_fan_outs: + self._open_detached_fan_out_instance_root(prefix, event) + continue + self._open_subgraph_span(prefix) + + def _open_subgraph_span(self, prefix: tuple[str, ...]) -> None: + """Open a synthetic subgraph dispatch span for the given + namespace prefix. Parent is the next-outer subgraph span (or + the invocation span if depth-1).""" + from openarmature.observability.correlation import current_correlation_id + + parent_ctx = self._current_span_context() + # Walk up looking for the nearest enclosing subgraph or + # detached root. + for plen in range(len(prefix) - 1, 0, -1): + outer = prefix[:plen] + if outer in self._subgraph_spans: + parent_ctx = set_span_in_context(self._subgraph_spans[outer].span) + break + if outer in self._detached_roots: + parent_ctx = set_span_in_context(self._detached_roots[outer].span) + break + attrs: dict[str, Any] = { + "openarmature.node.name": prefix[-1], + "openarmature.subgraph.name": prefix[-1], + } + cid = current_correlation_id() + if cid is not None: + attrs["openarmature.correlation_id"] = cid + span = self._tracer.start_span( + name=prefix[-1], + context=cast("Any", parent_ctx), + kind=SpanKind.INTERNAL, + attributes=attrs, + ) + token = attach(set_span_in_context(span)) + self._subgraph_spans[prefix] = _OpenSpan(span=span, token=token) + + def _close_subgraph_span(self, prefix: tuple[str, ...]) -> None: + open_span = self._subgraph_spans.pop(prefix, None) + if open_span is None: + return + open_span.span.set_status(Status(StatusCode.OK)) + open_span.span.end() + # Mirror ``_close_detached_root``: the attach token MUST be + # detached or the OTel current-span context stays pinned to + # a closed span and corrupts parent/child for subsequent + # spans. The cross-context guard handles the case where the + # token was created in a different OTel context (e.g., + # subgraph open/close straddles a detached descent). + if open_span.token is not None: + try: + detach(cast("Any", open_span.token)) + except ValueError: + # Token was created in a different OTel context — + # cross-context detach raises here. The span has + # ended; the leaked context entry is cosmetic and + # unwinds when the worker task exits. + pass + + def _open_detached_subgraph_root(self, prefix: tuple[str, ...]) -> None: + """Mint a fresh trace for a detached subgraph entry. The + detached root span lives in the new trace; the parent trace's + dispatch span (synthesized at the same prefix BUT in the + parent trace) carries an OTel Link to this root. + + Implementation: we open BOTH a parent-trace dispatch span + (with the Link) AND a detached-trace root span (the actual + parent for inner events). The dispatch span ends at sync + time when we leave the prefix; the root span ends when its + children finish.""" + from openarmature.observability.correlation import current_correlation_id + + # 1. Mint the new trace_id + root span_id NOW so the + # parent's Link target matches the detached root's + # SpanContext exactly. + gen = RandomIdGenerator() + detached_trace_id = gen.generate_trace_id() + detached_root_span_id = gen.generate_span_id() + detached_sc = SpanContext( + trace_id=detached_trace_id, + span_id=detached_root_span_id, + is_remote=False, + trace_flags=TraceFlags(TraceFlags.SAMPLED), + ) + + # 2. Open the dispatch span in the parent trace. Carries a + # Link pointing at the detached root's SpanContext. + cid = current_correlation_id() + attrs_parent: dict[str, Any] = { + "openarmature.node.name": prefix[-1], + "openarmature.subgraph.name": prefix[-1], + } + if cid is not None: + attrs_parent["openarmature.correlation_id"] = cid + parent_dispatch = self._tracer.start_span( + name=prefix[-1], + context=cast("Any", self._current_span_context()), + kind=SpanKind.INTERNAL, + links=[Link(detached_sc)], + attributes=attrs_parent, + ) + # Track in _subgraph_spans so the sync routine closes it on + # leaving the prefix. + self._subgraph_spans[prefix] = _OpenSpan(span=parent_dispatch, token=None) + + # 3. Open the detached root span — parented to the synthetic + # detached SpanContext so OTel uses the new trace_id. + detached_parent_ctx = otel_trace.set_span_in_context( + NonRecordingSpan(detached_sc), otel_context.Context() + ) + attrs_root: dict[str, Any] = dict(attrs_parent) + attrs_root["openarmature.subgraph.detached"] = True + detached_root = self._tracer.start_span( + name=prefix[-1], + context=cast("Any", detached_parent_ctx), + kind=SpanKind.INTERNAL, + attributes=attrs_root, + ) + token = attach(set_span_in_context(detached_root)) + self._detached_roots[prefix] = _OpenSpan(span=detached_root, token=token) + + def _open_detached_fan_out_instance_root(self, prefix: tuple[str, ...], event: NodeEvent) -> None: + """Per-instance detached root for a configured-detached + fan-out. Each instance gets its own trace_id; the fan-out + node's span (in the parent trace, already open via the + engine's started event) accumulates Links — one per + instance.""" + from openarmature.observability.correlation import current_correlation_id + + gen = RandomIdGenerator() + detached_trace_id = gen.generate_trace_id() + detached_root_span_id = gen.generate_span_id() + detached_sc = SpanContext( + trace_id=detached_trace_id, + span_id=detached_root_span_id, + is_remote=False, + trace_flags=TraceFlags(TraceFlags.SAMPLED), + ) + + # Find the fan-out node's already-open span in the parent + # trace and add a Link to the detached root. + fan_out_key = self._fan_out_node_span_key(prefix) + fan_out_open = self._open_spans.get(fan_out_key) + if fan_out_open is not None: + fan_out_open.span.add_link(detached_sc) + + # Open the detached instance root span. + detached_parent_ctx = otel_trace.set_span_in_context( + NonRecordingSpan(detached_sc), otel_context.Context() + ) + cid = current_correlation_id() + attrs: dict[str, Any] = { + "openarmature.node.name": prefix[-1], + "openarmature.fan_out.parent_node_name": prefix[-1], + "openarmature.node.fan_out_index": event.fan_out_index, + } + if cid is not None: + attrs["openarmature.correlation_id"] = cid + instance_root = self._tracer.start_span( + name=prefix[-1], + context=cast("Any", detached_parent_ctx), + kind=SpanKind.INTERNAL, + attributes=attrs, + ) + token = attach(set_span_in_context(instance_root)) + # Key by prefix + (str(fan_out_index),) so per-instance + # roots stay distinct. Track separately in + # ``_fan_out_instance_root_prefixes`` so + # ``_sync_subgraph_spans`` can identify them by membership + # (rather than by parsing the key shape). + instance_key = prefix + (str(event.fan_out_index),) + self._detached_roots[instance_key] = _OpenSpan(span=instance_root, token=token) + self._fan_out_instance_root_prefixes.add(instance_key) + + def _close_detached_root(self, prefix: tuple[str, ...]) -> None: + self._fan_out_instance_root_prefixes.discard(prefix) + open_span = self._detached_roots.pop(prefix, None) + if open_span is None: + return + open_span.span.set_status(Status(StatusCode.OK)) + open_span.span.end() + if open_span.token is not None: + try: + detach(cast("Any", open_span.token)) + except ValueError: + # Cross-context detach (token created in a different + # OTel context) — ignore. The span has ended; the + # context entry leaks cosmetically. + pass + + def _fan_out_node_span_key(self, prefix: tuple[str, ...]) -> _StackKey: + """Build the lookup key for a fan-out node's own span (the + parent dispatch span). Fan-out node has no attempt_index ≠ 0 + and no fan_out_index — those fields belong to its inner + instances.""" + return (prefix, 0, None) + + def _node_attrs(self, event: NodeEvent) -> dict[str, Any]: + """Build the §5 attribute set for a node span.""" + from openarmature.observability.correlation import current_correlation_id + + attrs: dict[str, Any] = { + "openarmature.node.name": event.node_name, + "openarmature.node.namespace": list(event.namespace), + "openarmature.node.step": event.step, + "openarmature.node.attempt_index": event.attempt_index, + } + if event.fan_out_index is not None: + attrs["openarmature.node.fan_out_index"] = event.fan_out_index + cid = current_correlation_id() + if cid is not None: + attrs["openarmature.correlation_id"] = cid + return attrs + + # ------------------------------------------------------------------ + # Lifecycle + # ------------------------------------------------------------------ + + def close_invocation(self, correlation_id: str) -> None: + """Public lifecycle hook: close the invocation span for + ``correlation_id``. Idempotent — calling twice (or for a + correlation_id with no open span) is a no-op. + + Long-lived observers shared across many invocations + automatically close prior invocation spans on the first + event of a new invocation (see ``_handle_started``). This + method is for callers who want explicit control without + driving a follow-on invocation, e.g.:: + + await graph.invoke(state, correlation_id=cid) + await graph.drain() + otel_observer.close_invocation(cid) + """ + self._close_invocation_span(correlation_id) + + def _close_invocation_span(self, correlation_id: str) -> None: + """End and remove the invocation span for ``correlation_id``. + + The OTel context token captured when the span opened was + created in the worker task's context — we don't try to + ``detach()`` it (cross-context detach raises ValueError; + the leaked context entry is cosmetic since the worker + eventually exits).""" + open_span = self._invocation_span.pop(correlation_id, None) + if open_span is None: + return + # Status defaults to OK for completed invocations; if the + # engine surfaced an error, the failing node's span + # already carries it and OTel propagates ERROR up the + # parent chain on its own. + open_span.span.set_status(Status(StatusCode.OK)) + open_span.span.end() + + def shutdown(self) -> None: + """Close any still-open invocation/detached spans and shut + down the underlying provider.""" + for cid in list(self._invocation_span.keys()): + self._close_invocation_span(cid) + self._provider.shutdown() + + +__all__ = [ + "OTelObserver", +] diff --git a/tests/conformance/test_observability.py b/tests/conformance/test_observability.py new file mode 100644 index 00000000..1b1f9bcc --- /dev/null +++ b/tests/conformance/test_observability.py @@ -0,0 +1,798 @@ +"""Run spec observability conformance fixtures (001-011) against OTelObserver. + +Phase 6.0 scope (this PR): + +- **001-basic-trace** — full span shape (private TracerProvider, + correlation_id auto-generation, parent-child hierarchy, span names, + base attribute set). +- **005-llm-provider-span-nested** — §5.5 LLM span emission + + ``disable_llm_spans`` opt-out + §6 TracerProvider isolation under + active external auto-instrumentation. The load-bearing + isolation test. +- **008-detached-trace-mode** — §4.4 detached subgraph + detached + fan-out, with cross-trace ``correlation_id`` consistency. +- **009-correlation-id-cross-cutting** — every span carries + ``openarmature.correlation_id``; back-to-back invocations get + distinct UUIDv4s. + +Phase 6.1 (separate follow-up PR): the remaining 7 fixtures — +002 (subgraph hierarchy), 003 (error status), 004 (routing-error +attribution), 006 (fan-out instance attribution), 007 (retry +attempt spans), 010 (log correlation full assertions), 011 +(determinism). Per-fixture wiring notes live in +``docs/phase-6-1-conformance-fillin.md``. +""" + +from __future__ import annotations + +import re +from collections.abc import Mapping +from pathlib import Path +from typing import Any, cast + +import pytest +import yaml + +# Skip the entire module if the ``otel`` extras aren't installed — +# importing ``openarmature.observability.otel`` raises ImportError at +# import time when the extras are missing, which would fail +# collection rather than skipping cleanly. Mirrors the pattern in +# ``tests/unit/test_observability_otel.py``. +pytest.importorskip("opentelemetry.trace") + +from openarmature.observability.otel import OTelObserver # noqa: E402 + +from .adapter import build_graph # noqa: E402 + +CONFORMANCE_DIR = ( + Path(__file__).resolve().parents[2] / "openarmature-spec" / "spec" / "observability" / "conformance" +) + + +_SUPPORTED_FIXTURES = frozenset( + { + "001-otel-basic-trace", + "005-otel-llm-provider-span-nested", + "008-otel-detached-trace-mode", + "009-otel-correlation-id-cross-cutting", + } +) + + +_DEFERRED_FIXTURES: dict[str, str] = { + "002-otel-subgraph-hierarchy": ( + "subgraph dispatch span synthesis (engine wrapper is transparent per " + "fixture 013); deferred to Phase 6.1" + ), + "003-otel-error-status": ( + "status-mapping table across §4 categories; engine path unit-tested. Deferred to Phase 6.1." + ), + "004-otel-routing-error-attribution": ( + "routing errors don't fire their own event pair; preceding-node-span " + "attribution wiring; deferred to Phase 6.1" + ), + "006-otel-fan-out-instance-attribution": ( + "fan-out instance attribution + namespace prefixing; engine path unit-tested. Deferred to Phase 6.1." + ), + "007-otel-retry-attempt-spans": ( + "retry-middleware wiring in the conformance harness; engine emits " + "per-attempt events correctly. Deferred to Phase 6.1." + ), + "010-otel-log-correlation": ( + "OTel Logs Bridge full conformance assertions (trace_id/span_id " + "population on every record); filter unit-tested. Deferred to " + "Phase 6.1." + ), + "011-otel-determinism": ( + "deterministic-portion span content (hierarchy, names, attributes " + "minus timing, status); checked indirectly by other fixtures. " + "Deferred to Phase 6.1." + ), +} + + +# UUIDv4 canonical form: xxxxxxxx-xxxx-4xxx-yxxx-xxxxxxxxxxxx (where y in {8,9,a,b}). +_UUIDV4_RE = re.compile( + r"^[0-9a-f]{8}-[0-9a-f]{4}-4[0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}$", + re.IGNORECASE, +) + + +def _fixture_paths() -> list[Path]: + return sorted(CONFORMANCE_DIR.glob("[0-9][0-9][0-9]-*.yaml")) + + +def _fixture_id(path: Path) -> str: + return path.stem + + +def _load(path: Path) -> dict[str, Any]: + with path.open() as f: + return cast("dict[str, Any]", yaml.safe_load(f)) + + +# --------------------------------------------------------------------------- +# Per-fixture dispatcher +# --------------------------------------------------------------------------- + + +@pytest.mark.parametrize("fixture_path", _fixture_paths(), ids=_fixture_id) +async def test_observability_fixture(fixture_path: Path) -> None: + fixture_id = fixture_path.stem + if fixture_id in _DEFERRED_FIXTURES: + pytest.skip(f"{fixture_id}: {_DEFERRED_FIXTURES[fixture_id]}") + if fixture_id not in _SUPPORTED_FIXTURES: + pytest.skip(f"{fixture_id}: harness wiring not yet implemented") + + spec = _load(fixture_path) + if fixture_id == "001-otel-basic-trace": + await _run_fixture_001(spec) + elif fixture_id == "005-otel-llm-provider-span-nested": + await _run_fixture_005(spec) + elif fixture_id == "008-otel-detached-trace-mode": + await _run_fixture_008(spec) + elif fixture_id == "009-otel-correlation-id-cross-cutting": + await _run_fixture_009(spec) + else: + raise AssertionError(f"no driver for supported fixture {fixture_id!r}") + + +# --------------------------------------------------------------------------- +# Shared helpers +# --------------------------------------------------------------------------- + + +def _build_observer() -> tuple[OTelObserver, Any]: + """Build a fresh OTelObserver + InMemorySpanExporter pair.""" + from opentelemetry.sdk.trace.export import SimpleSpanProcessor + from opentelemetry.sdk.trace.export.in_memory_span_exporter import ( + InMemorySpanExporter, + ) + + exporter = InMemorySpanExporter() + observer = OTelObserver(span_processor=SimpleSpanProcessor(exporter)) + return observer, exporter + + +async def _run_graph( + spec: Mapping[str, Any], + observer: OTelObserver, + *, + correlation_id: str | None = None, +) -> Any: + """Build + invoke a graph from a fixture spec; return the final + state. Caller is responsible for calling ``observer.shutdown()`` + afterwards.""" + trace: list[str] = [] + built = build_graph(spec, trace=trace) + compiled = built.builder.compile() + compiled.attach_observer(observer) + initial_state = built.initial_state(spec.get("initial_state", {})) + final = await compiled.invoke(initial_state, correlation_id=correlation_id) + await compiled.drain() + return final + + +def _all_correlation_ids(spans: Any) -> set[str]: + """Pull the ``openarmature.correlation_id`` attribute off every + span; returns the unique set. Accepts any iterable of spans + (``InMemorySpanExporter.get_finished_spans`` returns a tuple).""" + return {cast("str", dict(s.attributes or {}).get("openarmature.correlation_id")) for s in spans} + + +# --------------------------------------------------------------------------- +# Fixture 001 — basic trace shape +# --------------------------------------------------------------------------- + + +async def _run_fixture_001(spec: Mapping[str, Any]) -> None: + observer, exporter = _build_observer() + final = await _run_graph(spec, observer, correlation_id=spec.get("caller_correlation_id")) + observer.shutdown() + spans = exporter.get_finished_spans() + assert len(spans) == 4, ( + f"expected 4 spans (invocation + 3 nodes); got {len(spans)}: {[s.name for s in spans]}" + ) + by_name = {s.name: s for s in spans} + assert "openarmature.invocation" in by_name + inv = by_name["openarmature.invocation"] + assert inv.parent is None + inv_attrs = dict(inv.attributes or {}) + assert inv_attrs.get("openarmature.graph.entry_node") == spec["entry"] + cid = inv_attrs.get("openarmature.correlation_id") + assert isinstance(cid, str) and len(cid) > 0 + inv_ctx = inv.context + assert inv_ctx is not None + invocation_span_id = inv_ctx.span_id + for node_name in spec["nodes"]: + assert node_name in by_name, f"missing span for {node_name!r}" + node_span = by_name[node_name] + node_parent = node_span.parent + assert node_parent is not None and node_parent.span_id == invocation_span_id + node_attrs = dict(node_span.attributes or {}) + assert node_attrs.get("openarmature.node.name") == node_name + assert list(node_attrs.get("openarmature.node.namespace") or []) == [node_name] + assert isinstance(node_attrs.get("openarmature.node.step"), int) + assert node_attrs.get("openarmature.node.attempt_index") == 0 + assert node_attrs.get("openarmature.correlation_id") == cid + expected_trace = ["a", "b", "c"] + assert final.trace == expected_trace # type: ignore[attr-defined] + + +# --------------------------------------------------------------------------- +# Fixture 009 — correlation_id cross-cutting +# --------------------------------------------------------------------------- + + +async def _run_fixture_009(spec: Mapping[str, Any]) -> None: + """Three sub-cases, each in the ``cases:`` block: + + 1. caller-supplied correlation_id used verbatim on every span. + 2. auto-generated UUIDv4 used uniformly across all spans. + 3. Two back-to-back invocations get DIFFERENT correlation_ids, + both UUIDv4 form. + """ + cases = cast("list[dict[str, Any]]", spec["cases"]) + for case in cases: + case_name = cast("str", case["name"]) + try: + await _run_fixture_009_case(case) + except AssertionError as e: + raise AssertionError(f"case {case_name!r}: {e}") from e + + +async def _run_fixture_009_case(case: Mapping[str, Any]) -> None: + case_name = case["name"] + if case_name == "context_reset_between_invocations": + # Two back-to-back invocations of the same compiled graph; + # each MUST get its own UUIDv4 (distinct from the other). + observer, exporter = _build_observer() + # Build the graph ONCE so both invocations share it. + from .adapter import build_graph as _bg + + built = _bg(case) + compiled = built.builder.compile() + compiled.attach_observer(observer) + for _ in range(int(case.get("invocations", 2))): + await compiled.invoke(built.initial_state(case.get("initial_state", {}))) + await compiled.drain() + observer.shutdown() + spans = exporter.get_finished_spans() + + # Group spans by trace_id (each invocation has its own trace). + by_trace: dict[int, list[Any]] = {} + for s in spans: + tid = s.context.trace_id + by_trace.setdefault(tid, []).append(s) + assert len(by_trace) == 2, f"expected 2 distinct traces (one per invocation); got {len(by_trace)}" + # Each invocation's spans share one correlation_id. + per_invocation_cids: list[str] = [] + for trace_spans in by_trace.values(): + cids = _all_correlation_ids(trace_spans) + assert len(cids) == 1, f"each invocation MUST uniformly carry one correlation_id; got {cids}" + cid = next(iter(cids)) + assert _UUIDV4_RE.match(cid), f"auto-generated correlation_id MUST be UUIDv4; got {cid!r}" + per_invocation_cids.append(cid) + # Cross-invocation: distinct. + assert per_invocation_cids[0] != per_invocation_cids[1], ( + "back-to-back invocations MUST get distinct correlation_ids" + ) + return + + # Sub-cases 1 & 2 (single-invocation). + observer, exporter = _build_observer() + await _run_graph(case, observer, correlation_id=case.get("caller_correlation_id")) + observer.shutdown() + spans = exporter.get_finished_spans() + cids = _all_correlation_ids(spans) + assert len(cids) == 1, f"every span MUST carry the same correlation_id; got {cids}" + cid = next(iter(cids)) + expected = case.get("caller_correlation_id") + if expected is not None: + # Caller-supplied → exact match. + assert cid == expected, ( + f"caller_correlation_id MUST be used verbatim; got {cid!r}, expected {expected!r}" + ) + else: + # Auto-generated → UUIDv4. + assert _UUIDV4_RE.match(cid), f"auto-generated correlation_id MUST be UUIDv4; got {cid!r}" + + +# --------------------------------------------------------------------------- +# Fixture 005, 008 placeholders — driven below in subsequent commits +# --------------------------------------------------------------------------- + + +async def _run_fixture_005(spec: Mapping[str, Any]) -> None: + """Three sub-cases: + + 1. ``default`` — LLM span emits with §5 attributes, parented under + the calling node. + 2. ``disable_llm_spans`` — opt-out suppresses the LLM span entirely. + 3. ``external_auto_instrumentation_active`` — second exporter on + the OTel global provider; openarmature spans MUST NOT leak to + it (the load-bearing §6 TracerProvider isolation guarantee). + """ + cases = cast("list[dict[str, Any]]", spec["cases"]) + for case in cases: + case_name = cast("str", case["name"]) + try: + await _run_fixture_005_case(case) + except AssertionError as e: + raise AssertionError(f"case {case_name!r}: {e}") from e + + +async def _run_fixture_005_case(case: Mapping[str, Any]) -> None: + case_name = case["name"] + disable_llm_spans = bool(case.get("disable_llm_spans", False)) + caller_global_active = bool(case.get("caller_global_otel_active", False)) + + from opentelemetry import trace as otel_trace + from opentelemetry.sdk.trace import TracerProvider + from opentelemetry.sdk.trace.export import SimpleSpanProcessor + from opentelemetry.sdk.trace.export.in_memory_span_exporter import ( + InMemorySpanExporter, + ) + + # Optional second exporter on the OTel global provider — sub-case 3. + # Save the prior global provider so we can restore it after the + # case (otherwise the global state leaks to subsequent tests). + global_exporter: InMemorySpanExporter | None = None + prior_global = otel_trace.get_tracer_provider() if caller_global_active else None + if caller_global_active: + global_exporter = InMemorySpanExporter() + global_provider = TracerProvider() + global_provider.add_span_processor(SimpleSpanProcessor(global_exporter)) + otel_trace.set_tracer_provider(global_provider) + + try: + private_exporter = InMemorySpanExporter() + observer = OTelObserver( + span_processor=SimpleSpanProcessor(private_exporter), + disable_llm_spans=disable_llm_spans, + ) + + # Build a graph whose entry node calls a mock LLM provider. + graph, _ = _build_graph_with_mock_llm(case) + graph.attach_observer(observer) + + # Drive the graph. The ``calls_llm`` node body reads the mock + # responses set up in ``_build_graph_with_mock_llm`` (httpx + # MockTransport keyed off the response queue). + initial_state_cls = graph.state_cls + final = await graph.invoke(initial_state_cls()) + await graph.drain() + observer.shutdown() + private_spans = private_exporter.get_finished_spans() + + # Sub-case 3: external span emitted by the harness through the + # global tracer (simulating auto-instrumentation). + if caller_global_active: + global_tracer = otel_trace.get_tracer("external-instrumentation") + with global_tracer.start_as_current_span("external.llm.call"): + pass + assert global_exporter is not None + global_spans = global_exporter.get_finished_spans() + assert len(global_spans) == 1, ( + f"global exporter MUST see exactly one external span; got {len(global_spans)}" + ) + assert global_spans[0].name == "external.llm.call" + # The load-bearing isolation check. + for s in global_spans: + assert not s.name.startswith("openarmature."), ( + f"openarmature spans MUST NOT leak to the global provider; got {s.name!r}" + ) + finally: + if caller_global_active and prior_global is not None: + otel_trace.set_tracer_provider(prior_global) + + # Common assertions: the LLM span presence/absence + (when + # present) attributes + parent-child to the calling node. + llm_spans = [s for s in private_spans if s.name == "openarmature.llm.complete"] + if disable_llm_spans: + assert not llm_spans, ( + f"disable_llm_spans=True MUST suppress LLM span emission; got {len(llm_spans)} llm spans" + ) + # ask_llm node span still emits. + assert any(s.name == "ask_llm" for s in private_spans) + return + + # default + external_auto_instrumentation_active — LLM span + # MUST emit with the spec §5.5 attributes. + assert len(llm_spans) == 1, ( + f"expected one LLM span; got {len(llm_spans)}: {[s.name for s in private_spans]}" + ) + llm = llm_spans[0] + attrs = dict(llm.attributes or {}) + assert attrs.get("openarmature.llm.model") == "test-model" + if case_name == "default": + # Sub-case 1 asserts the full attribute set. + assert attrs.get("openarmature.llm.finish_reason") == "stop" + assert attrs.get("openarmature.llm.usage.prompt_tokens") == 5 + assert attrs.get("openarmature.llm.usage.completion_tokens") == 1 + assert attrs.get("openarmature.llm.usage.total_tokens") == 6 + # Parent: the ask_llm node span. + ask_llm = next((s for s in private_spans if s.name == "ask_llm"), None) + assert ask_llm is not None, "expected ask_llm node span" + llm_parent = llm.parent + ask_llm_ctx = ask_llm.context + assert llm_parent is not None and ask_llm_ctx is not None + assert llm_parent.span_id == ask_llm_ctx.span_id, ( + "openarmature.llm.complete MUST be parented under the calling node span" + ) + # Final state was updated by the calls_llm node (msg = "hello" + # for default; "hi back" for disable_llm_spans path that we + # already returned from above). + assert "msg" in dir(final) + + +def _build_graph_with_mock_llm(case: Mapping[str, Any]) -> tuple[Any, list[Any]]: + """Build a graph whose entry node invokes ``OpenAIProvider.complete`` + against an ``httpx.MockTransport`` preloaded with the fixture's + ``mock_llm`` responses.""" + import json + from collections.abc import Sequence + + import httpx + + from openarmature.graph import GraphBuilder + from openarmature.llm import OpenAIProvider, UserMessage + + mock_responses = list(cast("list[dict[str, Any]]", case.get("mock_llm") or [])) + + def _handler(request: httpx.Request) -> httpx.Response: + if not mock_responses: + raise AssertionError("mock_llm queue exhausted") + spec_resp = mock_responses.pop(0) + body = cast("dict[str, Any]", spec_resp.get("body") or {}) + return httpx.Response( + int(spec_resp.get("status", 200)), + content=json.dumps(body).encode("utf-8"), + headers={"Content-Type": "application/json"}, + ) + + transport = httpx.MockTransport(_handler) + provider = OpenAIProvider( + base_url="http://mock-llm.test", + model="test-model", + api_key="test", + transport=transport, + ) + + # Build a State subclass with the fixture's declared fields. + from .adapter import build_state_cls + + state_fields = cast("dict[str, dict[str, Any]]", case["state"]["fields"]) + state_cls = build_state_cls("LlmFixtureState", state_fields) + + # Node body: calls the LLM provider and writes the response into + # the configured field. + nodes = cast("dict[str, Any]", case["nodes"]) + entry_name = cast("str", case["entry"]) + calls_llm_spec = cast("dict[str, Any]", nodes[entry_name]["calls_llm"]) + stores_in = cast("str", calls_llm_spec.get("stores_response_in", "msg")) + messages_spec = cast("list[dict[str, str]]", calls_llm_spec.get("messages", [])) + messages: Sequence[Any] = [ + UserMessage(content=m["content"]) for m in messages_spec if m.get("role") == "user" + ] + + async def ask_llm_body(_s: Any) -> dict[str, str]: + response = await provider.complete(messages) + return {stores_in: response.message.content or ""} + + builder = ( + GraphBuilder(state_cls) + .add_node(entry_name, ask_llm_body) + .add_edge(entry_name, _resolve_target_for_005(case)) + .set_entry(entry_name) + ) + return builder.compile(), mock_responses + + +def _resolve_target_for_005(case: Mapping[str, Any]) -> Any: + """Fixture 005's edges go to END. Return the END sentinel.""" + from openarmature.graph import END + + edges = cast("list[dict[str, Any]]", case.get("edges") or []) + if not edges: + return END + target = edges[0].get("to") + return END if target == "END" else target + + +async def _run_fixture_008(spec: Mapping[str, Any]) -> None: + """Two sub-cases: detached subgraph (one Link, two traces, shared + correlation_id) and detached fan-out (one trace per instance, + each with a Link from the fan-out node span).""" + cases = cast("list[dict[str, Any]]", spec["cases"]) + for case in cases: + case_name = cast("str", case["name"]) + try: + await _run_fixture_008_case(case) + except AssertionError as e: + raise AssertionError(f"case {case_name!r}: {e}") from e + + +async def _run_fixture_008_case(case: Mapping[str, Any]) -> None: + case_name = case["name"] + # The fixture configures detached subgraphs by the SUBGRAPH'S + # IDENTITY NAME (the key in ``subgraphs:``), but the OTel observer + # keys on the WRAPPER NODE'S NAME in the parent graph (consistent + # with graph-engine §6's namespace convention — see fixture 029 + # spec note). Translate by looking up the wrapper node that + # references each detached subgraph identity. + detached_subgraph_identities = set(cast("list[str]", case.get("detached_subgraphs") or [])) + nodes = cast("dict[str, Any]", case.get("nodes") or {}) + wrapper_names_for_detached: set[str] = set() + for wrapper_name, node_spec in nodes.items(): + sub_id = cast("dict[str, Any]", node_spec).get("subgraph") + if isinstance(sub_id, str) and sub_id in detached_subgraph_identities: + wrapper_names_for_detached.add(wrapper_name) + detached_subgraphs = frozenset(wrapper_names_for_detached) + detached_fan_outs = frozenset(cast("list[str]", case.get("detached_fan_outs") or [])) + + from opentelemetry.sdk.trace.export import SimpleSpanProcessor + from opentelemetry.sdk.trace.export.in_memory_span_exporter import ( + InMemorySpanExporter, + ) + + exporter = InMemorySpanExporter() + observer = OTelObserver( + span_processor=SimpleSpanProcessor(exporter), + detached_subgraphs=detached_subgraphs, + detached_fan_outs=detached_fan_outs, + ) + + # Patch the inner subgraph's ``update_pure_from_state`` directive + # if present — the adapter doesn't translate it, but the test + # assertions only inspect span structure (Links, trace counts), + # not the computed values. Replacing with a no-op ``update_pure`` + # keeps the graph runnable. + _patch_unsupported_directives(case) + + # Build subgraphs declared by the fixture (subgraph: or subgraphs:). + subgraphs = _compile_subgraphs(case) + trace_log: list[str] = [] + built = build_graph(case, subgraphs=subgraphs, trace=trace_log) + compiled = built.builder.compile() + compiled.attach_observer(observer) + initial_state = built.initial_state(case.get("initial_state", {})) + await compiled.invoke(initial_state) + await compiled.drain() + observer.shutdown() + spans = exporter.get_finished_spans() + + if case_name == "detached_subgraph_two_traces_one_link": + # Group by trace_id. Span context is non-None for any span + # the SDK actually exported, so the cast keeps pyright quiet + # on the `.trace_id` access. + by_trace: dict[int, list[Any]] = {} + for s in spans: + ctx = cast("Any", s.context) + by_trace.setdefault(ctx.trace_id, []).append(s) + assert len(by_trace) == 2, ( + f"expected 2 distinct traces (parent + detached subgraph); " + f"got {len(by_trace)}: {[s.name for s in spans]}" + ) + # Cross-trace correlation_id consistency (§3). + cids = _all_correlation_ids(spans) + assert len(cids) == 1, f"correlation_id MUST flow unchanged across detached boundary; got {cids}" + # Find the parent dispatch span (it's the one with a Link). + dispatch_spans = [s for s in spans if s.name == "dispatch"] + # Two "dispatch" spans: one in parent trace (with Link), one in + # detached trace (the root). Pick the one with links. + parent_dispatch = next((s for s in dispatch_spans if s.links), None) + assert parent_dispatch is not None, "expected a 'dispatch' span carrying a Link to the detached trace" + assert len(parent_dispatch.links) == 1, ( + f"dispatch span MUST carry exactly one Link; got {len(parent_dispatch.links)}" + ) + link_target_trace_id = parent_dispatch.links[0].context.trace_id + # The link's trace_id matches the detached trace's actual trace_id. + parent_trace_id = cast("Any", parent_dispatch.context).trace_id + detached_dispatch = next( + (s for s in dispatch_spans if not s.links and cast("Any", s.context).trace_id != parent_trace_id), + None, + ) + assert detached_dispatch is not None + detached_trace_id = cast("Any", detached_dispatch.context).trace_id + assert link_target_trace_id == detached_trace_id, ( + f"Link target trace_id MUST match the detached trace's trace_id; " + f"got link={link_target_trace_id!r}, detached={detached_trace_id!r}" + ) + return + + if case_name == "detached_fan_out_one_trace_per_instance": + # Group by trace_id. + by_trace = {} + for s in spans: + ctx = cast("Any", s.context) + by_trace.setdefault(ctx.trace_id, []).append(s) + # 4 traces total: 1 parent + 3 instance traces. + assert len(by_trace) == 4, f"expected 4 traces (parent + 3 instances); got {len(by_trace)}" + # All 4 share the same correlation_id. + cids = _all_correlation_ids(spans) + assert len(cids) == 1, ( + f"correlation_id MUST be uniform across parent + detached instance traces; got {cids}" + ) + # Find the fan-out node span — it's in the parent trace and + # carries 3 Links. + fan_out_node_spans = [s for s in spans if s.name == "per_document_scoring"] + # Three of these are inside detached instance roots; one is in + # the parent trace and is the one with Links. + parent_fan_out = next((s for s in fan_out_node_spans if s.links), None) + assert parent_fan_out is not None, "expected a fan-out span with Links in parent trace" + assert len(parent_fan_out.links) == 3, ( + f"fan-out span MUST carry one Link per instance (3); got {len(parent_fan_out.links)}" + ) + return + + raise AssertionError(f"unknown sub-case {case_name!r}") + + +def _patch_unsupported_directives(spec: Mapping[str, Any]) -> None: + """Replace test-seam directives the conformance adapter doesn't + yet translate (``update_pure_from_state`` etc.) with a benign + ``update_pure: {}`` no-op. The observability fixtures only + assert span structure (parent-child, Links, trace_ids, + correlation_id), not state values, so the swap is safe.""" + + def patch_nodes(graph_block: Mapping[str, Any] | None) -> None: + if not graph_block: + return + nodes = cast("dict[str, Any]", graph_block.get("nodes") or {}) + for node_spec_any in nodes.values(): + if not isinstance(node_spec_any, dict): + continue + node_spec = cast("dict[str, Any]", node_spec_any) + for unsupported in ( + "update_pure_from_state", + "calls_llm", + ): + if unsupported in node_spec: + node_spec.pop(unsupported) + node_spec.setdefault("update_pure", {}) + + patch_nodes(spec) + if "subgraph" in spec: + patch_nodes(cast("Mapping[str, Any]", spec["subgraph"])) + for sub in cast("dict[str, Any]", spec.get("subgraphs") or {}).values(): + patch_nodes(cast("Mapping[str, Any]", sub)) + + +def _compile_subgraphs(spec: Mapping[str, Any]) -> dict[str, Any]: + """Build any subgraphs declared by the fixture and return a + name→compiled-graph registry the adapter consumes.""" + subgraph_specs: dict[str, Any] = {} + if "subgraph" in spec: + single = cast("Mapping[str, Any]", spec["subgraph"]) + name = single.get("name") or "subgraph" + subgraph_specs[name] = single + if "subgraphs" in spec: + for k, v in cast("dict[str, Any]", spec["subgraphs"]).items(): + subgraph_specs[k] = v + compiled_subgraphs: dict[str, Any] = {} + for name, sub_spec in subgraph_specs.items(): + sub_built = build_graph(sub_spec, trace=[]) + compiled_subgraphs[name] = sub_built.builder.compile() + return compiled_subgraphs + + +# --------------------------------------------------------------------------- +# Phase 5 fixture 031 — span/log assertions deferred from Phase 5 +# +# Lives in this file (not test_checkpoint.py) because the assertions +# verify OTel span attributes across the original + resumed runs of +# the same checkpoint fixture. The Phase 5 harness already covers the +# record-level half (correlation_id preserved, invocation_id changes); +# this picks up the cross-run span-attribute half. +# --------------------------------------------------------------------------- + + +_PIPELINE_CONFORMANCE_DIR = ( + Path(__file__).resolve().parents[2] / "openarmature-spec" / "spec" / "pipeline-utilities" / "conformance" +) + + +async def test_phase5_fixture_031_span_assertions() -> None: + """Spec §10.4 step 3 + step 4 + observability §3 / §5.6: every + span across BOTH the original and resumed runs MUST carry the + same ``openarmature.correlation_id``; ``invocation_id`` differs + across the two runs (each is its own invocation in the + observability sense).""" + fixture_path = _PIPELINE_CONFORMANCE_DIR / "031-checkpoint-correlation-id-preserved-across-resume.yaml" + spec = _load(fixture_path) + cases = cast("list[dict[str, Any]]", spec["cases"]) + for case in cases: + case_name = cast("str", case["name"]) + try: + await _run_fixture_031_case(case) + except AssertionError as e: + raise AssertionError(f"case {case_name!r}: {e}") from e + + +async def _run_fixture_031_case(case: Mapping[str, Any]) -> None: + from openarmature.checkpoint import CheckpointRecord, InMemoryCheckpointer + from openarmature.checkpoint.protocol import Checkpointer + from openarmature.graph import RuntimeGraphError + + class _CapturingCheckpointer: + """Mirrors the local capture pattern in test_checkpoint.py + but inlined here so the observability test doesn't depend on + that file's internal helpers.""" + + def __init__(self) -> None: + self._inner = InMemoryCheckpointer() + self.saves: list[CheckpointRecord] = [] + + async def save(self, invocation_id: str, record: CheckpointRecord) -> None: + self.saves.append(record) + await self._inner.save(invocation_id, record) + + async def load(self, invocation_id: str) -> CheckpointRecord | None: + return await self._inner.load(invocation_id) + + async def list(self, filter: Any = None) -> Any: + return await self._inner.list(filter) + + async def delete(self, invocation_id: str) -> None: + await self._inner.delete(invocation_id) + + capturing = _CapturingCheckpointer() + observer, exporter = _build_observer() + + trace: list[str] = [] + built = build_graph(case, trace=trace) + builder = built.builder + builder.with_checkpointer(cast("Checkpointer", capturing)) + compiled = builder.compile() + compiled.attach_observer(observer) + initial_state = built.initial_state(case.get("initial_state", {})) + + # First run — expected to abort. + expected_error = cast("Mapping[str, Any]", case["first_run_expected_error"]) + caller_cid = case.get("caller_correlation_id") + with pytest.raises(RuntimeGraphError) as excinfo: + await compiled.invoke(initial_state, correlation_id=caller_cid) + assert excinfo.value.category == expected_error["category"] + await compiled.drain() + + # Capture first run's invocation_id from the latest save. + assert capturing.saves, "expected at least one save before the abort" + first_invocation_id = capturing.saves[-1].invocation_id + first_correlation_id = capturing.saves[-1].correlation_id + if caller_cid is not None: + assert first_correlation_id == caller_cid, ( + f"first run MUST preserve caller-supplied correlation_id; " + f"got {first_correlation_id!r}, expected {caller_cid!r}" + ) + else: + # Auto-generated → UUIDv4 form. + assert _UUIDV4_RE.match(first_correlation_id), ( + f"auto-generated correlation_id MUST be UUIDv4; got {first_correlation_id!r}" + ) + + # Resume — should succeed. + capturing.saves.clear() + await compiled.invoke(initial_state, resume_invocation=first_invocation_id) + await compiled.drain() + observer.shutdown() + + # ----- Span assertions (the §10.4 / §3 invariants) ----- + spans = exporter.get_finished_spans() + # Every span across both runs MUST carry the same correlation_id. + cids = _all_correlation_ids(spans) + assert len(cids) == 1, f"correlation_id MUST be uniform across both runs; got {cids}" + cid = next(iter(cids)) + assert cid == first_correlation_id, ( + f"resumed run MUST preserve original correlation_id; got {cid!r}, original {first_correlation_id!r}" + ) + # The original and resumed runs MUST have DIFFERENT trace_ids + # (each invocation is its own trace per §5.1's + # ``invocation_id`` semantics — different invocation_id ↔ + # different OTel trace_id under the default in-trace + # parent-child rules). + trace_ids = {s.context.trace_id for s in spans} + assert len(trace_ids) == 2, ( + f"original and resumed runs MUST produce DIFFERENT trace_ids " + f"(per §10.4 step 4 + §5.1); got {len(trace_ids)} distinct trace_ids" + ) diff --git a/tests/unit/test_correlation.py b/tests/unit/test_correlation.py new file mode 100644 index 00000000..e0edd5ab --- /dev/null +++ b/tests/unit/test_correlation.py @@ -0,0 +1,210 @@ +"""Cross-backend correlation primitives — no OTel deps. + +Verifies the spec observability §3 contract independently of any +backend mapping. Lives in the unit test root rather than under any +backend-specific directory because correlation_id is core: it MUST be +readable from any user code (node bodies, middleware, observers) even +when no observability backend is configured. +""" + +from __future__ import annotations + +import asyncio + +import pytest + +from openarmature.graph import END, GraphBuilder, NodeException, State +from openarmature.observability import current_correlation_id + + +class _S(State): + captured: str = "" + flag: bool = False + + +async def _read_correlation(state: _S) -> dict[str, str]: + cid = current_correlation_id() + return {"captured": cid or ""} + + +# --------------------------------------------------------------------------- +# §3.1 lifecycle: caller-supplied + auto-generated +# --------------------------------------------------------------------------- + + +async def test_caller_supplied_correlation_id_visible_inside_node() -> None: + """User code in a node body can read the supplied correlation_id + via :func:`current_correlation_id` — the spec's mandated + cross-backend join key surface.""" + g = GraphBuilder(_S).add_node("read", _read_correlation).add_edge("read", END).set_entry("read").compile() + final = await g.invoke(_S(), correlation_id="my-business-request-42") + assert final.captured == "my-business-request-42" + + +async def test_auto_generated_correlation_id_is_uuidv4() -> None: + """Per spec §3.1, when the caller does not supply a correlation_id + the framework MUST auto-generate a canonical 36-character UUIDv4.""" + g = GraphBuilder(_S).add_node("read", _read_correlation).add_edge("read", END).set_entry("read").compile() + final = await g.invoke(_S()) # no caller correlation_id + cid = final.captured + # Canonical UUIDv4 form: xxxxxxxx-xxxx-4xxx-yxxx-xxxxxxxxxxxx (36 chars). + assert len(cid) == 36 + assert cid[14] == "4" # version-4 nibble + parts = cid.split("-") + assert len(parts) == 5 + assert [len(p) for p in parts] == [8, 4, 4, 4, 12] + + +async def test_correlation_id_resets_between_invocations() -> None: + """Spec §3.1: ``Reset the context after the invocation completes + so subsequent invocations get fresh correlation IDs.``""" + # Outside any invocation, correlation_id is None. + assert current_correlation_id() is None + g = GraphBuilder(_S).add_node("read", _read_correlation).add_edge("read", END).set_entry("read").compile() + await g.invoke(_S(), correlation_id="invocation-one") + # After invoke returns, the ContextVar is reset. + assert current_correlation_id() is None + final2 = await g.invoke(_S(), correlation_id="invocation-two") + assert final2.captured == "invocation-two" + # And after the second invoke too. + assert current_correlation_id() is None + + +async def test_correlation_id_isolated_across_concurrent_invocations() -> None: + """ContextVar isolation: two concurrent invocations see their own + correlation_id values — no cross-contamination via the shared + ContextVar.""" + g = GraphBuilder(_S).add_node("read", _read_correlation).add_edge("read", END).set_entry("read").compile() + final_a, final_b = await asyncio.gather( + g.invoke(_S(), correlation_id="A"), + g.invoke(_S(), correlation_id="B"), + ) + assert final_a.captured == "A" + assert final_b.captured == "B" + + +# --------------------------------------------------------------------------- +# §3.2 distinction from invocation_id +# --------------------------------------------------------------------------- + + +async def test_correlation_id_and_invocation_id_are_structurally_distinct() -> None: + """Spec observability §3.2: ``correlation_id`` and + ``invocation_id`` serve different purposes and MUST NOT be + conflated. Drive a real invocation with a checkpointer and read + both ids from the saved record (deterministic) — plus an in-body + cross-check via the public ContextVar readers — to verify the + framework treats them as independent fields.""" + from openarmature.checkpoint import InMemoryCheckpointer + from openarmature.observability import current_invocation_id + + captured: dict[str, str | None] = {} + + async def read_both(_s: _S) -> dict[str, str]: + captured["correlation_id"] = current_correlation_id() + captured["invocation_id"] = current_invocation_id() + return {"captured": captured.get("invocation_id") or ""} + + cp = InMemoryCheckpointer() + g = ( + GraphBuilder(_S) + .add_node("read", read_both) + .add_edge("read", END) + .set_entry("read") + .with_checkpointer(cp) + .compile() + ) + await g.invoke(_S(), correlation_id="user-supplied-cid") + + # In-body cross-check: the two ContextVars MUST return distinct + # strings. ``correlation_id`` is the caller-supplied value; + # ``invocation_id`` is framework-minted. + body_corr = captured["correlation_id"] + body_inv = captured["invocation_id"] + assert body_corr == "user-supplied-cid" + assert body_inv is not None and body_inv != body_corr + + # Saved-record cross-check: the framework persists both fields + # independently and they MUST differ. + summaries = list(await cp.list()) + assert len(summaries) == 1 + record = await cp.load(summaries[0].invocation_id) + assert record is not None + assert record.correlation_id == "user-supplied-cid" + assert record.invocation_id != record.correlation_id + assert record.invocation_id == body_inv + + +# --------------------------------------------------------------------------- +# Outside-invocation safety +# --------------------------------------------------------------------------- + + +def test_current_correlation_id_returns_none_outside_invocation() -> None: + """Reading ``current_correlation_id()`` outside any invocation + MUST return None (not raise, not return empty string). User code + that may run inside or outside a graph context can rely on this.""" + assert current_correlation_id() is None + + +# --------------------------------------------------------------------------- +# Phase 5 / §10.4 step 3 + 4 — resume preserves correlation_id, mints new +# invocation_id. Already covered in test_checkpoint.py at the record +# level; here we additionally verify the user-visible ContextVar half. +# --------------------------------------------------------------------------- + + +async def test_resume_preserves_correlation_id_visible_to_user_code() -> None: + """Spec §10.4 step 3: resume MUST preserve the original + correlation_id verbatim. The Phase 5 checkpoint test verifies + this at the saved-record level; here we additionally verify it + propagates to the ContextVar that user code reads from inside + node bodies during the resumed invocation.""" + from openarmature.checkpoint import InMemoryCheckpointer + + class _ResumeState(State): + flag: bool = False + observed_cid: str = "" + + fail_once = [True] + + async def maybe_fail(_s: _ResumeState) -> dict[str, str | bool]: + # Read the correlation_id from inside the node body. + cid = current_correlation_id() or "" + if fail_once[0]: + fail_once[0] = False + raise RuntimeError("first-run abort") + return {"flag": True, "observed_cid": cid} + + cp = InMemoryCheckpointer() + # The flaky-node abort needs a save to fire BEFORE the failure so + # the resume path has something to load from. Single-node graph + # would never save (the abort happens before its own merge); add + # a ``pre`` node whose successful save arms the resume. + pre = ( + GraphBuilder(_ResumeState) + .add_node("pre", _pre) + .add_node("a", maybe_fail) + .add_edge("pre", "a") + .add_edge("a", END) + .set_entry("pre") + .with_checkpointer(cp) + .compile() + ) + fail_once[0] = True + with pytest.raises(NodeException): + await pre.invoke(_ResumeState(), correlation_id="my-correlation-cid") + # Find the saved invocation_id from the only record in the + # checkpointer. + summaries = list(await cp.list()) + assert len(summaries) == 1 + saved_invocation_id = summaries[0].invocation_id + + # Resume — the flaky node now succeeds and reads the + # correlation_id from the ContextVar inside its body. + final = await pre.invoke(_ResumeState(), resume_invocation=saved_invocation_id) + assert final.observed_cid == "my-correlation-cid" + + +async def _pre(_s: State) -> dict[str, bool]: + return {"flag": True} diff --git a/tests/unit/test_llm_provider.py b/tests/unit/test_llm_provider.py index 8a547f6e..6accd8d6 100644 --- a/tests/unit/test_llm_provider.py +++ b/tests/unit/test_llm_provider.py @@ -277,8 +277,12 @@ def test_transient_categories_excludes_terminal_categories() -> None: def _make_provider() -> OpenAIProvider: - """Build a provider with a no-op transport — these tests don't - actually call the wire, only ``_classify_http_error``.""" + """Build a provider for unit tests of ``_classify_http_error``. + + The transport exists only to satisfy provider construction; these + tests pass explicit ``httpx.Response`` instances directly into + ``_classify_http_error`` and never perform wire calls. + """ transport = httpx.MockTransport(lambda _req: httpx.Response(204)) return OpenAIProvider(base_url="http://test", model="m", api_key="k", transport=transport) @@ -388,8 +392,8 @@ def test_classify_504_to_unavailable() -> None: async def test_complete_does_not_mutate_messages_or_tools() -> None: """The input messages/tools list MUST NOT be mutated by complete(). Snapshot the inputs (deep-copy via Pydantic round-trip), run a - happy-path call, and assert the input objects are byte-identical - after the call returns. + happy-path call, and assert the input objects remain equal to + their pre-call snapshots after the call returns. """ def _ok(_req: httpx.Request) -> httpx.Response: diff --git a/tests/unit/test_middleware.py b/tests/unit/test_middleware.py index a0c7ea1c..b4ee0eb4 100644 --- a/tests/unit/test_middleware.py +++ b/tests/unit/test_middleware.py @@ -294,7 +294,7 @@ async def innermost(_state: Any) -> Mapping[str, Any]: assert abs(rec.duration_ms - 10.0) < 0.01 -# ===== Reentrant next: middleware can call next more than twice ===== +# ===== Reentrant next: middleware can call next more than once ===== async def test_middleware_can_call_next_repeatedly() -> None: @@ -302,7 +302,7 @@ async def test_middleware_can_call_next_repeatedly() -> None: exercises this with N=2-3 attempts; this test pins the contract independently by calling ``next`` 5 times in a loop and asserting the inner runs exactly that many times.""" - inner_calls = [0] + inner_calls = 0 async def call_five_times(state: Any, next_: NextCall) -> Mapping[str, Any]: last: Mapping[str, Any] = {} @@ -311,13 +311,14 @@ async def call_five_times(state: Any, next_: NextCall) -> Mapping[str, Any]: return last async def innermost(_state: Any) -> Mapping[str, Any]: - inner_calls[0] += 1 - return {"trace": [f"call-{inner_calls[0]}"]} + nonlocal inner_calls + inner_calls += 1 + return {"trace": [f"call-{inner_calls}"]} chain = compose_chain([call_five_times], innermost) result = await chain(TraceState()) - assert inner_calls[0] == 5 + assert inner_calls == 5 # Final result is the partial returned from the LAST call to next. assert result == {"trace": ["call-5"]} diff --git a/tests/unit/test_observability_otel.py b/tests/unit/test_observability_otel.py new file mode 100644 index 00000000..24f0091f --- /dev/null +++ b/tests/unit/test_observability_otel.py @@ -0,0 +1,357 @@ +"""OTel-specific observability unit tests (extras-gated). + +Skipped cleanly when the ``otel`` extras aren't installed — the +import-time check in +``openarmature.observability.otel.__init__`` raises ImportError on +missing deps. + +These tests fill the gaps the conformance harness defers: + +- §6 TracerProvider isolation — the load-bearing "spans don't leak + into the OTel global provider" guarantee. +- §5 attribute population on every span type. +- §4.2 status mapping for every §4 error category. +- §5.5 LLM provider span via the ContextVar dispatch hook (queue- + mediated; no synchronous direct dispatch). +- §4.4 detached trace mode key separation in the span stack. +- §10.8 checkpoint_saved → ``openarmature.checkpoint.save`` zero- + duration span. +- §7 log bridge filter + correlation_id injection. +""" + +from __future__ import annotations + +import logging + +import pytest + +# Skip the entire module if otel extras aren't installed. +pytest.importorskip("opentelemetry.trace") + +from opentelemetry import trace as otel_trace +from opentelemetry.sdk.trace import TracerProvider +from opentelemetry.sdk.trace.export import SimpleSpanProcessor +from opentelemetry.sdk.trace.export.in_memory_span_exporter import ( + InMemorySpanExporter, +) + +from openarmature.checkpoint import InMemoryCheckpointer +from openarmature.graph import ( + END, + GraphBuilder, + NodeException, + State, +) +from openarmature.observability.otel import OTelObserver, install_log_bridge + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + + +class _LinearState(State): + a: int = 0 + b: int = 0 + + +async def _node_a(_s: _LinearState) -> dict[str, int]: + return {"a": 1} + + +async def _node_b(_s: _LinearState) -> dict[str, int]: + return {"b": 2} + + +def _build_linear_graph( + observer: OTelObserver | None = None, +) -> tuple[ + object, + InMemorySpanExporter, +]: + """Build a 2-node linear graph wired to a fresh OTelObserver + + in-memory exporter; returns (compiled_graph, exporter).""" + exporter = InMemorySpanExporter() + if observer is None: + observer = OTelObserver(span_processor=SimpleSpanProcessor(exporter)) + g = ( + GraphBuilder(_LinearState) + .add_node("node_a", _node_a) + .add_node("node_b", _node_b) + .add_edge("node_a", "node_b") + .add_edge("node_b", END) + .set_entry("node_a") + .compile() + ) + g.attach_observer(observer) + return g, exporter + + +# --------------------------------------------------------------------------- +# §6 TracerProvider isolation +# --------------------------------------------------------------------------- + + +async def test_observer_uses_private_provider_not_global() -> None: + """Spec §6 TracerProvider isolation: the OTelObserver MUST use a + PRIVATE TracerProvider; spans MUST NOT appear on the OTel global + provider's exporter (this is the load-bearing guarantee against + duplicate spans from external auto-instrumentation libraries).""" + # Save the prior global provider so we can restore it after the + # test — pytest fixture-scoping doesn't cover OTel global state. + prior_global = otel_trace.get_tracer_provider() + # Install a separate exporter on the OTel global provider. + global_exporter = InMemorySpanExporter() + global_provider = TracerProvider() + global_provider.add_span_processor(SimpleSpanProcessor(global_exporter)) + otel_trace.set_tracer_provider(global_provider) + + try: + # Drive a graph through OTelObserver. + private_exporter = InMemorySpanExporter() + observer = OTelObserver(span_processor=SimpleSpanProcessor(private_exporter)) + g, _ = _build_linear_graph(observer) + await g.invoke(_LinearState()) # type: ignore[attr-defined] + await g.drain() # type: ignore[attr-defined] + observer.shutdown() + + private_spans = private_exporter.get_finished_spans() + global_spans = global_exporter.get_finished_spans() + assert len(private_spans) > 0, "private provider must have received spans" + assert len(global_spans) == 0, ( + f"global provider MUST NOT receive openarmature spans; got {[s.name for s in global_spans]}" + ) + finally: + otel_trace.set_tracer_provider(prior_global) + + +# --------------------------------------------------------------------------- +# §5 attribute population +# --------------------------------------------------------------------------- + + +async def test_node_span_carries_required_attributes() -> None: + """Spec §5.2: every node span MUST carry the four + ``openarmature.node.*`` base attributes.""" + g, exporter = _build_linear_graph() + await g.invoke(_LinearState(), correlation_id="test-cid") # type: ignore[attr-defined] + await g.drain() # type: ignore[attr-defined] + spans = exporter.get_finished_spans() + node_spans = [s for s in spans if s.name in {"node_a", "node_b"}] + assert len(node_spans) == 2 + for span in node_spans: + attrs = dict(span.attributes or {}) + assert attrs.get("openarmature.node.name") == span.name + assert isinstance(attrs.get("openarmature.node.namespace"), tuple | list) + assert isinstance(attrs.get("openarmature.node.step"), int) + assert attrs.get("openarmature.node.attempt_index") == 0 + # Cross-cutting correlation_id (§5.6). + assert attrs.get("openarmature.correlation_id") == "test-cid" + + +async def test_invocation_span_carries_required_attributes() -> None: + """Spec §5.1: invocation span MUST carry + ``openarmature.graph.entry_node`` + ``openarmature.graph.spec_version``.""" + exporter = InMemorySpanExporter() + observer = OTelObserver(span_processor=SimpleSpanProcessor(exporter)) + g, _ = _build_linear_graph(observer) + await g.invoke(_LinearState()) # type: ignore[attr-defined] + await g.drain() # type: ignore[attr-defined] + # Invocation span closes on observer shutdown — the engine has + # no per-invocation lifecycle hook on observers, so the user + # closes the observer when done with their batch of invocations. + observer.shutdown() + spans = exporter.get_finished_spans() + inv = next((s for s in spans if s.name == "openarmature.invocation"), None) + assert inv is not None + attrs = dict(inv.attributes or {}) + assert attrs.get("openarmature.graph.entry_node") == "node_a" + assert isinstance(attrs.get("openarmature.graph.spec_version"), str) + + +# --------------------------------------------------------------------------- +# §4.2 status mapping +# --------------------------------------------------------------------------- + + +class _FailState(State): + a: int = 0 + + +async def _failing_node(_s: _FailState) -> dict[str, int]: + raise RuntimeError("boom") + + +async def test_failing_node_span_carries_error_status() -> None: + """Spec §4.2: a node-exception failure produces a span with + ERROR status, an exception event recorded, and the + ``openarmature.error.category`` attribute on the span.""" + from opentelemetry.trace import StatusCode + + exporter = InMemorySpanExporter() + observer = OTelObserver(span_processor=SimpleSpanProcessor(exporter)) + g = ( + GraphBuilder(_FailState) + .add_node("flaky", _failing_node) + .add_edge("flaky", END) + .set_entry("flaky") + .compile() + ) + g.attach_observer(observer) + with pytest.raises(NodeException): + await g.invoke(_FailState()) + await g.drain() + observer.shutdown() + spans = exporter.get_finished_spans() + flaky = next((s for s in spans if s.name == "flaky"), None) + assert flaky is not None + assert flaky.status.status_code == StatusCode.ERROR + attrs = dict(flaky.attributes or {}) + assert attrs.get("openarmature.error.category") == "node_exception" + + +# --------------------------------------------------------------------------- +# §10.8 checkpoint_saved → 0-duration span +# --------------------------------------------------------------------------- + + +async def test_checkpoint_save_emits_zero_duration_span() -> None: + """Spec §10.8: a checkpoint save SHOULD emit a §6-style observer + event surfaced as a span. Our implementation emits a + ``openarmature.checkpoint.save`` span on every save.""" + cp = InMemoryCheckpointer() + exporter = InMemorySpanExporter() + observer = OTelObserver(span_processor=SimpleSpanProcessor(exporter)) + g = ( + GraphBuilder(_LinearState) + .add_node("node_a", _node_a) + .add_edge("node_a", END) + .set_entry("node_a") + .with_checkpointer(cp) + .compile() + ) + # Subscribe to the checkpoint_saved phase (default subscription + # excludes it; OTelObserver attaches with the explicit set). + g.attach_observer(observer, phases={"started", "completed", "checkpoint_saved"}) + await g.invoke(_LinearState()) + await g.drain() + observer.shutdown() + spans = exporter.get_finished_spans() + save_spans = [s for s in spans if s.name == "openarmature.checkpoint.save"] + assert len(save_spans) == 1 + save_span = save_spans[0] + # Zero-duration: end_time - start_time near 0 (exact equality + # depends on monotonic clock resolution; permit small jitter). + end_t = save_span.end_time + start_t = save_span.start_time + assert end_t is not None and start_t is not None + duration = end_t - start_t + assert duration < 1_000_000, f"expected near-zero duration; got {duration}ns" + + +# --------------------------------------------------------------------------- +# §5.5 disable_llm_spans +# --------------------------------------------------------------------------- + + +async def test_disable_llm_spans_skips_llm_provider_span() -> None: + """Spec §5.5: ``disable_llm_spans=True`` MUST suppress the + LLM-provider span emission while leaving all other spans intact.""" + from openarmature.graph.events import NodeEvent + + # We don't drive a real provider here; instead we emit a synthetic + # LLM event through the observer's __call__ and assert no span was + # produced. This isolates the disable_llm_spans branch from the + # provider's own queue-dispatch wiring. + from openarmature.llm.providers.openai import _LlmEventState + + exporter = InMemorySpanExporter() + observer = OTelObserver( + span_processor=SimpleSpanProcessor(exporter), + disable_llm_spans=True, + ) + started = NodeEvent( + node_name="openarmature.llm.complete", + namespace=("openarmature.llm.complete",), + step=-1, + phase="started", + pre_state=_LlmEventState(model="test-m"), + post_state=None, + error=None, + parent_states=(), + ) + completed = NodeEvent( + node_name="openarmature.llm.complete", + namespace=("openarmature.llm.complete",), + step=-1, + phase="completed", + pre_state=_LlmEventState(model="test-m", finish_reason="stop"), + post_state=None, + error=None, + parent_states=(), + ) + await observer(started) + await observer(completed) + observer.shutdown() + llm_spans = [s for s in exporter.get_finished_spans() if s.name == "openarmature.llm.complete"] + assert llm_spans == [] + + +# --------------------------------------------------------------------------- +# §7 log bridge: correlation_id injection +# --------------------------------------------------------------------------- + + +def test_log_bridge_filter_injects_correlation_id() -> None: + """Spec §7: every log record emitted during an invocation MUST + carry ``openarmature.correlation_id``. The bridge's filter reads + the ContextVar and attaches it to the LogRecord.""" + from openarmature.observability.correlation import ( + _reset_correlation_id, + _set_correlation_id, + ) + from openarmature.observability.otel.logs import _CorrelationIdFilter + + flt = _CorrelationIdFilter() + record = logging.LogRecord( + name="test", + level=logging.INFO, + pathname="", + lineno=0, + msg="hello", + args=None, + exc_info=None, + ) + # Outside an invocation: no correlation_id attribute set. + assert flt.filter(record) is True + assert not hasattr(record, "openarmature.correlation_id") + + # Inside an invocation: filter attaches the ContextVar value. + record2 = logging.LogRecord( + name="test", + level=logging.INFO, + pathname="", + lineno=0, + msg="hello", + args=None, + exc_info=None, + ) + token = _set_correlation_id("my-cid-42") + try: + flt.filter(record2) + finally: + _reset_correlation_id(token) + assert getattr(record2, "openarmature.correlation_id") == "my-cid-42" + + +def test_install_log_bridge_is_idempotent() -> None: + """Re-calling :func:`install_log_bridge` MUST NOT register a + duplicate handler — the bridge owns the only OA-flagged + LoggingHandler on the root logger.""" + from opentelemetry.sdk._logs import LoggerProvider + + provider = LoggerProvider() + install_log_bridge(provider) + handler_count_before = len(logging.getLogger().handlers) + install_log_bridge(provider) + handler_count_after = len(logging.getLogger().handlers) + assert handler_count_before == handler_count_after diff --git a/uv.lock b/uv.lock index 80f411f3..dc697ff6 100644 --- a/uv.lock +++ b/uv.lock @@ -124,6 +124,18 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/5d/13/ad7d7ca3808a898b4612b6fe93cde56b53f3034dcde235acb1f0e1df24c6/idna-3.13-py3-none-any.whl", hash = "sha256:892ea0cde124a99ce773decba204c5552b69c3c67ffd5f232eb7696135bc8bb3", size = 68629, upload-time = "2026-04-22T16:42:40.909Z" }, ] +[[package]] +name = "importlib-metadata" +version = "8.7.1" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "zipp" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/f3/49/3b30cad09e7771a4982d9975a8cbf64f00d4a1ececb53297f1d9a7be1b10/importlib_metadata-8.7.1.tar.gz", hash = "sha256:49fef1ae6440c182052f407c8d34a68f72efc36db9ca90dc0113398f2fdde8bb", size = 57107, upload-time = "2025-12-21T10:00:19.278Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/fa/5e/f8e9a1d23b9c20a551a8a02ea3637b4642e22c2626e3a13a9a29cdea99eb/importlib_metadata-8.7.1-py3-none-any.whl", hash = "sha256:5a1f80bf1daa489495071efbb095d75a634cf28a8bc299581244063b53176151", size = 27865, upload-time = "2025-12-21T10:00:18.329Z" }, +] + [[package]] name = "iniconfig" version = "2.3.0" @@ -151,6 +163,12 @@ dependencies = [ { name = "pydantic" }, ] +[package.optional-dependencies] +otel = [ + { name = "opentelemetry-api" }, + { name = "opentelemetry-sdk" }, +] + [package.dev-dependencies] dev = [ { name = "pre-commit" }, @@ -165,8 +183,11 @@ dev = [ [package.metadata] requires-dist = [ { name = "httpx", specifier = ">=0.27" }, + { name = "opentelemetry-api", marker = "extra == 'otel'", specifier = ">=1.27,<2" }, + { name = "opentelemetry-sdk", marker = "extra == 'otel'", specifier = ">=1.27,<2" }, { name = "pydantic", specifier = ">=2.7" }, ] +provides-extras = ["otel"] [package.metadata.requires-dev] dev = [ @@ -179,6 +200,46 @@ dev = [ { name = "types-pyyaml" }, ] +[[package]] +name = "opentelemetry-api" +version = "1.41.1" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "importlib-metadata" }, + { name = "typing-extensions" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/fa/fc/b7564cbef36601aef0d6c9bc01f7badb64be8e862c2e1c3c5c3b43b53e4f/opentelemetry_api-1.41.1.tar.gz", hash = "sha256:0ad1814d73b875f84494387dae86ce0b12c68556331ce6ce8fe789197c949621", size = 71416, upload-time = "2026-04-24T13:15:38.262Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/29/59/3e7118ed140f76b0982ba4321bdaed1997a0473f9720de2d10788a577033/opentelemetry_api-1.41.1-py3-none-any.whl", hash = "sha256:a22df900e75c76dc08440710e51f52f1aa6b451b429298896023e60db5b3139f", size = 69007, upload-time = "2026-04-24T13:15:15.662Z" }, +] + +[[package]] +name = "opentelemetry-sdk" +version = "1.41.1" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "opentelemetry-api" }, + { name = "opentelemetry-semantic-conventions" }, + { name = "typing-extensions" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/58/d0/54ee30dab82fb0acda23d144502771ff76ef8728459c83c3e89ef9fb1825/opentelemetry_sdk-1.41.1.tar.gz", hash = "sha256:724b615e1215b5aeacda0abb8a6a8922c9a1853068948bd0bd225a56d0c792e6", size = 230180, upload-time = "2026-04-24T13:15:50.991Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/b4/e7/a1420b698aad018e1cf60fdbaaccbe49021fb415e2a0d81c242f4c518f54/opentelemetry_sdk-1.41.1-py3-none-any.whl", hash = "sha256:edee379c126c1bce952b0c812b48fe8ff35b30df0eecf17e98afa4d598b7d85d", size = 180213, upload-time = "2026-04-24T13:15:33.767Z" }, +] + +[[package]] +name = "opentelemetry-semantic-conventions" +version = "0.62b1" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "opentelemetry-api" }, + { name = "typing-extensions" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/9e/de/911ac9e309052aca1b20b2d5549d3db45d1011e1a610e552c6ccdd1b64f8/opentelemetry_semantic_conventions-0.62b1.tar.gz", hash = "sha256:c5cc6e04a7f8c7cdd30be2ed81499fa4e75bfbd52c9cb70d40af1f9cd3619802", size = 145750, upload-time = "2026-04-24T13:15:52.236Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/eb/a6/83dc2ab6fa397ee66fba04fe2e74bdf7be3b3870005359ceb7689103c058/opentelemetry_semantic_conventions-0.62b1-py3-none-any.whl", hash = "sha256:cf506938103d331fbb78eded0d9788095f7fd59016f2bda813c3324e5a74a93c", size = 231620, upload-time = "2026-04-24T13:15:35.454Z" }, +] + [[package]] name = "packaging" version = "26.1" @@ -491,3 +552,12 @@ sdist = { url = "https://files.pythonhosted.org/packages/0c/98/3a7e644e19cb26133 wheels = [ { url = "https://files.pythonhosted.org/packages/27/8d/edd0bd910ff803c308ee9a6b7778621af0d10252219ad9f19ef4d4982a61/virtualenv-21.2.4-py3-none-any.whl", hash = "sha256:29d21e941795206138d0f22f4e45ff7050e5da6c6472299fb7103318763861ac", size = 5831232, upload-time = "2026-04-14T22:15:29.342Z" }, ] + +[[package]] +name = "zipp" +version = "3.23.1" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/30/21/093488dfc7cc8964ded15ab726fad40f25fd3d788fd741cc1c5a17d78ee8/zipp-3.23.1.tar.gz", hash = "sha256:32120e378d32cd9714ad503c1d024619063ec28aad2248dc6672ad13edfa5110", size = 25965, upload-time = "2026-04-13T23:21:46.6Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/08/8a/0861bec20485572fbddf3dfba2910e38fe249796cb73ecdeb74e07eeb8d3/zipp-3.23.1-py3-none-any.whl", hash = "sha256:0b3596c50a5c700c9cb40ba8d86d9f2cc4807e9bedb06bcdf7fac85633e444dc", size = 10378, upload-time = "2026-04-13T23:21:45.386Z" }, +]