Render rerank observability (0060b)

Render the typed rerank events in both bundled observers, replacing the
safe-skip guards: the OTel openarmature.rerank.complete span
(observability §5.5.13) and the Langfuse Retriever observation (§8.4.7,
a new retriever() client/adapter surface + the searchUnits usageDetails
convention on LangfuseUsage). Usage figures emit conditionally per
proposal 0093; the span renders the scored results
(RerankEvent.output_results per 0089) and is not gated by
disable_llm_spans.

Un-defer conformance fixtures 099-108 / 138 / 141 / 142 (109 metrics
rides 0067). conformance.toml 0060 and 0089 move to implemented,
completing the rerank capability and the 0089 output-rendering surface.

Author: chris-colinsky

conformance.toml

@@ -612,9 +612,9 @@ note = "Retrieval-provider embedding capability. The EmbeddingProvider protocol
 # variants + OTel ``openarmature.rerank.complete`` span / Langfuse Retriever
 # observation / rerank metrics.  Sibling to the embedding surface (0059).
 [proposals."0060"]
-status = "partial"
+status = "implemented"
 since = "0.16.0"
-note = "Retrieval-provider rerank protocol. The RerankProvider protocol (retrieval-provider §5) + the RerankResponse / ScoredDocument / RerankUsage / RerankRuntimeConfig response types (§6) + the RerankRuntimeConfig runtime config (§2) ship, alongside the CohereRerankProvider reference reranker against POST /v2/rerank (§8.4, Cohere-shape): the request body is {model, query, documents (string array), top_n from top_k}, return_documents is a silent no-op (the Cohere wire has no such field), and max_tokens_per_doc rides the extras pass-through bag. The provider parses {id, model?, results: [{index, relevance_score, document?}], meta.billed_units.{search_units, input_tokens}}, sorts results by relevance_score descending, reads the document echo only when present (never auto-filled), builds a RerankUsage record only when the provider surfaces >= 1 usage figure (else usage = null, never fabricated), and enforces the §6 invariants -- valid index into the input documents, no duplicate index, len(results) <= top_k when supplied -- raising provider_invalid_response otherwise. Pre-send validation raises provider_invalid_request on an empty query, an empty documents list, or top_k <= 0 (top_k MAY exceed len(documents)). The provider dispatches the typed RerankEvent / RerankFailedEvent (graph-engine §6) per rerank() call, mutually exclusive, alongside the §7-category exception on failure; RerankEvent.output_results is populated unconditionally on success (proposal 0089). The §7 categories reuse the shared llm-provider taxonomy. Protocol fixtures 006-012 pass. partial because the observability rendering (OTel §5.5.13 openarmature.rerank.complete span + Langfuse §8.4.7 Retriever observation + the rerank-metrics surface; fixtures 099-109 / 138 / 141 / 142) lands in the 0060b follow-on -- the bundled OTel + Langfuse observers safely skip the rerank events for now. Wire-format details (0090 Cohere rerank fixtures 028-031) are a separate proposal."
+note = "Retrieval-provider rerank protocol. The RerankProvider protocol (retrieval-provider §5) + the RerankResponse / ScoredDocument / RerankUsage / RerankRuntimeConfig response types (§6) + the RerankRuntimeConfig runtime config (§2) ship, alongside the CohereRerankProvider reference reranker against POST /v2/rerank (§8.4, Cohere-shape): the request body is {model, query, documents (string array), top_n from top_k}, return_documents is a silent no-op (the Cohere wire has no such field), and max_tokens_per_doc rides the extras pass-through bag. The provider parses {id, model?, results: [{index, relevance_score, document?}], meta.billed_units.{search_units, input_tokens}}, sorts results by relevance_score descending, reads the document echo only when present (never auto-filled), builds a RerankUsage record only when the provider surfaces >= 1 usage figure (else usage = null, never fabricated), and enforces the §6 invariants -- valid index into the input documents, no duplicate index, len(results) <= top_k when supplied -- raising provider_invalid_response otherwise. Pre-send validation raises provider_invalid_request on an empty query, an empty documents list, or top_k <= 0 (top_k MAY exceed len(documents)). The provider dispatches the typed RerankEvent / RerankFailedEvent (graph-engine §6) per rerank() call, mutually exclusive, alongside the §7-category exception on failure; RerankEvent.output_results is populated unconditionally on success (proposal 0089). The §7 categories reuse the shared llm-provider taxonomy. Both bundled observers render the rerank events (0060b): the OTelObserver emits an openarmature.rerank.complete span (observability §5.5.13, SpanKind.CLIENT) parented lineage-aware under the calling node, carrying the GenAI semconv subset (gen_ai.system / request.model / response.model / response.id + the conditional usage.input_tokens, gated by disable_genai_semconv) + the OA openarmature.rerank.query_length (UTF-8 bytes) / .document_count / .top_k (omitted when null) / .result_count identity attrs + the conditional .search_units + the payload-gated (disable_provider_payload) .query / .documents / .results (the results sourced from RerankEvent.output_results per proposal 0089) -- no gen_ai.operation.name (deferred, no upstream rerank coverage), not gated by disable_llm_spans; failure renders ERROR status + error.type + an exception event + openarmature.error.category. The LangfuseObserver renders a dedicated Retriever observation (asType retriever, §8.4.7) with model = response_model-or-model, usageDetails.{input, searchUnits} (empty map when usage = null), the openarmature_query_length / _document_count / _top_k / _result_count / _response_id metadata, and the payload-gated input {query, documents} + output results; failure renders an ERROR-level Retriever observation with error_category as statusMessage, error_type / error_message in metadata, and no output. Protocol fixtures 006-012 + observability fixtures 099-108 / 138 / 141 / 142 pass. The §11 rerank-metrics path (fixture 109) is proposal 0067's metrics surface (one GenAI instrument across LLM + embedding + rerank calls), tracked and deferred under 0067 (cf. 089), not part of 0060's own observability surface. Wire-format details (0090 Cohere rerank fixtures 028-031) are a separate proposal."
 
 # Spec v0.55.0 (proposal 0065; repo pins v0.55.1).  Failure-isolation
 # cause fidelity at non-node placements (pipeline-utilities §6.3 /
@@ -856,9 +856,9 @@ status = "not-yet"
 # (graph-engine §6 EmbeddingEvent.output_vectors / RerankEvent.
 # output_results + observability output mappings).
 [proposals."0089"]
-status = "partial"
+status = "implemented"
 since = "0.16.0"
-note = "The embedding half shipped in 0059b (#198): EmbeddingEvent.output_vectors is populated by OpenAIEmbeddingProvider and rendered as the Langfuse Embedding observation output (observability §8.4.5), sourced event-side per 0089; observability fixtures 083 (re-sourced) + 137 pass. partial because the rerank half is unshipped -- RerankEvent.output_results and the rerank observation output rendering ride the rerank capability (proposal 0060, not-yet); observability fixtures 108 + 138 defer with it."
+note = "The embedding half shipped in 0059b (#198): EmbeddingEvent.output_vectors is populated by OpenAIEmbeddingProvider and rendered as the Langfuse Embedding observation output (observability §8.4.5), sourced event-side per 0089; observability fixtures 083 (re-sourced) + 137 pass. The rerank half now ships with the 0060b rerank observability: RerankEvent.output_results is populated unconditionally on success by CohereRerankProvider and rendered as both the OTel openarmature.rerank.results attribute (observability §5.5.13, payload-gated) and the Langfuse Retriever observation output (§8.4.7), sourced event-side per 0089; observability fixtures 108 + 138 pass. This completes 0089 across the embedding + rerank surfaces."
 
 # Spec v0.85.0 (proposal 0090).  Cohere rerank wire mapping
 # (retrieval-provider §8.4, POST /v2/rerank; backs 0060's Cohere-shaped

src/openarmature/observability/langfuse/adapter.py

@@ -519,6 +519,65 @@ def embedding(
             )
         return _SpanHandle(obs)
 
+    def retriever(
+        self,
+        *,
+        trace_id: str,
+        name: str | None = None,
+        model: str | None = None,
+        usage: LangfuseUsage | None = None,
+        metadata: dict[str, Any] | None = None,
+        parent_observation_id: str | None = None,
+        level: ObservationLevel = "DEFAULT",
+        status_message: str | None = None,
+        input: Any = None,
+        output: Any = None,
+        start_time: datetime | None = None,
+    ) -> LangfuseSpanHandle:
+        # v4 unifies observations under start_observation(as_type=); a Retriever
+        # observation routes through as_type="retriever" (proposal 0060), the
+        # SDK's native LangfuseRetriever span. It carries model + usage like an
+        # Embedding but is a distinct type; usage maps to v4's usage_details.
+        # The rerank usageDetails convention (§8.4.7) exposes input tokens under
+        # "input" and search-units under "searchUnits". Back-date via the
+        # private OTel tracer when start_time is supplied, exactly as
+        # generation()/tool()/embedding() do.
+        extra_kwargs: dict[str, Any] = {"model": model, "input": input, "output": output}
+        if usage is not None:
+            usage_details: dict[str, int] = {}
+            if usage.input is not None:
+                usage_details["input"] = usage.input
+            if usage.search_units is not None:
+                usage_details["searchUnits"] = usage.search_units
+            extra_kwargs["usage_details"] = usage_details
+        present_extra = {k: v for k, v in extra_kwargs.items() if v is not None}
+        if start_time is not None:
+            from langfuse._client.span import LangfuseRetriever
+
+            obs = self._start_back_dated_observation(
+                LangfuseRetriever,
+                trace_id=trace_id,
+                name=name,
+                metadata=metadata,
+                parent_observation_id=parent_observation_id,
+                level=level,
+                status_message=status_message,
+                start_time=start_time,
+                **present_extra,
+            )
+        else:
+            obs = self._start_observation(
+                as_type="retriever",
+                trace_id=trace_id,
+                name=name,
+                metadata=metadata,
+                parent_observation_id=parent_observation_id,
+                level=level,
+                status_message=status_message,
+                **present_extra,
+            )
+        return _SpanHandle(obs)
+
     def _start_back_dated_observation(
         self,
         observation_cls: type[Any],

src/openarmature/observability/langfuse/client.py

@@ -33,7 +33,7 @@
 from datetime import datetime
 from typing import Any, Literal, Protocol, runtime_checkable
 
-ObservationType = Literal["span", "generation", "event", "tool", "embedding"]
+ObservationType = Literal["span", "generation", "event", "tool", "embedding", "retriever"]
 
 # Langfuse-supported `level` values per spec §8.4.2 (statusMessage pair).
 ObservationLevel = Literal["DEFAULT", "DEBUG", "INFO", "WARNING", "ERROR"]
@@ -46,6 +46,12 @@ class LangfuseUsage:
     input: int | None = None
     output: int | None = None
     total: int | None = None
+    # Rerank observability (proposal 0060, observability §8.4.7): Langfuse's
+    # usageDetails is an open-shape map, and the OA convention adds a
+    # `searchUnits` figure for the Retriever observation. Modeled here as a
+    # snake_case field; the SDK adapter maps it to the camelCase `searchUnits`
+    # usage_details key. None when the provider reported no search-units count.
+    search_units: int | None = None
 
 
 @dataclass
@@ -297,6 +303,35 @@ def embedding(
         the caller ``.end()``s at outcome time."""
         ...
 
+    # Proposal 0060 rerank observability: a dedicated Retriever observation
+    # (Langfuse asType="retriever", spec observability §8.4.7), NOT a
+    # Generation with a discriminator. Langfuse positions Retriever for data
+    # retrieval steps (broad enough to cover reranking); the `usage` record's
+    # `search_units` / `input` figures render into the open-shape usageDetails
+    # map (the OA `searchUnits` key convention is defined in §8.4.7).
+    def retriever(
+        self,
+        *,
+        trace_id: str,
+        name: str | None = None,
+        model: str | None = None,
+        usage: LangfuseUsage | None = None,
+        metadata: dict[str, Any] | None = None,
+        parent_observation_id: str | None = None,
+        level: ObservationLevel = "DEFAULT",
+        status_message: str | None = None,
+        input: Any = None,
+        output: Any = None,
+        start_time: datetime | None = None,
+    ) -> LangfuseSpanHandle:
+        """Open a dedicated Retriever observation. Like :meth:`embedding`
+        for the model / usage surface but a distinct observation type:
+        carries ``model`` / ``usage`` (input-token + search-unit figures) /
+        ``input`` (the query + documents) / ``output`` (the scored results) /
+        metadata / level. Returns a minimal handle the caller ``.end()``s at
+        outcome time."""
+        ...
+
     def force_flush(self, timeout_ms: int = 30_000) -> bool:
         """Flush any pending outbound buffer in the underlying sink.
 
@@ -591,6 +626,39 @@ def embedding(
         trace.observations.append(observation)
         return _InMemorySpanHandle(observation=observation)
 
+    def retriever(
+        self,
+        *,
+        trace_id: str,
+        name: str | None = None,
+        model: str | None = None,
+        usage: LangfuseUsage | None = None,
+        metadata: dict[str, Any] | None = None,
+        parent_observation_id: str | None = None,
+        level: ObservationLevel = "DEFAULT",
+        status_message: str | None = None,
+        input: Any = None,
+        output: Any = None,
+        start_time: datetime | None = None,
+    ) -> LangfuseSpanHandle:
+        trace = self._get_trace(trace_id)
+        observation = LangfuseObservation(
+            id=self._mint_observation_id(),
+            type="retriever",
+            name=name,
+            metadata=dict(metadata) if metadata is not None else {},
+            parent_observation_id=parent_observation_id,
+            level=level,
+            status_message=status_message,
+            input=input,
+            output=output,
+            model=model,
+            usage=usage,
+            start_time=start_time,
+        )
+        trace.observations.append(observation)
+        return _InMemorySpanHandle(observation=observation)
+
     def force_flush(self, timeout_ms: int = 30_000) -> bool:
         # In-memory recorder has no outbound buffer; every observation
         # is captured synchronously on its create call. The ``timeout_ms``