diff --git a/docs/docs/observability/tracing.md b/docs/docs/observability/tracing.md index 2161c28ed..56bc42c36 100644 --- a/docs/docs/observability/tracing.md +++ b/docs/docs/observability/tracing.md @@ -247,6 +247,71 @@ span event per streamed chunk, carrying its index, added text length, and the ap time since the previous chunk (omitted on the first chunk). This is opt-in and off by default, since a long response produces one event per chunk. +#### `adapter_function` span and its phase children + +Covers the adapter-function lifecycle (Epic #929) — LoRA/aLoRA adapters used +for RAG, safety, and core capability checks (`answerability`, +`requirement_check`, etc.). On `mellea.backend`: adapter/model lifecycle work +is a backend concern, not a user-facing operation. + +**As of #1466, this covers `activate`/`deactivate` only.** +`generate`/`parse` fire no spans yet — that lands with #1465, which wires real +generation through `AdapterMixin.adapter_scope`. `prepare` remains metric-only: +it is setup work rather than an adapter-function invocation, so tracing it +needs a separate contract that does not change invocation metrics. `release` +fires no span at all: `WeightsBinding.release()` runs outside any invocation +and has no hook-firing site (see below). + +One `adapter_function` parent span per invocation: + +| Attribute | Description | +| --------- | ----------- | +| `mellea.adapter_function.name` | Adapter function name (e.g. `answerability`) | +| `mellea.adapter_function.revision` | Catalog revision (Hugging Face SHA); omitted when unpinned | +| `mellea.adapter_function.binding_type` | Weight-binding reality (e.g. `local_file`) | +| `mellea.adapter_function.adapter_type` | Adapter mechanism (`lora` or `alora`) | +| `mellea.adapter_function.outcome` | `success`, `schema_error`, or `error` — set when the span closes | + +One `adapter_function.` child span per lifecycle phase that ran: + +| Attribute | Description | +| --------- | ----------- | +| `mellea.adapter_function.phase` | `activate`, `deactivate` (`generate`/`parse` once #1465 lands) | +| `mellea.adapter_function.revision` | Same revision as the parent, recorded directly on the phase span too | + +A phase that raises opens its child span but never fires its own completion +event (matching `mellea.adapter_function.phase_duration`'s metric semantics: +a phase that didn't finish contributes no duration sample) — the enclosing +invocation's own close defensively ends that child span with `ERROR` status +instead, so the in-flight span registry still drains to zero. + +**Nesting is unconditional across Python versions, unlike every other span +pair in this document, with one edge exception.** +Every other family nests via ambient OTel context attach, which needs +Python 3.12+ (see the note under "Span hierarchy" below) — `adapter_function` +children instead parent explicitly via `trace.set_span_in_context`, because +`ADAPTER_FUNCTION_*_START`/`_COMPLETE` fire from **synchronous** code +(`adapter_scope`) through `_run_async_in_thread`, under which +ambient attach can't establish a parent/child edge at all (each dispatched +hook call gets an independent `contextvars` snapshot of the calling thread). +So `adapter_function.` nests under `adapter_function` the same way on +Python 3.11 and 3.12+. The edge exception is not version-related: every +firing site swallows a failed dispatch (an observability failure must never +block the operation it observes), so if an invocation's *start* dispatch +fails, its parent span never opens and that invocation's phase spans open +unparented, falling back to whatever ambient context exists. + +**Known gap: no exemplar linkage to `mellea.adapter_function.phase_duration`.** +Because no span in this family is attached as ambient context (see above), the +`AdapterFunctionMetricsPlugin` histogram sample (a separate plugin subscribed +to the same hooks) never has the `adapter_function`/`adapter_function.` +span ambiently current when it records — at best it would sample whatever +*enclosing application span* happens to be ambient (e.g. `action`), not the +adapter-function span the metric is actually about — regardless of the two +plugins' firing order. Fixing this would mean recording the metric from inside +`AdapterFunctionTracingPlugin` itself, so it can pass the span's context +explicitly — a larger change left for a follow-up. + ### Span hierarchy Backend spans nest inside application spans: diff --git a/mellea/backends/adapters/adapter.py b/mellea/backends/adapters/adapter.py index a173aaaa5..e6bd45811 100644 --- a/mellea/backends/adapters/adapter.py +++ b/mellea/backends/adapters/adapter.py @@ -19,6 +19,7 @@ import pathlib import re import time +import uuid import warnings from collections.abc import Callable from typing import Literal, TypeAlias, TypeVar, cast @@ -323,17 +324,57 @@ def get_adapter_for_intrinsic( return adapter -def _fire_phase_complete_hook(name: str, phase: str, duration_ms: float) -> None: +def _fire_phase_start_hook( + invocation_id: str, name: str, phase: str, revision: str | None +) -> None: + """Fire the `adapter_function_phase_start` hook for a phase about to run. + + A hook-dispatch failure is logged and ignored: observability must not block + a phase from running. + + Args: + invocation_id: Correlation id of the enclosing invocation. + name: Adapter function name, recorded as a span attribute. + phase: Lifecycle phase name; must be a valid + `AdapterFunctionPhaseStartPayload.phase` value. + revision: Catalog revision of the adapter, or `None` if unpinned. + """ + if not has_plugins(HookType.ADAPTER_FUNCTION_PHASE_START): + return + from ...plugins.hooks.adapter_function import AdapterFunctionPhaseStartPayload + + try: + payload = AdapterFunctionPhaseStartPayload( + adapter_function_invocation_id=invocation_id, + name=name, + phase=phase, + revision=revision, + ) + hook_coro = invoke_hook(HookType.ADAPTER_FUNCTION_PHASE_START, payload) + _run_async_in_thread(hook_coro) + except Exception: + MelleaLogger.get_logger().warning( + f"adapter_function_phase_start hook dispatch failed for {name!r} " + f"during {phase!r}; ignoring so it does not block the phase from " + "running.", + exc_info=True, + ) + + +def _fire_phase_complete_hook( + invocation_id: str, name: str, phase: str, duration_ms: float +) -> None: """Fire the `adapter_function_phase_complete` metric hook for a phase that already ran. Split out of `_run_adapter_phase` so a caller that must guarantee cleanup after a phase's side effect — e.g. `adapter_scope` guaranteeing `deactivate()` runs once `activate()` has succeeded — can run the side - effect and this hook fire under separate exception handling. A hook-dispatch - failure is logged and ignored: observability must not turn a completed - lifecycle phase into an operation failure. + effect and this hook fire under separate exception handling. A + construction or dispatch failure is logged and ignored: observability + must not turn a completed lifecycle phase into an operation failure. Args: + invocation_id: Correlation id of the enclosing invocation. name: Adapter function name, used as the metric's `name` field. phase: Lifecycle phase name; must be a valid `AdapterFunctionPhaseCompletePayload.phase` value. @@ -343,10 +384,13 @@ def _fire_phase_complete_hook(name: str, phase: str, duration_ms: float) -> None return from ...plugins.hooks.adapter_function import AdapterFunctionPhaseCompletePayload - payload = AdapterFunctionPhaseCompletePayload( - name=name, phase=phase, duration_ms=duration_ms - ) try: + payload = AdapterFunctionPhaseCompletePayload( + adapter_function_invocation_id=invocation_id, + name=name, + phase=phase, + duration_ms=duration_ms, + ) hook_coro = invoke_hook(HookType.ADAPTER_FUNCTION_PHASE_COMPLETE, payload) _run_async_in_thread(hook_coro) except Exception: @@ -358,32 +402,83 @@ def _fire_phase_complete_hook(name: str, phase: str, duration_ms: float) -> None ) -def _run_adapter_phase(name: str, phase: str, phase_fn: Callable[[], None]) -> None: - """Run one lifecycle phase and fire its phase-complete metric hook. +def _run_adapter_phase( + invocation_id: str, + name: str, + phase: str, + revision: str | None, + phase_fn: Callable[[], None], +) -> None: + """Run one lifecycle phase, firing its phase-start and phase-complete hooks. - Fires the hook only; it does not open a span. Span production belongs to a + Fires hooks only; it does not open a span. Span production belongs to a plugin (#1464, #1466), not to code under `mellea/backends/`. - The hook fires **only when the phase succeeds**, matching the name of - `ADAPTER_FUNCTION_PHASE_COMPLETE`: a phase that raised did not complete. If - `phase_fn` raises, the exception propagates and no phase event is emitted, so - a consumer reconciling phase counts against invocation counts will see the - failure only at invocation level, where `outcome` and `error` carry it. + The complete hook fires **only when the phase succeeds**, matching the name + of `ADAPTER_FUNCTION_PHASE_COMPLETE`: a phase that raised did not complete. + If `phase_fn` raises, the exception propagates and no phase-complete event is + emitted (the phase-start event already fired), so a consumer reconciling + phase counts against invocation counts will see the failure only at + invocation level, where `outcome` and `error` carry it. Args: + invocation_id: Correlation id of the enclosing invocation. name: Adapter function name, used as the metric's `name` field. phase: Lifecycle phase name; must be a valid `AdapterFunctionPhaseCompletePayload.phase` value. + revision: Catalog revision of the adapter, or `None` if unpinned. phase_fn: The zero-argument callable implementing the phase (e.g. `adapter.weights.activate`). """ + _fire_phase_start_hook(invocation_id, name, phase, revision) started_at = time.monotonic() phase_fn() - _fire_phase_complete_hook(name, phase, (time.monotonic() - started_at) * 1000.0) + _fire_phase_complete_hook( + invocation_id, name, phase, (time.monotonic() - started_at) * 1000.0 + ) + + +def _fire_invocation_start_hook( + invocation_id: str, + *, + name: str, + revision: str | None, + binding_type: str, + adapter_type: str, +) -> None: + """Fire the `adapter_function_invocation_start` hook. + + Unlike `_fire_phase_start_hook`, neither the payload construction nor the + dispatch is guarded here: a failure escapes to the call site, which must + wrap the call in `try`/`except` if the operation must not be blocked (see + `adapter_scope`). + + Args: + invocation_id: Correlation id shared with the matching + `_fire_invocation_complete` call. + name: Adapter function name. + revision: Catalog revision of the adapter, or `None` if unpinned. + binding_type: Weight-binding reality the adapter will run under. + adapter_type: Adapter mechanism (e.g. `"lora"`, `"alora"`). + """ + if not has_plugins(HookType.ADAPTER_FUNCTION_INVOCATION_START): + return + from ...plugins.hooks.adapter_function import AdapterFunctionInvocationStartPayload + + payload = AdapterFunctionInvocationStartPayload( + adapter_function_invocation_id=invocation_id, + name=name, + revision=revision, + binding_type=binding_type, + adapter_type=adapter_type, + ) + hook_coro = invoke_hook(HookType.ADAPTER_FUNCTION_INVOCATION_START, payload) + _run_async_in_thread(hook_coro) def _fire_invocation_complete( *, + invocation_id: str, name: str, revision: str | None, binding_type: str, @@ -393,7 +488,14 @@ def _fire_invocation_complete( ) -> None: """Fire the `adapter_function_invocation_complete` metric hook. + Payload construction is unguarded here, as in + `_fire_invocation_start_hook`: the call site carries the `try`/`except` — + a complete-hook failure must be logged and swallowed, never mask the real + outcome (see `adapter_scope`). + Args: + invocation_id: Correlation id shared with the matching + `_fire_invocation_start_hook` call. name: Adapter function name. revision: Catalog revision of the adapter, or `None` if unpinned. binding_type: Weight-binding reality the adapter ran under. @@ -408,6 +510,7 @@ def _fire_invocation_complete( ) payload = AdapterFunctionInvocationCompletePayload( + adapter_function_invocation_id=invocation_id, name=name, revision=revision, binding_type=binding_type, @@ -695,14 +798,15 @@ def adapter_scope(self, adapter: "_AdapterCore | None"): # type: ignore[type-ar A no-op when `adapter` is `None`. Otherwise: activates `adapter.weights`, yields, then always deactivates — even if the `with` - body raises. Each phase fires `ADAPTER_FUNCTION_PHASE_COMPLETE`, and - `ADAPTER_FUNCTION_INVOCATION_COMPLETE` fires on the way out, carrying the - overall outcome. + body raises. Fires `ADAPTER_FUNCTION_INVOCATION_START` on the way in; + each phase fires `ADAPTER_FUNCTION_PHASE_START` then + `ADAPTER_FUNCTION_PHASE_COMPLETE`; `ADAPTER_FUNCTION_INVOCATION_COMPLETE` + fires on the way out, carrying the overall outcome. This method fires hooks only; it does not open spans. Span production is a - plugin's job (see #1464 for the rule and #1466 for the adapter-function - spans), and the `ADAPTER_FUNCTION_*` family currently has no start hook for - a plugin to open a span on. Hook dispatch goes through + plugin's job (see #1464 for the rule). `AdapterFunctionTracingPlugin` + (`mellea/telemetry/tracing_plugins.py`) turns the paired lifecycle hooks + into the `adapter_function` span tree. Hook dispatch goes through `_run_async_in_thread` (no timeout): the dispatching call blocks the calling thread, but the hook coroutine itself runs on the shared `_EventLoopHandler` event-loop thread. A subscriber that blocks on @@ -789,7 +893,6 @@ def adapter_scope(self, adapter: "_AdapterCore | None"): # type: ignore[type-ar else: revision = cast(str | None, getattr(adapter.weights, "revision", None)) binding_type = adapter.weights.binding_type - adapter_type = adapter.identity.adapter_type # adapter_scope drives the WeightsBinding lifecycle (activate/deactivate); # a binding with no lifecycle (e.g. EmbeddedBinding) activates through its @@ -801,17 +904,39 @@ def adapter_scope(self, adapter: "_AdapterCore | None"): # type: ignore[type-ar "scope. Call apply_activation() directly instead." ) + adapter_type = adapter.identity.adapter_type + invocation_id = str(uuid.uuid4()) + + try: + _fire_invocation_start_hook( + invocation_id, + name=name, + revision=revision, + binding_type=binding_type, + adapter_type=adapter_type, + ) + except Exception: + MelleaLogger.get_logger().warning( + f"adapter_function_invocation_start hook dispatch failed for " + f"{name!r}; ignoring so it does not block activation.", + exc_info=True, + ) + outcome: Literal["success", "schema_error", "error"] = "success" exception: BaseException | None = None activated = False body_exception: BaseException | None = None try: + _fire_phase_start_hook(invocation_id, name, "activate", revision) started_at = time.monotonic() try: adapter.weights.activate() activated = True _fire_phase_complete_hook( - name, "activate", (time.monotonic() - started_at) * 1000.0 + invocation_id, + name, + "activate", + (time.monotonic() - started_at) * 1000.0, ) try: yield @@ -822,7 +947,11 @@ def adapter_scope(self, adapter: "_AdapterCore | None"): # type: ignore[type-ar if activated: try: _run_adapter_phase( - name, "deactivate", adapter.weights.deactivate + invocation_id, + name, + "deactivate", + revision, + adapter.weights.deactivate, ) except BaseException as deactivate_exc: if body_exception is None: @@ -852,6 +981,7 @@ def adapter_scope(self, adapter: "_AdapterCore | None"): # type: ignore[type-ar # telemetry-plumbing one. Log and swallow instead. try: _fire_invocation_complete( + invocation_id=invocation_id, name=name, revision=revision, binding_type=binding_type, diff --git a/mellea/plugins/hooks/__init__.py b/mellea/plugins/hooks/__init__.py index 3095fc2cf..5cccf5031 100644 --- a/mellea/plugins/hooks/__init__.py +++ b/mellea/plugins/hooks/__init__.py @@ -5,7 +5,9 @@ from .adapter_function import ( AdapterFunctionInvocationCompletePayload, + AdapterFunctionInvocationStartPayload, AdapterFunctionPhaseCompletePayload, + AdapterFunctionPhaseStartPayload, ) from .component import ( ComponentPostErrorPayload, @@ -31,7 +33,9 @@ __all__ = [ # Adapter Function "AdapterFunctionInvocationCompletePayload", + "AdapterFunctionInvocationStartPayload", "AdapterFunctionPhaseCompletePayload", + "AdapterFunctionPhaseStartPayload", # Component "ComponentPostErrorPayload", "ComponentPostSuccessPayload", diff --git a/mellea/plugins/hooks/adapter_function.py b/mellea/plugins/hooks/adapter_function.py index 924015275..f4dda727d 100644 --- a/mellea/plugins/hooks/adapter_function.py +++ b/mellea/plugins/hooks/adapter_function.py @@ -9,11 +9,54 @@ from mellea.plugins.base import MelleaBasePayload +# Every lifecycle phase that fires (or is expected to fire in a future issue) +# ADAPTER_FUNCTION_PHASE_START / ADAPTER_FUNCTION_PHASE_COMPLETE. Firing sites, +# issue #1466: +# prepare -- metric-only completion in LocalFileBinding.prepare(); tracing +# is deferred because setup is not an adapter-function invocation +# activate -- AdapterMixin.adapter_scope() +# deactivate -- AdapterMixin.adapter_scope() +# generate -- none yet; blocked on #1465 wiring generation through adapter_scope +# parse -- none yet; blocked on #1465 wiring parsing through adapter_scope +# release -- none yet, deliberately: WeightsBinding.release() runs outside +# adapter_scope and is not wrapped in an invocation, unlike +# prepare/activate/deactivate (see LocalFileBinding.release() and +# the pre-existing note this carries forward from Epic #929 Phase 1 +# that "release" has no phase-duration metric in this contract). +# It keeps a Literal value so downstream consumers can name it, but +# has no firing site — tracked as remaining work, not silently +# dropped. +AdapterFunctionPhase = Literal[ + "prepare", "activate", "generate", "parse", "deactivate", "release" +] + + +class AdapterFunctionInvocationStartPayload(MelleaBasePayload): + """Payload for `adapter_function_invocation_start` — before an adapter function invocation begins. + + Attributes: + adapter_function_invocation_id: Correlation id shared with the matching + `adapter_function_invocation_complete` event. + name: Adapter function name (e.g. `"answerability"`). + revision: Catalog revision of the adapter, or `None` if unpinned. + binding_type: Weight-binding reality the adapter will run under (e.g. + `"local_file"`, `"embedded"`, `"server_mediated"`). + adapter_type: Adapter mechanism (e.g. `"lora"`, `"alora"`). + """ + + adapter_function_invocation_id: str + name: str + revision: str | None = None + binding_type: str = "unknown" + adapter_type: str = "unknown" + class AdapterFunctionInvocationCompletePayload(MelleaBasePayload): """Payload for `adapter_function_invocation_complete` — after an adapter function invocation finishes. Attributes: + adapter_function_invocation_id: Correlation id shared with the `adapter_function_invocation_start` + event that opened this invocation. name: Adapter function name (e.g. `"answerability"`). revision: Catalog revision of the adapter, or `None` if unpinned. binding_type: Weight-binding reality the adapter ran under (e.g. @@ -23,6 +66,7 @@ class AdapterFunctionInvocationCompletePayload(MelleaBasePayload): error: The exception raised during invocation, or `None` on success. """ + adapter_function_invocation_id: str name: str revision: str | None = None binding_type: str = "unknown" @@ -37,20 +81,45 @@ class AdapterFunctionInvocationCompletePayload(MelleaBasePayload): error: Any = None +class AdapterFunctionPhaseStartPayload(MelleaBasePayload): + """Payload for `adapter_function_phase_start` — before one lifecycle phase begins. + + Attributes: + adapter_function_invocation_id: Correlation id of the enclosing invocation (shared with the + `adapter_function_invocation_start`/`_complete` events). + name: Adapter function name (e.g. `"answerability"`). + phase: Lifecycle phase about to run. See `AdapterFunctionPhase` for which + values currently have a firing site. + revision: Catalog revision of the adapter, or `None` if unpinned. + """ + + adapter_function_invocation_id: str + name: str + phase: AdapterFunctionPhase + revision: str | None = None + + class AdapterFunctionPhaseCompletePayload(MelleaBasePayload): """Payload for `adapter_function_phase_complete` — after one lifecycle phase finishes. + Only fires when the phase itself succeeded — a phase that raised did not + complete, and its failure is reported once, at invocation level, via + `adapter_function_invocation_complete`'s `outcome`/`error`. + Attributes: + adapter_function_invocation_id: Correlation id of the enclosing invocation, + or `None` for an existing metric-only completion with no span. name: Adapter function name (e.g. `"answerability"`). - phase: Lifecycle phase (`"prepare"`, `"activate"`, `"generate"`, - `"parse"`, or `"deactivate"`). + phase: Lifecycle phase that completed. See `AdapterFunctionPhase` for + which values currently have a firing site. duration_ms: Wall-clock duration of the phase in milliseconds. """ + adapter_function_invocation_id: str | None = None name: str # Constrained to a Literal so a typo can't silently spawn a new metric-label # series (the phase becomes a metric dimension). Required, with no unset # sentinel: a phase-complete event always has a real phase. (The payload is a # pydantic model, so a required field after the base's defaulted ones is fine.) - phase: Literal["prepare", "activate", "generate", "parse", "deactivate"] + phase: AdapterFunctionPhase duration_ms: float diff --git a/mellea/plugins/types.py b/mellea/plugins/types.py index cddc6b865..b7704fe4b 100644 --- a/mellea/plugins/types.py +++ b/mellea/plugins/types.py @@ -72,7 +72,9 @@ class HookType(StrEnum): TOOL_POST_INVOKE = "tool_post_invoke" # Adapter Function Lifecycle + ADAPTER_FUNCTION_INVOCATION_START = "adapter_function_invocation_start" ADAPTER_FUNCTION_INVOCATION_COMPLETE = "adapter_function_invocation_complete" + ADAPTER_FUNCTION_PHASE_START = "adapter_function_phase_start" ADAPTER_FUNCTION_PHASE_COMPLETE = "adapter_function_phase_complete" # Streaming Pipeline @@ -93,7 +95,9 @@ def _build_hook_registry() -> dict[str, tuple[type, type]]: """ from mellea.plugins.hooks.adapter_function import ( AdapterFunctionInvocationCompletePayload, + AdapterFunctionInvocationStartPayload, AdapterFunctionPhaseCompletePayload, + AdapterFunctionPhaseStartPayload, ) from mellea.plugins.hooks.component import ( ComponentPostErrorPayload, @@ -180,10 +184,18 @@ def _build_hook_registry() -> dict[str, tuple[type, type]]: HookType.TOOL_PRE_INVOKE.value: (ToolPreInvokePayload, PluginResult), HookType.TOOL_POST_INVOKE.value: (ToolPostInvokePayload, PluginResult), # Adapter Function Lifecycle + HookType.ADAPTER_FUNCTION_INVOCATION_START.value: ( + AdapterFunctionInvocationStartPayload, + PluginResult, + ), HookType.ADAPTER_FUNCTION_INVOCATION_COMPLETE.value: ( AdapterFunctionInvocationCompletePayload, PluginResult, ), + HookType.ADAPTER_FUNCTION_PHASE_START.value: ( + AdapterFunctionPhaseStartPayload, + PluginResult, + ), HookType.ADAPTER_FUNCTION_PHASE_COMPLETE.value: ( AdapterFunctionPhaseCompletePayload, PluginResult, diff --git a/mellea/telemetry/tracing.py b/mellea/telemetry/tracing.py index e359d0776..4db0378a5 100644 --- a/mellea/telemetry/tracing.py +++ b/mellea/telemetry/tracing.py @@ -997,6 +997,199 @@ def finish_validation_span( _finish_application_span_error(validation_id, exception=exception) +# Child phase spans are stashed under a key derived from the invocation id so +# they can't collide with the parent's own `_in_flight_spans` entry (keyed by +# the bare invocation id) or with each other across phases of the same +# invocation. +_ADAPTER_FUNCTION_PHASE_KEY_INFIX = ":phase:" + + +def _adapter_function_phase_key(invocation_id: str, phase: str) -> str: + return f"{invocation_id}{_ADAPTER_FUNCTION_PHASE_KEY_INFIX}{phase}" + + +def start_adapter_function_span( + invocation_id: str, + *, + name: str, + revision: str | None, + binding_type: str, + adapter_type: str, +) -> Span | None: + """Open the `adapter_function` parent span for one adapter-function invocation. + + On the `mellea.backend` tracer: adapter/model lifecycle work is a backend + concern, not a user-facing operation (see + `docs/docs/observability/tracing.md`). + + Never attached as the ambient OTel context, unlike every other + `start_*_span` helper — deliberately, not as an oversight. + `ADAPTER_FUNCTION_INVOCATION_START`/`_PHASE_START` fire from sync code + (`AdapterMixin.adapter_scope`, `LocalFileBinding.prepare`) via + `_run_async_in_thread`, which runs each hook as an independent task on a + shared background event loop, seeded from a *fresh* `contextvars.copy_context()` + snapshot of the calling thread taken at that call — mutations inside one + hook's task (like an ambient-context attach) never leak back to the + calling thread and so are invisible to the *next* `_run_async_in_thread` + call's snapshot. Ambient attach/detach across two such calls therefore + can't establish a parent/child edge (and mismatched attach/detach tasks + would trigger "Detaching an OTel context token across asyncio tasks" + warnings) — `start_adapter_function_phase_span` instead parents explicitly + via `trace.set_span_in_context` using the span object looked up by + `invocation_id`, which needs no ambient context at all. Exposing an + `attach_context` parameter here (as the sibling helpers do) would only + offer a setting that breaks whenever used, so there isn't one. + + Args: + invocation_id: Correlation key for the matching `finish_adapter_function_span` call. + name: Adapter function name (e.g. `"answerability"`). + revision: Catalog revision of the adapter, or `None` if unpinned. + binding_type: Weight-binding reality the adapter is running under (e.g. + `"local_file"`, `"embedded"`, `"server_mediated"`). + adapter_type: Adapter mechanism (e.g. `"lora"`, `"alora"`). + + Returns: + The span, or `None` if tracing is disabled. + """ + tracer = get_backend_tracer() + if tracer is None: + return None + + span = tracer.start_span("adapter_function") + set_attribute_safe(span, "mellea.adapter_function.name", name) + set_attribute_safe(span, "mellea.adapter_function.revision", revision) + set_attribute_safe(span, "mellea.adapter_function.binding_type", binding_type) + set_attribute_safe(span, "mellea.adapter_function.adapter_type", adapter_type) + + _in_flight_spans[invocation_id] = (span, None) + return span + + +def finish_adapter_function_span( + invocation_id: str, *, outcome: str, exception: BaseException | None +) -> None: + """End the `adapter_function` span, recording its outcome. + + Defensively closes any `adapter_function.` child span still open + under this invocation first — a phase that raised fires + `adapter_function_phase_start` but never its own + `adapter_function_phase_complete` (that hook's contract is success-only), + so without this the child span would never close and the in-flight + registry would never drain. The dangling child receives the invocation + exception unless it is `deactivate` and the invocation body also failed: + that exception remains primary, so copying it onto the deactivation child + would misattribute the failure. + + Args: + invocation_id: Correlation key from the matching `start_adapter_function_span` call. + outcome: `"success"`, `"schema_error"`, or `"error"`. + exception: The exception raised during the invocation, or `None` on success. + """ + # `list(...)` snapshots the keys in one call before filtering, so a + # concurrent insert from another thread (e.g. a different invocation's + # sync-dispatched hook, on the shared `_run_async_in_thread` background + # loop) can't trigger a "dictionary changed size during iteration" error + # here — the first site in this module to iterate `_in_flight_spans` + # rather than do a keyed lookup. + prefix = f"{invocation_id}{_ADAPTER_FUNCTION_PHASE_KEY_INFIX}" + for key in [k for k in list(_in_flight_spans) if k.startswith(prefix)]: + entry = _in_flight_spans.pop(key, None) + if entry is None: + continue + phase_span, _phase_token = entry + phase = key.removeprefix(prefix) + deactivation_failure_noted = ( + phase == "deactivate" + and exception is not None + and any( + note.startswith("Adapter deactivation also failed:") + for note in getattr(exception, "__notes__", ()) + ) + ) + if exception is not None: + if deactivation_failure_noted: + phase_span.set_status( + trace.Status(trace.StatusCode.ERROR, "Phase did not complete") + ) + else: + phase_span.record_exception(exception) + phase_span.set_status( + trace.Status(trace.StatusCode.ERROR, str(exception)) + ) + phase_span.set_attribute("error.type", type(exception).__name__) + phase_span.end() + + entry = _in_flight_spans.pop(invocation_id, None) + if entry is None: + return + span, _token = entry + set_attribute_safe(span, "mellea.adapter_function.outcome", outcome) + if exception is not None: + span.record_exception(exception) + span.set_status(trace.Status(trace.StatusCode.ERROR, str(exception))) + span.set_attribute("error.type", type(exception).__name__) + span.end() + + +def start_adapter_function_phase_span( + invocation_id: str, phase: str, *, revision: str | None = None +) -> Span | None: + """Open an `adapter_function.` child span, explicitly parented under the invocation span. + + Parents via `trace.set_span_in_context` on the `adapter_function` span + looked up by `invocation_id`, rather than via ambient context — see + `start_adapter_function_span`'s docstring for why ambient attach can't + establish this edge for hooks fired via `_run_async_in_thread`. Explicit + parenting works the same on every Python version; no `_CONTEXT_ATTACH_SUPPORTED` + gating applies here. A missing parent (invocation not in flight, or + tracing was enabled only after the invocation started) falls back to + whatever's ambient — the phase span still opens; it just can't nest. + + Args: + invocation_id: Correlation key of the enclosing `adapter_function` span. + phase: Lifecycle phase name (e.g. `"prepare"`, `"activate"`). + revision: Catalog revision of the adapter, or `None` if unpinned. Recorded + directly on this phase span — e.g. `adapter_function.prepare` records + the resolved Hugging Face SHA here, not just on the parent. + + Returns: + The span, or `None` if tracing is disabled. + """ + tracer = get_backend_tracer() + if tracer is None: + return None + + parent_entry = _in_flight_spans.get(invocation_id) + parent_context = ( + trace.set_span_in_context(parent_entry[0]) if parent_entry is not None else None + ) + span = tracer.start_span(f"adapter_function.{phase}", context=parent_context) + set_attribute_safe(span, "mellea.adapter_function.phase", phase) + set_attribute_safe(span, "mellea.adapter_function.revision", revision) + + _in_flight_spans[_adapter_function_phase_key(invocation_id, phase)] = (span, None) + return span + + +def finish_adapter_function_phase_span(invocation_id: str, phase: str) -> None: + """End an `adapter_function.` child span successfully. + + A no-op if the phase span isn't in flight — e.g. it was already closed + defensively by `finish_adapter_function_span` because the phase raised. + + Args: + invocation_id: Correlation key of the enclosing `adapter_function` span. + phase: Lifecycle phase name. + """ + entry = _in_flight_spans.pop( + _adapter_function_phase_key(invocation_id, phase), None + ) + if entry is None: + return + span, _token = entry + span.end() + + __all__ = [ "get_application_tracer", "get_backend_tracer", diff --git a/mellea/telemetry/tracing_plugins.py b/mellea/telemetry/tracing_plugins.py index 112a5d6a1..f0c25901b 100644 --- a/mellea/telemetry/tracing_plugins.py +++ b/mellea/telemetry/tracing_plugins.py @@ -17,6 +17,10 @@ - SamplingTracingPlugin: Emits a `sampling` span per sampling loop, with a span event per iteration and repair. - ValidationTracingPlugin: Emits a `validation` span per requirement-check batch. +- AdapterFunctionTracingPlugin: Emits the `adapter_function` span tree (one + parent span per invocation, one `adapter_function.` child per + lifecycle phase) for adapter activation. Covers activate/deactivate only as + of #1466; generate/parse land with #1465. """ from __future__ import annotations @@ -35,6 +39,12 @@ _CONTEXT_ATTACH_SUPPORTED: bool = sys.version_info >= (3, 12) if TYPE_CHECKING: + from mellea.plugins.hooks.adapter_function import ( + AdapterFunctionInvocationCompletePayload, + AdapterFunctionInvocationStartPayload, + AdapterFunctionPhaseCompletePayload, + AdapterFunctionPhaseStartPayload, + ) from mellea.plugins.hooks.component import ( ComponentPostErrorPayload, ComponentPostSuccessPayload, @@ -612,6 +622,110 @@ async def on_post_check( ) +class AdapterFunctionTracingPlugin( + Plugin, name="adapter_function_tracing", priority=1046 +): + """Emits the `adapter_function` span tree for the adapter-function lifecycle. + + `adapter_function_invocation_start` opens the `adapter_function` parent + span; `adapter_function_invocation_complete` closes it, recording the + outcome and defensively closing any `adapter_function.` child span + left open by a phase that raised (see `finish_adapter_function_span`). + `adapter_function_phase_start`/`adapter_function_phase_complete` open/close + one child span per lifecycle phase, correlated with the parent via + `adapter_function_invocation_id`. + + On the `mellea.backend` tracer (adapter/model lifecycle work, not a + user-facing operation — see `docs/docs/observability/tracing.md`). + + Covers `prepare`/`activate`/`deactivate` only as of #1466. `generate`/ + `parse` fire no hooks yet (blocked on #1465 wiring generation through + `AdapterMixin.adapter_scope`), so this plugin does not yet emit + `adapter_function.generate`/`adapter_function.parse` — it will, once those + hooks fire, with no changes needed here. `release` fires no hooks at all + (see `AdapterFunctionPhaseCompletePayload`'s `phase` field for why) and so + never gets a span either. Content capture (`MELLEA_TRACES_CONTENT`) is not + wired here: no phase in scope carries adapter input/output content — + `generate`/`parse` are where that will apply. + + Unlike every other plugin in this module, these hooks don't rely on + `_CONTEXT_ATTACH_SUPPORTED` ambient-context attach/detach at all — they + fire from sync code (`AdapterMixin.adapter_scope`, `LocalFileBinding.prepare`) + via `_run_async_in_thread`, under which ambient attach can't establish a + parent/child edge (see `start_adapter_function_span`'s docstring). + `start_adapter_function_phase_span` parents each child explicitly instead, + so nesting works identically on every Python version. + + Exemplar linkage (see `docs/docs/observability/tracing.md`) is **not** + established for these spans, and can't be with the current architecture: + an OTel histogram exemplar samples whatever span is *ambiently* current at + the moment the metric is recorded, but no span in this family is ever + attached as ambient context (see above) — `AdapterFunctionMetricsPlugin` + (`mellea/telemetry/metrics_plugins.py`, a separate plugin subscribed to the + same hooks) at best samples whatever enclosing application span happens to + be ambient, never the adapter-function span the metric is actually about, + regardless of firing order between the two plugins. This is a known, + structural gap, not an oversight: fixing it would need the metric recorded + from inside this plugin (so it can pass the span's context explicitly) + rather than from a same-hook sibling plugin, which is a larger change than + adding spans and is left for a follow-up. + """ + + @hook("adapter_function_invocation_start") + async def on_invocation_start( + self, payload: AdapterFunctionInvocationStartPayload, context: dict[str, Any] + ) -> None: + """Open the `adapter_function` parent span for this invocation.""" + from mellea.telemetry.tracing import start_adapter_function_span + + start_adapter_function_span( + payload.adapter_function_invocation_id, + name=payload.name, + revision=payload.revision, + binding_type=payload.binding_type, + adapter_type=payload.adapter_type, + ) + + @hook("adapter_function_invocation_complete") + async def on_invocation_complete( + self, payload: AdapterFunctionInvocationCompletePayload, context: dict[str, Any] + ) -> None: + """Close the `adapter_function` span with its outcome.""" + from mellea.telemetry.tracing import finish_adapter_function_span + + finish_adapter_function_span( + payload.adapter_function_invocation_id, + outcome=payload.outcome, + exception=payload.error, + ) + + @hook("adapter_function_phase_start") + async def on_phase_start( + self, payload: AdapterFunctionPhaseStartPayload, context: dict[str, Any] + ) -> None: + """Open the `adapter_function.` child span.""" + from mellea.telemetry.tracing import start_adapter_function_phase_span + + start_adapter_function_phase_span( + payload.adapter_function_invocation_id, + payload.phase, + revision=payload.revision, + ) + + @hook("adapter_function_phase_complete") + async def on_phase_complete( + self, payload: AdapterFunctionPhaseCompletePayload, context: dict[str, Any] + ) -> None: + """Close the `adapter_function.` child span.""" + if payload.adapter_function_invocation_id is None: + return + from mellea.telemetry.tracing import finish_adapter_function_phase_span + + finish_adapter_function_phase_span( + payload.adapter_function_invocation_id, payload.phase + ) + + # All tracing plugins to auto-register when tracing is enabled. _TRACING_PLUGIN_CLASSES = ( BackendTracingPlugin, @@ -620,4 +734,5 @@ async def on_post_check( ToolTracingPlugin, SamplingTracingPlugin, ValidationTracingPlugin, + AdapterFunctionTracingPlugin, ) diff --git a/test/backends/test_adapters/_hook_capture.py b/test/backends/test_adapters/_hook_capture.py index 67429030c..2a9bd79bf 100644 --- a/test/backends/test_adapters/_hook_capture.py +++ b/test/backends/test_adapters/_hook_capture.py @@ -16,6 +16,13 @@ from collections.abc import Iterator from unittest.mock import MagicMock, patch +from mellea.plugins.hooks.adapter_function import ( + AdapterFunctionInvocationCompletePayload, + AdapterFunctionInvocationStartPayload, + AdapterFunctionPhaseCompletePayload, + AdapterFunctionPhaseStartPayload, +) + _TARGET = "mellea.backends.adapters.adapter" @@ -66,6 +73,22 @@ def hook_payloads(mock_invoke: MagicMock) -> list: return [call.args[1] for call in mock_invoke.call_args_list] +def phase_start_payloads(mock_invoke: MagicMock) -> list: + """Returns only the phase-start payloads. + + Args: + mock_invoke: The mock yielded by `capture_adapter_hooks`. + + Returns: + The recorded `AdapterFunctionPhaseStartPayload`s, ordered as fired. + """ + return [ + p + for p in hook_payloads(mock_invoke) + if isinstance(p, AdapterFunctionPhaseStartPayload) + ] + + def phase_payloads(mock_invoke: MagicMock) -> list: """Returns only the phase-complete payloads. @@ -75,7 +98,27 @@ def phase_payloads(mock_invoke: MagicMock) -> list: Returns: The recorded `AdapterFunctionPhaseCompletePayload`s, ordered as fired. """ - return [p for p in hook_payloads(mock_invoke) if hasattr(p, "phase")] + return [ + p + for p in hook_payloads(mock_invoke) + if isinstance(p, AdapterFunctionPhaseCompletePayload) + ] + + +def invocation_start_payloads(mock_invoke: MagicMock) -> list: + """Returns only the invocation-start payloads. + + Args: + mock_invoke: The mock yielded by `capture_adapter_hooks`. + + Returns: + The recorded `AdapterFunctionInvocationStartPayload`s, ordered as fired. + """ + return [ + p + for p in hook_payloads(mock_invoke) + if isinstance(p, AdapterFunctionInvocationStartPayload) + ] def invocation_payloads(mock_invoke: MagicMock) -> list: @@ -87,4 +130,8 @@ def invocation_payloads(mock_invoke: MagicMock) -> list: Returns: The recorded `AdapterFunctionInvocationCompletePayload`s, ordered as fired. """ - return [p for p in hook_payloads(mock_invoke) if hasattr(p, "outcome")] + return [ + p + for p in hook_payloads(mock_invoke) + if isinstance(p, AdapterFunctionInvocationCompletePayload) + ] diff --git a/test/backends/test_adapters/test_adapter_scope.py b/test/backends/test_adapters/test_adapter_scope.py index 85e3af14f..d0e7025f4 100644 --- a/test/backends/test_adapters/test_adapter_scope.py +++ b/test/backends/test_adapters/test_adapter_scope.py @@ -20,6 +20,7 @@ Identity, IOContract, LocalFileBinding, + WeightsBinding, ) from mellea.backends.adapters.adapter import AdapterMixin from mellea.backends.adapters.catalog import AdapterType @@ -27,6 +28,8 @@ from test.backends.test_adapters._hook_capture import ( capture_adapter_hooks, invocation_payloads, + phase_payloads, + phase_start_payloads, ) @@ -201,12 +204,15 @@ def test_adapter_scope_reports_other_exceptions_as_error(): def test_phase_hook_not_fired_when_the_phase_itself_fails(): - """A phase that raised did not complete, so no phase event is emitted. + """A phase that raised opens (phase-start fires) but never completes. - `ADAPTER_FUNCTION_PHASE_COMPLETE` means the phase finished. The failure is - reported once, at invocation level, where `outcome`/`error` carry it — so a + `ADAPTER_FUNCTION_PHASE_COMPLETE` means the phase finished, so it does not + fire for a phase that raised. `ADAPTER_FUNCTION_PHASE_START` fires + regardless, since it only marks the phase as about to run. The failure is + reported at invocation level, where `outcome`/`error` carry it — so a consumer reconciling phase counts against invocation counts sees one - invocation error and no phase event, not both. + invocation error, one phase-start with no matching phase-complete, and no + phase-complete event. """ mock_backend = MagicMock(spec=AdapterMixin) adapter, weights = _make_adapter() @@ -217,8 +223,8 @@ def test_phase_hook_not_fired_when_the_phase_itself_fails(): with AdapterMixin.adapter_scope(mock_backend, adapter): pytest.fail("body must not run when activate() raises") - payloads = [c.args[1] for c in mock_invoke.call_args_list] - assert [p for p in payloads if hasattr(p, "phase")] == [] + assert [p.phase for p in phase_start_payloads(mock_invoke)] == ["activate"] + assert phase_payloads(mock_invoke) == [] invocations = invocation_payloads(mock_invoke) assert [p.outcome for p in invocations] == ["error"] @@ -251,6 +257,98 @@ def test_adapter_scope_reports_resolved_revision_not_raw_none(): assert invocations[0].revision != "main" +class _IntRevisionBinding(WeightsBinding): + """Non-`LocalFileBinding` weights whose `.revision` is not a `str`. + + `adapter_scope` treats `WeightsBinding.revision` as unverified duck-typed + data for any binding that isn't a `LocalFileBinding` — it reads it via a + bare `getattr(...)` with no type check. This double stands in for any + third-party binding that sets a non-`str` revision. + """ + + binding_type = "custom" + revision = 7 + + def prepare(self) -> None: + pass + + def activate(self) -> None: + pass + + def deactivate(self) -> None: + pass + + def release(self) -> None: + pass + + +def test_adapter_scope_swallows_non_str_revision_on_phase_start(): + """A non-`str` `.revision` must not abort activation. + + Regression guard: `_fire_phase_start_hook` used to construct + `AdapterFunctionPhaseStartPayload` outside its own `try`, so a duck-typed + binding with a non-`str` `.revision` raised a pydantic `ValidationError` + that escaped `adapter_scope` entirely, aborting before `activate()` ever + ran — exactly the failure the function's docstring says it prevents. + """ + mock_backend = MagicMock(spec=AdapterMixin) + weights = _IntRevisionBinding() + identity = Identity(name="answerability", adapter_type="lora") + adapter = Adapter(identity=identity, io_contract=_Contract(), weights=weights) + + body_ran = False + with AdapterMixin.adapter_scope(mock_backend, adapter): + body_ran = True + + assert body_ran + + +def test_adapter_scope_swallows_invocation_start_hook_failure(): + """A failing invocation-start hook must not block activation. + + Regression guard: `_fire_invocation_start_hook` is called unguarded inside + `adapter_scope` itself, wrapped in its own try/except at the call site + (unlike `_fire_phase_start_hook`, which guards internally). Pin that the + call-site guard actually works — `_fire_invocation_complete` already has + two dedicated tests for this; `_fire_invocation_start_hook` had none. + """ + mock_backend = MagicMock(spec=AdapterMixin) + adapter, weights = _make_adapter() + + with patch( + "mellea.backends.adapters.adapter._fire_invocation_start_hook", + side_effect=RuntimeError("start hook dispatch blew up"), + ): + with AdapterMixin.adapter_scope(mock_backend, adapter): + pass # must not raise despite the start hook failing on entry + + weights.activate.assert_called_once() + weights.deactivate.assert_called_once() + + +def test_adapter_scope_swallows_non_str_identity_name_on_phase_complete(): + """A non-`str` adapter identity name must not abort activation. + + Regression guard: `_fire_phase_complete_hook` constructed its payload + outside its guard, so an adapter whose `Identity.name` was not a + `str` (`Identity` is a plain frozen dataclass and performs no runtime + type coercion) raised a pydantic `ValidationError` from the activate + phase-complete site after `activate()` had already succeeded — the + body never ran and a healthy invocation was reported as an error. + Construction now sits under the same guard as the dispatch. + """ + mock_backend = MagicMock(spec=AdapterMixin) + weights = _IntRevisionBinding() + identity = Identity(name=7, adapter_type="lora") # type: ignore[arg-type] + adapter = Adapter(identity=identity, io_contract=_Contract(), weights=weights) + + body_ran = False + with AdapterMixin.adapter_scope(mock_backend, adapter): + body_ran = True + + assert body_ran + + def test_adapter_scope_swallows_invocation_hook_failure_on_clean_run(): """A failing invocation-complete hook must not turn a clean run into an error. diff --git a/test/backends/test_adapters/test_local_file_integration.py b/test/backends/test_adapters/test_local_file_integration.py index a369ffad7..67ccc67c0 100644 --- a/test/backends/test_adapters/test_local_file_integration.py +++ b/test/backends/test_adapters/test_local_file_integration.py @@ -37,6 +37,9 @@ from test.backends.test_adapters._hook_capture import ( capture_adapter_hooks, hook_payloads, + invocation_start_payloads, + phase_payloads, + phase_start_payloads, ) pytestmark = pytest.mark.integration @@ -118,11 +121,23 @@ def test_prepare_activate_deactivate_release_full_lifecycle(): assert binding.backend is None recorded = hook_payloads(mock_invoke) - phases = [p.phase for p in recorded if hasattr(p, "phase")] + phases = [p.phase for p in phase_payloads(mock_invoke)] assert phases == ["activate", "deactivate"] + assert [p.phase for p in phase_start_payloads(mock_invoke)] == [ + "activate", + "deactivate", + ] invocations = [p for p in recorded if hasattr(p, "outcome")] assert len(invocations) == 1 + starts = invocation_start_payloads(mock_invoke) + assert len(starts) == 1 + invocation_id = starts[0].adapter_function_invocation_id + assert invocations[0].adapter_function_invocation_id == invocation_id + assert all( + p.adapter_function_invocation_id == invocation_id + for p in [*phase_start_payloads(mock_invoke), *phase_payloads(mock_invoke)] + ) assert invocations[0].outcome == "success" assert invocations[0].name == "answerability" assert invocations[0].binding_type == "local_file" diff --git a/test/backends/test_huggingface_unit.py b/test/backends/test_huggingface_unit.py index 13fa8bc45..b1d4cb8aa 100644 --- a/test/backends/test_huggingface_unit.py +++ b/test/backends/test_huggingface_unit.py @@ -46,6 +46,7 @@ from test.backends.test_adapters._hook_capture import ( capture_adapter_hooks, hook_payloads, + phase_payloads, ) # Minimal 1x1 PNG for testing @@ -481,7 +482,7 @@ def test_generate_intrinsic_with_adapter_scope_fires_hooks_with_correct_payload( assert out == "output" payloads = hook_payloads(mock_invoke) - phases = [p.phase for p in payloads if hasattr(p, "phase")] + phases = [p.phase for p in phase_payloads(mock_invoke)] assert phases == ["activate", "deactivate"] invocations = [p for p in payloads if hasattr(p, "outcome")] @@ -535,7 +536,7 @@ def failing_generate(): backend._generate_intrinsic_with_adapter_scope(adapter, failing_generate) payloads = hook_payloads(mock_invoke) - phases = [p.phase for p in payloads if hasattr(p, "phase")] + phases = [p.phase for p in phase_payloads(mock_invoke)] assert phases == ["activate", "deactivate"] invocations = [p for p in payloads if hasattr(p, "outcome")] diff --git a/test/telemetry/test_metrics_plugins.py b/test/telemetry/test_metrics_plugins.py index a952c2207..df6e49aff 100644 --- a/test/telemetry/test_metrics_plugins.py +++ b/test/telemetry/test_metrics_plugins.py @@ -1217,6 +1217,7 @@ def adapter_function_plugin(): async def test_record_adapter_function_invocation_success(adapter_function_plugin): """A successful invocation records the invocations counter, not parse_failures.""" payload = AdapterFunctionInvocationCompletePayload( + adapter_function_invocation_id="inv-1", name="answerability", revision="r1", binding_type="local_file", @@ -1250,6 +1251,7 @@ async def test_record_adapter_function_invocation_schema_error_also_records_pars ): """A schema_error outcome records both the invocations counter and parse_failures.""" payload = AdapterFunctionInvocationCompletePayload( + adapter_function_invocation_id="inv-2", name="answerability", revision="r1", binding_type="local_file", @@ -1287,6 +1289,7 @@ async def test_record_adapter_function_invocation_none_revision_passed_through( not in the plugin — so the plugin passes the raw payload.revision value. """ payload = AdapterFunctionInvocationCompletePayload( + adapter_function_invocation_id="inv-3", name="answerability", revision=None, binding_type="embedded", @@ -1312,7 +1315,10 @@ async def test_record_adapter_function_invocation_none_revision_passed_through( async def test_record_adapter_function_phase_duration(adapter_function_plugin): """Phase-complete events record the phase-duration histogram in seconds.""" payload = AdapterFunctionPhaseCompletePayload( - name="answerability", phase="prepare", duration_ms=12.5 + adapter_function_invocation_id="inv-4", + name="answerability", + phase="prepare", + duration_ms=12.5, ) with patch( diff --git a/test/telemetry/test_tracing_adapter_function.py b/test/telemetry/test_tracing_adapter_function.py new file mode 100644 index 000000000..d0a8e0db9 --- /dev/null +++ b/test/telemetry/test_tracing_adapter_function.py @@ -0,0 +1,522 @@ +# Copyright IBM Corp. All Rights Reserved. +# SPDX-License-Identifier: Apache-2.0 + +"""Tests for adapter-function tracing — the `adapter_function` span tree (#1466). + +Covers activate/deactivate only; generate/parse have no firing hooks yet +(blocked on #1465). + +Three layers, mirroring `test_tracing_application.py`: + +1. Helper-level unit tests (mock tracer): pin attribute shapes and key + derivation in `tracing.py`'s `start_adapter_function_span`/ + `finish_adapter_function_span`/`start_adapter_function_phase_span`/ + `finish_adapter_function_phase_span`. +2. Plugin unit tests (mock tracer): `AdapterFunctionTracingPlugin`'s hooks + translate payload fields into span opens/closes. +3. Integration tests (real OTel SDK, in-memory exporter, real + `AdapterMixin.adapter_scope` call sites): verify the full + `adapter_function` > `adapter_function.` nesting, the dangling-child + span cleanup on a raised phase, and that the in-flight span registry drains + to zero. +""" + +from unittest.mock import MagicMock, patch + +import pytest + +pytest.importorskip( + "opentelemetry", reason="opentelemetry not installed — install mellea[telemetry]" +) +pytest.importorskip("cpex", reason="cpex not installed — install mellea[hooks]") + +from opentelemetry.sdk.trace.export import SimpleSpanProcessor +from opentelemetry.sdk.trace.export.in_memory_span_exporter import InMemorySpanExporter + +from mellea.backends.adapters._core import ( + Adapter, + Identity, + IOContract, + LocalFileBinding, +) +from mellea.backends.adapters.adapter import AdapterMixin +from mellea.core import Component +from mellea.telemetry import tracing +from mellea.telemetry.tracing import ( + finish_adapter_function_phase_span, + finish_adapter_function_span, + start_adapter_function_phase_span, + start_adapter_function_span, +) +from mellea.telemetry.tracing_plugins import AdapterFunctionTracingPlugin +from test.telemetry.conftest import reset_tracing_state + +# --------------------------------------------------------------------------- +# Fixtures +# --------------------------------------------------------------------------- + + +@pytest.fixture +def enabled_tracing(monkeypatch): + monkeypatch.setenv("MELLEA_TRACES_ENABLED", "true") + reset_tracing_state() + yield + reset_tracing_state() + + +@pytest.fixture +def disabled_tracing(monkeypatch): + monkeypatch.delenv("MELLEA_TRACES_ENABLED", raising=False) + reset_tracing_state() + yield + reset_tracing_state() + + +@pytest.fixture +def span_exporter(enabled_tracing): + """Attach an in-memory span exporter to the active tracer provider.""" + if tracing._tracer_provider is None: + pytest.skip("Telemetry not initialized") + exporter = InMemorySpanExporter() + tracing._tracer_provider.add_span_processor(SimpleSpanProcessor(exporter)) + yield exporter + exporter.clear() + + +@pytest.fixture +def adapter_function_plugin(): + return AdapterFunctionTracingPlugin() + + +def _patch_backend_tracer() -> tuple[MagicMock, MagicMock]: + fake_span = MagicMock() + fake_tracer = MagicMock() + fake_tracer.start_span.return_value = fake_span + return fake_span, fake_tracer + + +def _attrs(span: MagicMock) -> dict: + return {c.args[0]: c.args[1] for c in span.set_attribute.call_args_list} + + +def _spans_by_name(exporter: InMemorySpanExporter) -> dict: + tracing._tracer_provider.force_flush() # type: ignore[union-attr] + return {s.name: s for s in exporter.get_finished_spans()} + + +# --------------------------------------------------------------------------- +# Helper-level unit tests (mock tracer) +# --------------------------------------------------------------------------- + + +def test_start_adapter_function_span_stamps_attrs_and_stashes_by_invocation_id( + enabled_tracing, +): + fake_span, fake_tracer = _patch_backend_tracer() + with patch("mellea.telemetry.tracing.get_backend_tracer", return_value=fake_tracer): + start_adapter_function_span( + "inv-1", + name="answerability", + revision="abc123", + binding_type="local_file", + adapter_type="lora", + ) + + fake_tracer.start_span.assert_called_once_with("adapter_function") + assert "inv-1" in tracing._in_flight_spans + assert tracing._in_flight_spans["inv-1"] == (fake_span, None) + attrs = _attrs(fake_span) + assert attrs["mellea.adapter_function.name"] == "answerability" + assert attrs["mellea.adapter_function.revision"] == "abc123" + assert attrs["mellea.adapter_function.binding_type"] == "local_file" + assert attrs["mellea.adapter_function.adapter_type"] == "lora" + + +def test_start_adapter_function_span_omits_none_revision(enabled_tracing): + fake_span, fake_tracer = _patch_backend_tracer() + with patch("mellea.telemetry.tracing.get_backend_tracer", return_value=fake_tracer): + start_adapter_function_span( + "inv-unpinned", + name="answerability", + revision=None, + binding_type="local_file", + adapter_type="lora", + ) + + assert "mellea.adapter_function.revision" not in _attrs(fake_span) + + +def test_finish_adapter_function_span_success_records_outcome(enabled_tracing): + fake_span, fake_tracer = _patch_backend_tracer() + with patch("mellea.telemetry.tracing.get_backend_tracer", return_value=fake_tracer): + start_adapter_function_span( + "inv-2", + name="answerability", + revision="r1", + binding_type="local_file", + adapter_type="lora", + ) + finish_adapter_function_span("inv-2", outcome="success", exception=None) + + fake_span.end.assert_called_once() + assert _attrs(fake_span)["mellea.adapter_function.outcome"] == "success" + fake_span.record_exception.assert_not_called() + assert "inv-2" not in tracing._in_flight_spans + + +def test_finish_adapter_function_span_error_records_exception(enabled_tracing): + fake_span, fake_tracer = _patch_backend_tracer() + err = RuntimeError("boom") + with patch("mellea.telemetry.tracing.get_backend_tracer", return_value=fake_tracer): + start_adapter_function_span( + "inv-err", + name="answerability", + revision="r1", + binding_type="local_file", + adapter_type="lora", + ) + finish_adapter_function_span("inv-err", outcome="error", exception=err) + + fake_span.record_exception.assert_called_once_with(err) + fake_span.set_status.assert_called_once() + attrs = _attrs(fake_span) + assert attrs["mellea.adapter_function.outcome"] == "error" + assert attrs["error.type"] == "RuntimeError" + + +def test_finish_adapter_function_span_no_op_when_not_in_flight(enabled_tracing): + finish_adapter_function_span("never-opened", outcome="success", exception=None) + assert "never-opened" not in tracing._in_flight_spans + + +def test_start_adapter_function_phase_span_stamps_phase_and_revision(enabled_tracing): + fake_span, fake_tracer = _patch_backend_tracer() + with patch("mellea.telemetry.tracing.get_backend_tracer", return_value=fake_tracer): + start_adapter_function_phase_span("inv-3", "prepare", revision="sha123") + + # No invocation "inv-3" in flight, so no explicit parent context is passed. + fake_tracer.start_span.assert_called_once_with( + "adapter_function.prepare", context=None + ) + attrs = _attrs(fake_span) + assert attrs["mellea.adapter_function.phase"] == "prepare" + assert attrs["mellea.adapter_function.revision"] == "sha123" + # Keyed distinctly from the parent's own `_in_flight_spans` entry. + assert "inv-3" not in tracing._in_flight_spans + assert "inv-3:phase:prepare" in tracing._in_flight_spans + + +def test_phase_span_key_does_not_collide_with_parent_or_other_phases(enabled_tracing): + fake_tracer = MagicMock() + fake_tracer.start_span.side_effect = lambda name, context=None: MagicMock(name=name) + with patch("mellea.telemetry.tracing.get_backend_tracer", return_value=fake_tracer): + start_adapter_function_span( + "inv-4", + name="answerability", + revision=None, + binding_type="local_file", + adapter_type="lora", + ) + start_adapter_function_phase_span("inv-4", "activate") + start_adapter_function_phase_span("inv-4", "deactivate") + + assert set(tracing._in_flight_spans) == { + "inv-4", + "inv-4:phase:activate", + "inv-4:phase:deactivate", + } + + +def test_finish_adapter_function_phase_span_closes_and_removes(enabled_tracing): + fake_span, fake_tracer = _patch_backend_tracer() + with patch("mellea.telemetry.tracing.get_backend_tracer", return_value=fake_tracer): + start_adapter_function_phase_span("inv-5", "prepare") + finish_adapter_function_phase_span("inv-5", "prepare") + + fake_span.end.assert_called_once() + assert "inv-5:phase:prepare" not in tracing._in_flight_spans + + +def test_finish_adapter_function_phase_span_no_op_when_not_in_flight(enabled_tracing): + # Contract: a phase never opened (or already closed) is a silent no-op — + # this is what lets finish_adapter_function_span's defensive cleanup run + # unconditionally without double-closing a phase that completed normally. + finish_adapter_function_phase_span("inv-6", "prepare") + + # The no-op must leave the registry untouched, not merely avoid raising. + assert tracing._in_flight_spans == {} + + +def test_finish_adapter_function_span_closes_dangling_phase_span(enabled_tracing): + """A phase that raised (start fired, complete never did) is closed by the invocation's own finish.""" + fake_tracer = MagicMock() + fake_parent_span = MagicMock() + fake_phase_span = MagicMock() + fake_tracer.start_span.side_effect = [fake_parent_span, fake_phase_span] + err = RuntimeError("activation failed") + + with patch("mellea.telemetry.tracing.get_backend_tracer", return_value=fake_tracer): + start_adapter_function_span( + "inv-7", + name="answerability", + revision="r1", + binding_type="local_file", + adapter_type="lora", + ) + start_adapter_function_phase_span("inv-7", "activate") + # No matching finish_adapter_function_phase_span("inv-7", "activate") — + # the phase itself raised. + finish_adapter_function_span("inv-7", outcome="error", exception=err) + + fake_phase_span.record_exception.assert_called_once_with(err) + fake_phase_span.set_status.assert_called_once() + fake_phase_span.end.assert_called_once() + fake_parent_span.end.assert_called_once() + assert "inv-7" not in tracing._in_flight_spans + assert "inv-7:phase:activate" not in tracing._in_flight_spans + + +def test_finish_adapter_function_span_avoids_misattributing_dual_failure( + enabled_tracing, +): + """A body error must not be attached to a failed deactivate child span.""" + fake_tracer = MagicMock() + fake_parent_span = MagicMock() + fake_phase_span = MagicMock() + fake_tracer.start_span.side_effect = [fake_parent_span, fake_phase_span] + body_error = ValueError("body failed") + body_error.add_note("Adapter deactivation also failed: RuntimeError: stop failed") + + with patch("mellea.telemetry.tracing.get_backend_tracer", return_value=fake_tracer): + start_adapter_function_span( + "inv-dual", + name="answerability", + revision="r1", + binding_type="local_file", + adapter_type="lora", + ) + start_adapter_function_phase_span("inv-dual", "deactivate") + finish_adapter_function_span("inv-dual", outcome="error", exception=body_error) + + fake_phase_span.record_exception.assert_not_called() + assert ( + fake_phase_span.set_status.call_args.args[0].description + == "Phase did not complete" + ) + + +def test_helpers_are_silent_when_tracing_disabled(disabled_tracing): + assert ( + start_adapter_function_span( + "inv-d", + name="x", + revision=None, + binding_type="local_file", + adapter_type="lora", + ) + is None + ) + assert start_adapter_function_phase_span("inv-d", "prepare") is None + finish_adapter_function_span("inv-d", outcome="success", exception=None) + finish_adapter_function_phase_span("inv-d", "prepare") + assert tracing._in_flight_spans == {} + + +# --------------------------------------------------------------------------- +# Plugin unit tests (mock tracer) +# --------------------------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_plugin_invocation_start_opens_parent_span( + adapter_function_plugin, enabled_tracing +): + from mellea.plugins.hooks.adapter_function import ( + AdapterFunctionInvocationStartPayload, + ) + + _fake_span, fake_tracer = _patch_backend_tracer() + payload = AdapterFunctionInvocationStartPayload( + adapter_function_invocation_id="p-inv-1", + name="answerability", + revision="r1", + binding_type="local_file", + adapter_type="lora", + ) + with patch("mellea.telemetry.tracing.get_backend_tracer", return_value=fake_tracer): + await adapter_function_plugin.on_invocation_start(payload, {}) + + fake_tracer.start_span.assert_called_once_with("adapter_function") + assert "p-inv-1" in tracing._in_flight_spans + + +@pytest.mark.asyncio +async def test_plugin_invocation_complete_closes_parent_span( + adapter_function_plugin, enabled_tracing +): + from mellea.plugins.hooks.adapter_function import ( + AdapterFunctionInvocationCompletePayload, + AdapterFunctionInvocationStartPayload, + ) + + fake_span, fake_tracer = _patch_backend_tracer() + start_payload = AdapterFunctionInvocationStartPayload( + adapter_function_invocation_id="p-inv-2", + name="answerability", + revision="r1", + binding_type="local_file", + adapter_type="lora", + ) + complete_payload = AdapterFunctionInvocationCompletePayload( + adapter_function_invocation_id="p-inv-2", + name="answerability", + revision="r1", + binding_type="local_file", + adapter_type="lora", + outcome="success", + ) + with patch("mellea.telemetry.tracing.get_backend_tracer", return_value=fake_tracer): + await adapter_function_plugin.on_invocation_start(start_payload, {}) + await adapter_function_plugin.on_invocation_complete(complete_payload, {}) + + fake_span.end.assert_called_once() + assert "p-inv-2" not in tracing._in_flight_spans + + +@pytest.mark.asyncio +async def test_plugin_phase_start_and_complete_open_and_close_child_span( + adapter_function_plugin, enabled_tracing +): + from mellea.plugins.hooks.adapter_function import ( + AdapterFunctionPhaseCompletePayload, + AdapterFunctionPhaseStartPayload, + ) + + fake_span, fake_tracer = _patch_backend_tracer() + start_payload = AdapterFunctionPhaseStartPayload( + adapter_function_invocation_id="p-inv-3", + name="answerability", + phase="activate", + revision="sha1", + ) + complete_payload = AdapterFunctionPhaseCompletePayload( + adapter_function_invocation_id="p-inv-3", + name="answerability", + phase="activate", + duration_ms=5.0, + ) + with patch("mellea.telemetry.tracing.get_backend_tracer", return_value=fake_tracer): + await adapter_function_plugin.on_phase_start(start_payload, {}) + await adapter_function_plugin.on_phase_complete(complete_payload, {}) + + # No invocation "p-inv-3" in flight, so no explicit parent context is passed. + fake_tracer.start_span.assert_called_once_with( + "adapter_function.activate", context=None + ) + fake_span.end.assert_called_once() + assert "p-inv-3:phase:activate" not in tracing._in_flight_spans + + +@pytest.mark.asyncio +async def test_plugin_ignores_uncorrelated_phase_complete(adapter_function_plugin): + """A metric-only completion must not be treated as a trace lifecycle event.""" + from mellea.plugins.hooks.adapter_function import ( + AdapterFunctionPhaseCompletePayload, + ) + + payload = AdapterFunctionPhaseCompletePayload( + name="answerability", phase="activate", duration_ms=5.0 + ) + + with patch( + "mellea.telemetry.tracing.finish_adapter_function_phase_span" + ) as finish_phase: + await adapter_function_plugin.on_phase_complete(payload, {}) + + finish_phase.assert_not_called() + + +# --------------------------------------------------------------------------- +# Integration: real call sites + real OTel SDK + real adapter lifecycle +# --------------------------------------------------------------------------- + + +class _Contract(IOContract): + def build_prompt(self, **kwargs: object) -> Component: + raise NotImplementedError + + def parse(self, raw: str) -> dict[str, object]: + return {} + + +def _make_scope_adapter(): + weights = MagicMock(spec=LocalFileBinding) + weights.binding_type = "local_file" + weights.revision = "abc123" + weights.resolved_revision.return_value = "abc123" + identity = Identity(name="answerability", adapter_type="lora") + return Adapter(identity=identity, io_contract=_Contract(), weights=weights) + + +@pytest.mark.integration +def test_activate_deactivate_emits_nested_span_tree(span_exporter): + """`adapter_scope` emits `adapter_function` > `adapter_function.{activate,deactivate}`. + + Nesting here is via explicit `trace.set_span_in_context` parenting (see + `start_adapter_function_phase_span`), not ambient-context attach, so unlike + every other span family in this codebase it holds on Python 3.11 too — no + `_CONTEXT_ATTACH_SUPPORTED` gating needed. + """ + adapter = _make_scope_adapter() + mock_backend = MagicMock(spec=AdapterMixin) + + with AdapterMixin.adapter_scope(mock_backend, adapter): + pass + + by_name = _spans_by_name(span_exporter) + assert "adapter_function" in by_name + assert "adapter_function.activate" in by_name + assert "adapter_function.deactivate" in by_name + + parent = by_name["adapter_function"] + activate_span = by_name["adapter_function.activate"] + deactivate_span = by_name["adapter_function.deactivate"] + + assert parent.parent is None + assert parent.attributes is not None + assert parent.attributes.get("mellea.adapter_function.name") == "answerability" + assert parent.attributes.get("mellea.adapter_function.outcome") == "success" + + assert activate_span.parent is not None + assert activate_span.parent.span_id == parent.context.span_id + assert deactivate_span.parent is not None + assert deactivate_span.parent.span_id == parent.context.span_id + + assert tracing._in_flight_spans == {} + + +@pytest.mark.integration +def test_activate_raising_still_drains_registry_and_marks_error(span_exporter): + """A phase that raises still gets its span closed via the invocation's own finish.""" + adapter = _make_scope_adapter() + mock_backend = MagicMock(spec=AdapterMixin) + adapter.weights.activate.side_effect = RuntimeError("activation failed") # type: ignore[union-attr] + + with pytest.raises(RuntimeError, match="activation failed"): + with AdapterMixin.adapter_scope(mock_backend, adapter): + pytest.fail("body must not run when activate() raises") + + by_name = _spans_by_name(span_exporter) + assert "adapter_function" in by_name + assert "adapter_function.activate" in by_name + # activate() raised before deactivate ever ran. + assert "adapter_function.deactivate" not in by_name + + from opentelemetry.trace import StatusCode + + assert by_name["adapter_function"].status.status_code == StatusCode.ERROR + assert by_name["adapter_function.activate"].status.status_code == StatusCode.ERROR + + # The registry drains to zero even though the phase never fired its own + # completion hook — finish_adapter_function_span's defensive cleanup closed it. + assert tracing._in_flight_spans == {}