diff --git a/src/openarmature/graph/__init__.py b/src/openarmature/graph/__init__.py index 4cd13453..af55d4f4 100644 --- a/src/openarmature/graph/__init__.py +++ b/src/openarmature/graph/__init__.py @@ -38,6 +38,7 @@ from .events import ( InvocationCompletedEvent, InvocationStartedEvent, + LlmCompletionEvent, MetadataAugmentationEvent, NodeEvent, ) @@ -84,6 +85,7 @@ "GraphError", "InvocationCompletedEvent", "InvocationStartedEvent", + "LlmCompletionEvent", "MappingReferencesUndeclaredField", "MetadataAugmentationEvent", "Middleware", diff --git a/src/openarmature/graph/events.py b/src/openarmature/graph/events.py index aa145ca4..bb6596f5 100644 --- a/src/openarmature/graph/events.py +++ b/src/openarmature/graph/events.py @@ -16,13 +16,21 @@ from collections.abc import Mapping from dataclasses import dataclass, field from types import MappingProxyType -from typing import Any, Literal +from typing import TYPE_CHECKING, Any, Literal from openarmature.observability.metadata import AttributeValue from .errors import RuntimeGraphError from .state import State +# TYPE_CHECKING import — the runtime Usage class lives in the llm +# package, which transitively imports from graph.events (the +# OpenAI provider imports NodeEvent). Using a TYPE_CHECKING import +# plus a string annotation on LlmCompletionEvent.usage avoids the +# circular runtime import while keeping pyright type-safe. +if TYPE_CHECKING: + from openarmature.llm.response import Usage + # Sentinel empty metadata mapping for events constructed without a # live caller-metadata snapshot (test helpers, synthetic events). # Read-only proxy keeps the default allocation-free. @@ -430,10 +438,82 @@ class InvocationCompletedEvent: correlation_id: str | None +# Spec: realizes proposal 0049's first spec-normatively-typed event +# variant on the observer event union (graph-engine §6 + +# observability §5.5.7). Dispatched on every LLM provider call that +# returns a structured response, alongside the calling node's +# NodeEvent pair. Failure cases (provider exceptions, malformed +# responses) flow through the existing exception path and do NOT +# emit this variant. Not subject to the §6 ``phases`` subscription +# filter (matches MetadataAugmentationEvent / InvocationStartedEvent +# / InvocationCompletedEvent treatment). +# +# Field naming matches the spec-canonical names verbatim per the spec +# Q5 ack — Python snake_case happens to match the spec table 1:1. +@dataclass(frozen=True) +class LlmCompletionEvent: + """A typed LLM provider call event delivered to observers. + + Carries identity, scoping, and outcome data for an LLM call as + structured fields. Observer code filters by type discrimination + (``isinstance(event, LlmCompletionEvent)``) rather than by the + impl-current sentinel-namespace string match the legacy + NodeEvent pattern uses. + + Field set: + + - ``invocation_id``: the outer invocation's identifier. + - ``correlation_id``: cross-backend correlation id when present. + - ``node_name``: the user-defined node that issued the call. + - ``namespace``: the calling node's namespace tuple (NOT the + legacy sentinel namespace). + - ``attempt_index``: retry-attempt index (0 on first attempt). + - ``fan_out_index``: fan-out instance index when the calling + node ran inside a fan-out instance; ``None`` otherwise. + - ``branch_name``: parallel-branches branch name when the + calling node ran inside a branch; ``None`` otherwise. + - ``provider``: provider identifier; matches ``gen_ai.system``. + - ``model``: the model identifier the call targeted. + - ``request_id``: provider-returned response id; ``None`` when + the provider didn't return one. + - ``usage``: token-accounting record per ``Response.usage`` + shape. Reuses the existing ``openarmature.llm.response.Usage`` + class. ``None`` when the call returned no usage at all. + - ``latency_ms``: wall-clock latency measured at the adapter + boundary, in milliseconds. ``None`` when latency was not + measured. + - ``finish_reason``: the call's finish reason; ``None`` when + the call did not complete normally. + - ``caller_invocation_metadata``: optional snapshot of caller- + supplied invocation metadata at LLM-call time. Populated + only when the provider's opt-in flag is set (per-language + mechanism); default ``None``. + """ + + invocation_id: str + correlation_id: str | None + node_name: str + namespace: tuple[str, ...] + attempt_index: int + fan_out_index: int | None + branch_name: str | None + provider: str + model: str + request_id: str | None + # Usage is a string-typed forward reference per the TYPE_CHECKING + # import above — keeps the runtime import direction graph → llm + # off the module-load path while preserving pyright resolution. + usage: "Usage | None" + latency_ms: float | None + finish_reason: str | None + caller_invocation_metadata: Mapping[str, AttributeValue] | None = None + + __all__ = [ "FanOutEventConfig", "InvocationCompletedEvent", "InvocationStartedEvent", + "LlmCompletionEvent", "MetadataAugmentationEvent", "NodeEvent", "ParallelBranchesEventConfig", diff --git a/src/openarmature/graph/observer.py b/src/openarmature/graph/observer.py index 66cf988c..9a82ce01 100644 --- a/src/openarmature/graph/observer.py +++ b/src/openarmature/graph/observer.py @@ -37,17 +37,28 @@ from .events import ( InvocationCompletedEvent, InvocationStartedEvent, + LlmCompletionEvent, MetadataAugmentationEvent, NodeEvent, ) from .state import State # Union of every event variant an Observer may receive. NodeEvent is -# the original §6 started/completed/checkpoint shape; the other three -# are side-channel events (proposal 0040 for augmentation; proposal -# 0043 for invocation-boundary trace.input/output sourcing) that -# bypass the phase filter and reach every subscribed observer. -ObserverEvent = NodeEvent | MetadataAugmentationEvent | InvocationStartedEvent | InvocationCompletedEvent +# the original §6 started/completed/checkpoint shape; the other +# variants are side-channel events that bypass the phase filter and +# reach every subscribed observer — MetadataAugmentationEvent +# (proposal 0040 mid-invocation metadata augmentation), +# InvocationStartedEvent / InvocationCompletedEvent (proposal 0043 +# trace.input/output sourcing), and LlmCompletionEvent (proposal +# 0049 typed LLM provider call event, dispatched on every successful +# LLM completion alongside the calling node's NodeEvent pair). +ObserverEvent = ( + NodeEvent + | MetadataAugmentationEvent + | InvocationStartedEvent + | InvocationCompletedEvent + | LlmCompletionEvent +) class Observer(Protocol): diff --git a/src/openarmature/llm/providers/openai.py b/src/openarmature/llm/providers/openai.py index abc71e00..3ace84f6 100644 --- a/src/openarmature/llm/providers/openai.py +++ b/src/openarmature/llm/providers/openai.py @@ -53,8 +53,9 @@ import hashlib import json import re +import time import uuid -from collections.abc import Sequence +from collections.abc import Mapping, Sequence from typing import Any, Literal, cast from urllib.parse import urlparse @@ -62,16 +63,18 @@ import jsonschema from pydantic import BaseModel, ValidationError -from openarmature.graph.events import NodeEvent +from openarmature.graph.events import LlmCompletionEvent, NodeEvent from openarmature.observability.correlation import ( current_attempt_index, current_branch_name, + current_correlation_id, current_dispatch, current_fan_out_index, + current_invocation_id, current_namespace_prefix, ) from openarmature.observability.llm_event import LlmEventPayload -from openarmature.observability.metadata import current_invocation_metadata +from openarmature.observability.metadata import AttributeValue, current_invocation_metadata # ``current_prompt_group`` / ``current_prompt_result`` are imported # lazily inside :meth:`OpenAIProvider.complete` to avoid a module-load @@ -157,6 +160,7 @@ def __init__( force_prompt_augmentation_fallback: bool = False, genai_system: str = "openai", readiness_probe: Literal["models", "chat_completions", "both"] = "chat_completions", + populate_caller_metadata: bool = False, ) -> None: self.base_url = _validate_and_normalize_base_url(base_url) self.model = model @@ -189,6 +193,14 @@ def __init__( f"readiness_probe must be one of {sorted(_VALID_READINESS_PROBES)} (got {readiness_probe!r})" ) self._readiness_probe = readiness_probe + # Proposal 0049's caller_invocation_metadata field is OPTIONAL + # on the typed LlmCompletionEvent: default absent, populated + # only when the consumer opts in. The per-language opt-in + # mechanism is constructor-knob here so the provider can decide + # at emission time without engine-level observer introspection. + # Off by default to avoid bloating every event with potentially- + # large metadata snapshots when nothing downstream consumes them. + self._populate_caller_metadata = populate_caller_metadata self._headers: dict[str, str] = {"Content-Type": "application/json"} if api_key is not None: self._headers["Authorization"] = f"Bearer {api_key}" @@ -443,10 +455,30 @@ async def complete( ) ) + # Wall-clock latency measured at the adapter boundary per + # proposal 0049's LlmCompletionEvent.latency_ms contract. The + # boundary spans from "just before _do_complete is called" to + # "_do_complete returns with a parsed Response in hand" — + # covers HTTP setup, request emission, provider compute, + # response receive, AND response parsing into the typed + # Response. The spec text "wall-clock latency of the LLM call + # measured at the adapter boundary" is silent on whether + # parsing is included; including it matches the operator's + # mental model of "how long until I had a usable answer" + # better than just-the-HTTP-call. perf_counter is the monotonic + # high-resolution clock for elapsed-time measurements. + adapter_start = time.perf_counter() try: response = await self._do_complete(body, schema_dict, schema_class) except Exception as exc: if dispatch is not None: + # Failure path: only the sentinel NodeEvent pair fires. + # Per proposal 0049 §3 (alternative 3): LlmCompletionEvent + # is completion-only; failures flow through the + # llm-provider §7 exception path. The error continues + # to surface through the existing observer chain via + # the sentinel NodeEvent's error_type / error_category + # fields on LlmEventPayload. dispatch( _make_llm_event( "completed", @@ -462,8 +494,14 @@ async def complete( ) ) raise + latency_ms = (time.perf_counter() - adapter_start) * 1000.0 if dispatch is not None: + # Sentinel NodeEvent pair stays during the dual-emit window + # per proposal 0049 §5.5.7 SHOULD-emit-both transition. The + # window stays open through v0.13.0 with the sentinel + # emission removed in v0.15.0 (CHANGELOG callout pinned to + # the v0.13.0 release notes). dispatch( _make_llm_event( "completed", @@ -482,8 +520,63 @@ async def complete( active_prompt_group=active_prompt_group, ) ) + # The new typed LlmCompletionEvent — observers filtering via + # isinstance(event, LlmCompletionEvent) receive this; legacy + # observers filtering on the sentinel namespace see the + # NodeEvent pair above. Failure path doesn't reach here. + dispatch( + self._build_llm_completion_event(response, latency_ms), + ) return response + def _build_llm_completion_event(self, response: Response, latency_ms: float) -> LlmCompletionEvent: + """Construct the typed LlmCompletionEvent for the success path. + + Sources identity / scoping fields from the calling-node + ContextVars and outcome fields from the response. The calling- + node namespace is the FULL namespace tuple (not the legacy + sentinel pseudo-namespace); node_name is the last element of + the namespace (the user-defined node that issued the call). + Outside any node body (namespace empty), node_name is the + empty string. + """ + + namespace = current_namespace_prefix() + node_name = namespace[-1] if namespace else "" + # invocation_id is normally always present once invoke() entry + # has run, but the LLM provider can be exercised in test + # fixtures outside an invocation. Spec proposal 0049's field + # table types invocation_id as a non-nullable string, so we + # fall back to empty string rather than None to keep the event + # constructable. Downstream observers using invocation_id as a + # correlation key should treat "" as "not in an invocation" + # and either skip or special-case those events; collisions + # across multiple out-of-invocation calls are theoretically + # possible but not a path production code should hit. + invocation_id = current_invocation_id() or "" + caller_metadata: Mapping[str, AttributeValue] | None = None + if self._populate_caller_metadata: + # Snapshot via dict() so downstream consumers see a stable + # frozen view; if a node body mutates metadata after the + # snapshot, the event still carries the at-emission view. + caller_metadata = dict(current_invocation_metadata()) + return LlmCompletionEvent( + invocation_id=invocation_id, + correlation_id=current_correlation_id(), + node_name=node_name, + namespace=namespace, + attempt_index=current_attempt_index(), + fan_out_index=current_fan_out_index(), + branch_name=current_branch_name(), + provider=self._genai_system, + model=self.model, + request_id=response.response_id, + usage=response.usage, + latency_ms=latency_ms, + finish_reason=response.finish_reason, + caller_invocation_metadata=caller_metadata, + ) + async def _do_complete( self, body: dict[str, Any], diff --git a/src/openarmature/observability/correlation.py b/src/openarmature/observability/correlation.py index 20174426..2967bc44 100644 --- a/src/openarmature/observability/correlation.py +++ b/src/openarmature/observability/correlation.py @@ -39,6 +39,7 @@ from openarmature.graph.events import ( InvocationCompletedEvent, InvocationStartedEvent, + LlmCompletionEvent, MetadataAugmentationEvent, NodeEvent, ) @@ -215,7 +216,13 @@ def _reset_active_observers(token: Token[tuple[SubscribedObserver, ...]]) -> Non _active_dispatch_var: ContextVar[ Callable[ - [NodeEvent | MetadataAugmentationEvent | InvocationStartedEvent | InvocationCompletedEvent], + [ + NodeEvent + | MetadataAugmentationEvent + | InvocationStartedEvent + | InvocationCompletedEvent + | LlmCompletionEvent + ], None, ] | None @@ -224,7 +231,13 @@ def _reset_active_observers(token: Token[tuple[SubscribedObserver, ...]]) -> Non def current_dispatch() -> ( Callable[ - [NodeEvent | MetadataAugmentationEvent | InvocationStartedEvent | InvocationCompletedEvent], + [ + NodeEvent + | MetadataAugmentationEvent + | InvocationStartedEvent + | InvocationCompletedEvent + | LlmCompletionEvent + ], None, ] | None @@ -244,12 +257,24 @@ def current_dispatch() -> ( def _set_active_dispatch( dispatch: Callable[ - [NodeEvent | MetadataAugmentationEvent | InvocationStartedEvent | InvocationCompletedEvent], + [ + NodeEvent + | MetadataAugmentationEvent + | InvocationStartedEvent + | InvocationCompletedEvent + | LlmCompletionEvent + ], None, ], ) -> Token[ Callable[ - [NodeEvent | MetadataAugmentationEvent | InvocationStartedEvent | InvocationCompletedEvent], + [ + NodeEvent + | MetadataAugmentationEvent + | InvocationStartedEvent + | InvocationCompletedEvent + | LlmCompletionEvent + ], None, ] | None @@ -262,7 +287,13 @@ def _set_active_dispatch( def _reset_active_dispatch( token: Token[ Callable[ - [NodeEvent | MetadataAugmentationEvent | InvocationStartedEvent | InvocationCompletedEvent], + [ + NodeEvent + | MetadataAugmentationEvent + | InvocationStartedEvent + | InvocationCompletedEvent + | LlmCompletionEvent + ], None, ] | None diff --git a/src/openarmature/observability/langfuse/observer.py b/src/openarmature/observability/langfuse/observer.py index 6736ae9d..55016ce0 100644 --- a/src/openarmature/observability/langfuse/observer.py +++ b/src/openarmature/observability/langfuse/observer.py @@ -31,6 +31,7 @@ from openarmature.graph.events import ( InvocationCompletedEvent, InvocationStartedEvent, + LlmCompletionEvent, MetadataAugmentationEvent, NodeEvent, ) @@ -351,7 +352,13 @@ def __post_init__(self) -> None: async def __call__( self, - event: (NodeEvent | MetadataAugmentationEvent | InvocationStartedEvent | InvocationCompletedEvent), + event: ( + NodeEvent + | MetadataAugmentationEvent + | InvocationStartedEvent + | InvocationCompletedEvent + | LlmCompletionEvent + ), ) -> None: if isinstance(event, InvocationStartedEvent): self._handle_invocation_started(event) @@ -359,6 +366,15 @@ async def __call__( if isinstance(event, InvocationCompletedEvent): self._handle_invocation_completed(event) return + # Proposal 0049 typed LlmCompletionEvent: ignored during the + # dual-emit window — the Langfuse mapping continues to drive + # its §5.5 Generation observation lifecycle off the sentinel + # NodeEvent pair the provider emits alongside the typed event. + # Migration to type discrimination lands in a subsequent PR; + # this early-return keeps the observer Protocol-compatible + # without changing behavior. + if isinstance(event, LlmCompletionEvent): + return if isinstance(event, MetadataAugmentationEvent): self._handle_metadata_augmentation(event) return diff --git a/src/openarmature/observability/otel/observer.py b/src/openarmature/observability/otel/observer.py index 1c0ef1a0..781dfbb1 100644 --- a/src/openarmature/observability/otel/observer.py +++ b/src/openarmature/observability/otel/observer.py @@ -99,6 +99,7 @@ from openarmature.graph.events import ( InvocationCompletedEvent, InvocationStartedEvent, + LlmCompletionEvent, MetadataAugmentationEvent, NodeEvent, ) @@ -567,7 +568,13 @@ def _inv_state_for(self, invocation_id: str) -> _InvState: async def __call__( self, - event: (NodeEvent | MetadataAugmentationEvent | InvocationStartedEvent | InvocationCompletedEvent), + event: ( + NodeEvent + | MetadataAugmentationEvent + | InvocationStartedEvent + | InvocationCompletedEvent + | LlmCompletionEvent + ), ) -> None: # Proposal 0043 invocation-boundary events: OTel has no # Trace-level input/output payload concept (a trace is a @@ -576,6 +583,15 @@ async def __call__( # before any node-specific logic runs. if isinstance(event, InvocationStartedEvent | InvocationCompletedEvent): return + # Proposal 0049 typed LlmCompletionEvent: ignored during the + # dual-emit window — the OTel mapping continues to drive its + # §5.5 LLM span lifecycle off the sentinel NodeEvent pair the + # provider emits alongside the typed event. Migration to type + # discrimination lands in a subsequent PR; this early-return + # keeps the observer Protocol-compatible without changing + # behavior. + if isinstance(event, LlmCompletionEvent): + return if isinstance(event, MetadataAugmentationEvent): self._handle_metadata_augmentation(event) return diff --git a/tests/unit/test_llm_provider.py b/tests/unit/test_llm_provider.py index 55eeffab..469480a5 100644 --- a/tests/unit/test_llm_provider.py +++ b/tests/unit/test_llm_provider.py @@ -11,10 +11,15 @@ from __future__ import annotations +from collections.abc import Callable +from contextvars import Token + import httpx import pytest from pydantic import ValidationError +from openarmature.graph.events import LlmCompletionEvent, NodeEvent +from openarmature.graph.observer import ObserverEvent from openarmature.llm import ( PROVIDER_AUTHENTICATION, PROVIDER_INVALID_MODEL, @@ -44,6 +49,18 @@ validate_message_list, validate_tools, ) +from openarmature.observability.correlation import ( + _set_active_dispatch, + _set_attempt_index, + _set_branch_name, + _set_correlation_id, + _set_fan_out_index, + _set_invocation_id, + _set_namespace_prefix, +) +from openarmature.observability.metadata import set_invocation_metadata + +_DispatchToken = Token[Callable[[ObserverEvent], None] | None] # --------------------------------------------------------------------------- # Per-role message construction (Pydantic-on-Message layer) @@ -1178,3 +1195,290 @@ def _503(_req: httpx.Request) -> httpx.Response: await provider.ready() finally: await provider.aclose() + + +# --------------------------------------------------------------------------- +# LlmCompletionEvent dual-emit (proposal 0049) +# --------------------------------------------------------------------------- + + +def _collecting_dispatch() -> tuple[list[ObserverEvent], _DispatchToken]: + """Install a collecting dispatch callback into the + ``current_dispatch`` ContextVar and return ``(events, token)``. + The caller is responsible for resetting the token in a try/finally. + """ + events: list[ObserverEvent] = [] + + def _dispatch(event: ObserverEvent) -> None: + events.append(event) + + token = _set_active_dispatch(_dispatch) + return events, token + + +def _release_dispatch(token: _DispatchToken) -> None: + from openarmature.observability.correlation import _reset_active_dispatch + + _reset_active_dispatch(token) + + +async def test_complete_success_emits_both_sentinel_and_typed_event() -> None: + # Dual-emit window per proposal 0049 §5.5.7 SHOULD-emit-both + # transition: the provider emits BOTH the existing sentinel + # NodeEvent pair (started + completed) AND the new typed + # LlmCompletionEvent on success. + events, token = _collecting_dispatch() + transport = _make_openai_response_with_usage( + {"prompt_tokens": 10, "completion_tokens": 5, "total_tokens": 15} + ) + provider = OpenAIProvider(base_url="http://test", model="m", api_key="k", transport=transport) + try: + await provider.complete([UserMessage(content="hi")]) + finally: + await provider.aclose() + _release_dispatch(token) + + node_events = [e for e in events if isinstance(e, NodeEvent)] + typed_events = [e for e in events if isinstance(e, LlmCompletionEvent)] + # Sentinel pair: started + completed (the existing pattern). + assert len(node_events) == 2 + assert [e.phase for e in node_events] == ["started", "completed"] + # One typed event (success-only emission). + assert len(typed_events) == 1 + + +async def test_complete_failure_emits_only_sentinel_no_typed_event() -> None: + # Per proposal 0049 §3 alternative 3: LlmCompletionEvent fires on + # successful structured-response completion only. Provider + # exceptions (provider_unavailable etc.) flow through the + # existing exception path; the sentinel NodeEvent(completed, + # error=...) keeps firing for backwards-compat failure + # observability. + from openarmature.graph.events import LlmCompletionEvent, NodeEvent + + def _503(_req: httpx.Request) -> httpx.Response: + return httpx.Response(503, json={"error": {"message": "down"}}) + + events, token = _collecting_dispatch() + provider = OpenAIProvider( + base_url="http://test", model="m", api_key="k", transport=httpx.MockTransport(_503) + ) + try: + with pytest.raises(ProviderUnavailable): + await provider.complete([UserMessage(content="hi")]) + finally: + await provider.aclose() + _release_dispatch(token) + + node_events = [e for e in events if isinstance(e, NodeEvent)] + typed_events = [e for e in events if isinstance(e, LlmCompletionEvent)] + # Sentinel pair fires; the completed event carries error details + # on its LlmEventPayload. + assert len(node_events) == 2 + assert [e.phase for e in node_events] == ["started", "completed"] + # No typed event on the failure path. + assert typed_events == [] + + +async def test_llm_completion_event_carries_typed_outcome_fields() -> None: + # Field sourcing: provider / model / usage / request_id / finish_reason + # / latency_ms are all populated from the response + instance state. + events, token = _collecting_dispatch() + transport = _make_openai_response_with_usage( + { + "prompt_tokens": 100, + "completion_tokens": 20, + "total_tokens": 120, + "prompt_tokens_details": {"cached_tokens": 50}, + } + ) + provider = OpenAIProvider( + base_url="http://test", model="m-test", api_key="k", transport=transport, genai_system="vllm" + ) + try: + await provider.complete([UserMessage(content="hi")]) + finally: + await provider.aclose() + _release_dispatch(token) + + typed_events = [e for e in events if isinstance(e, LlmCompletionEvent)] + assert len(typed_events) == 1 + typed = typed_events[0] + assert typed.provider == "vllm" + assert typed.model == "m-test" + assert typed.finish_reason == "stop" + assert typed.request_id == "x" # the helper returns id="x" + # usage flows through the shared Usage shape; cache field surfaces + # via the typed event without separate plumbing per the + # proposal-0047 + proposal-0049 architectural pair. + assert typed.usage is not None + assert typed.usage.cached_tokens == 50 + assert typed.latency_ms is not None + assert typed.latency_ms >= 0.0 + + +async def test_caller_invocation_metadata_off_by_default() -> None: + # Per proposal 0049's OPT-IN contract: default absent / None. + events, token = _collecting_dispatch() + transport = _make_openai_response_with_usage( + {"prompt_tokens": 10, "completion_tokens": 5, "total_tokens": 15} + ) + provider = OpenAIProvider(base_url="http://test", model="m", api_key="k", transport=transport) + try: + await provider.complete([UserMessage(content="hi")]) + finally: + await provider.aclose() + _release_dispatch(token) + + typed = next(e for e in events if isinstance(e, LlmCompletionEvent)) + assert typed.caller_invocation_metadata is None + + +async def test_caller_invocation_metadata_populated_when_opted_in() -> None: + # Per proposal 0049 Q2 ack: opt-in via provider constructor knob. + # When True, the typed event carries a snapshot of the metadata + # mapping at LLM-call time. + events, token = _collecting_dispatch() + transport = _make_openai_response_with_usage( + {"prompt_tokens": 10, "completion_tokens": 5, "total_tokens": 15} + ) + provider = OpenAIProvider( + base_url="http://test", + model="m", + api_key="k", + transport=transport, + populate_caller_metadata=True, + ) + try: + set_invocation_metadata(user_id="u-123") + await provider.complete([UserMessage(content="hi")]) + finally: + await provider.aclose() + _release_dispatch(token) + + typed = next(e for e in events if isinstance(e, LlmCompletionEvent)) + assert typed.caller_invocation_metadata is not None + assert typed.caller_invocation_metadata.get("user_id") == "u-123" + + +async def test_llm_completion_event_request_id_none_when_response_omits_id() -> None: + # Spec proposal 0049: request_id is the provider-returned response + # id when present; None otherwise. Pin the None case explicitly. + def _handler(_req: httpx.Request) -> httpx.Response: + return httpx.Response( + 200, + json={ + # Note: no "id" field on the response body. + "object": "chat.completion", + "created": 0, + "model": "m", + "choices": [ + { + "index": 0, + "message": {"role": "assistant", "content": "ok"}, + "finish_reason": "stop", + } + ], + "usage": {"prompt_tokens": 1, "completion_tokens": 1, "total_tokens": 2}, + }, + ) + + events, token = _collecting_dispatch() + provider = OpenAIProvider( + base_url="http://test", model="m", api_key="k", transport=httpx.MockTransport(_handler) + ) + try: + await provider.complete([UserMessage(content="hi")]) + finally: + await provider.aclose() + _release_dispatch(token) + + typed = next(e for e in events if isinstance(e, LlmCompletionEvent)) + assert typed.request_id is None + + +async def test_llm_completion_event_arrives_after_sentinel_completed_within_provider_emission() -> None: + # Within the provider's own emission window (sentinel started → + # sentinel completed → typed event), the typed LlmCompletionEvent + # MUST arrive after the sentinel NodeEvent(completed). Spec + # fixture 056 pins the broader contract (typed event arrives + # between the CALLING NODE's started/completed pair); this test + # locks down the provider-internal sub-ordering that contract + # depends on. The full bracketing (calling-node started → ... → + # calling-node completed) is covered by the conformance fixture + # which exercises a real CompiledGraph. + events, token = _collecting_dispatch() + transport = _make_openai_response_with_usage( + {"prompt_tokens": 1, "completion_tokens": 1, "total_tokens": 2} + ) + provider = OpenAIProvider(base_url="http://test", model="m", api_key="k", transport=transport) + try: + await provider.complete([UserMessage(content="hi")]) + finally: + await provider.aclose() + _release_dispatch(token) + + # Build a type+phase tag for each event in arrival order. + sequence: list[str] = [] + for ev in events: + if isinstance(ev, NodeEvent): + sequence.append(f"sentinel:{ev.phase}") + elif isinstance(ev, LlmCompletionEvent): + sequence.append("typed:completed") + assert sequence == ["sentinel:started", "sentinel:completed", "typed:completed"] + + +async def test_llm_completion_event_sources_node_identity_from_calling_context() -> None: + # When the provider is called inside a node body, the typed event + # sources node_name / namespace / attempt_index / fan_out_index / + # branch_name from the calling-node ContextVars. The five tests + # above all ran outside any node body (default empty/None values); + # this test installs the ContextVars manually to confirm the + # sourcing path actually reaches the typed event. + events, token = _collecting_dispatch() + namespace_token = _set_namespace_prefix(("outer", "scoring")) + attempt_token = _set_attempt_index(2) + fan_out_token = _set_fan_out_index(3) + branch_token = _set_branch_name("fast") + invocation_token = _set_invocation_id("inv-abc") + correlation_token = _set_correlation_id("corr-xyz") + transport = _make_openai_response_with_usage( + {"prompt_tokens": 1, "completion_tokens": 1, "total_tokens": 2} + ) + provider = OpenAIProvider(base_url="http://test", model="m", api_key="k", transport=transport) + try: + await provider.complete([UserMessage(content="hi")]) + finally: + await provider.aclose() + _release_dispatch(token) + # Reset the calling-node ContextVars in the same order would + # work but the engine's normal teardown handles this in + # production; for the test, the order doesn't matter since + # we're at end-of-scope. + from openarmature.observability.correlation import ( + _reset_attempt_index, + _reset_branch_name, + _reset_correlation_id, + _reset_fan_out_index, + _reset_invocation_id, + _reset_namespace_prefix, + ) + + _reset_correlation_id(correlation_token) + _reset_invocation_id(invocation_token) + _reset_branch_name(branch_token) + _reset_fan_out_index(fan_out_token) + _reset_attempt_index(attempt_token) + _reset_namespace_prefix(namespace_token) + + typed = next(e for e in events if isinstance(e, LlmCompletionEvent)) + assert typed.invocation_id == "inv-abc" + assert typed.correlation_id == "corr-xyz" + assert typed.namespace == ("outer", "scoring") + # node_name is the last element of the namespace per the spec + # field-table description ("the user-defined node that issued + # the call"). + assert typed.node_name == "scoring" + assert typed.attempt_index == 2 + assert typed.fan_out_index == 3 + assert typed.branch_name == "fast"