Skip to content

Fix/HYBIM-961 retrieval and agent span semantics #210

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
pradystar wants to merge 4 commits into
base: main
Choose a base branch
from
Open

Fix/HYBIM-961 retrieval and agent span semantics #210

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
541de2d
fix(otel): align retrieval and remote-agent span semantics
pradystar Aug 6, 2026
8e02dc1
fix(otel): treat empty data source IDs as absent
pradystar Aug 6, 2026
94faa32
fix(otel): preserve retriever display names without data source IDs
pradystar Aug 6, 2026
557f420
address review comments
pradystar Aug 6, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
8 changes: 8 additions & 0 deletions CHANGELOG.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -7,11 +7,19 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0

## [Unreleased]

### Added

- Retriever logging accepts an explicit `data_source_id`, and agent logging
accepts `span_kind=SpanKind.CLIENT` for calls to remote agent services.

### Fixed

- Agent Control spans exported over OTLP now include the control discriminator
and complete `agent_control.*` field set required for backend classification
and Controls-card rendering.
- Retriever spans exported over OTLP now use client operation semantics and are
named `retrieval {data_source_id}` when an explicit data-source ID is supplied,
falling back to the captured display name and then to `retrieval`.

## [0.1.1] - 2026-08-03

Expand Down
1 change: 1 addition & 0 deletions src/splunk_ao/converter/attribute_mapping.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -423,6 +423,7 @@ def set_retriever_attributes(attrs: MutableMapping[str, AttributeValue], span: R
attrs["gen_ai.retrieval.documents"] = _content_value(span.output)
attrs["splunk_ao.retrieval.documents.count"] = len(span.output)
attrs["db.operation"] = "search"
_set_if_present(attrs, "gen_ai.data_source.id", _field(span, "data_source_id"))

@fercor-cisco fercor-cisco Aug 6, 2026

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🟡 minor (bug): gen_ai.data_source.id is emitted but never registered in SPLUNK_ALIAS_BY_GEN_AI, so normalize_attributes_for_export will not produce a splunk_ao.* mirror for it. Every other gen_ai.* attribute this function emits has an alias — including the two immediate neighbours here, gen_ai.retrieval.query.text and gen_ai.retrieval.top_k. The result is that with normalization enabled (SPLUNK_AO_DEV_ENABLE_ATTRIBUTE_NORMALIZATION), the new authoritative retrieval identity is the only retrieval field with no Splunk-namespaced counterpart, so any backend or dashboard consuming the splunk_ao.retrieval.* set silently sees no data source ID.

Impact is limited today because normalization is opt-in and off by default, which is why I'm rating this minor rather than major — but it will become a silent data gap the moment normalization is turned on. Add the alias next to the other retrieval entries. test_every_alias_uses_an_explicit_destination_namespace only checks namespacing, not coverage, so nothing currently catches this; consider a companion assertion that every gen_ai.* key produced by build_span_attributes appears in the alias map.

Suggested change
_set_if_present(attrs, "gen_ai.data_source.id", _field(span, "data_source_id"))
_set_if_present(attrs, "gen_ai.data_source.id", _field(span, "data_source_id"))
# and in SPLUNK_ALIAS_BY_GEN_AI, alongside the other retrieval entries:
# "gen_ai.data_source.id": "splunk_ao.data_source.id",

🤖 Generated by the Astra agent

@pradystar pradystar Aug 6, 2026

Copy link
Copy Markdown
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

gen_ai.data_source.id must remain standard only, the normalization behavior will be removed fully in future

requested_top_k = _field(span, "num_documents")
_set_if_present(attrs, "gen_ai.retrieval.top_k", requested_top_k)

Expand Down
20 changes: 14 additions & 6 deletions src/splunk_ao/converter/span_converter.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -20,16 +20,12 @@
_NAME_PARTS: dict[StepType, tuple[str, str]] = {
StepType.llm: ("chat", "model"),
StepType.tool: ("execute_tool", "name"),
StepType.retriever: ("retrieval", "name"),
StepType.retriever: ("retrieval", "data_source_id"),
StepType.workflow: ("invoke_workflow", "name"),
StepType.agent: ("invoke_agent", "name"),
StepType.control: ("", "name"),
}

_KIND_BY_STEP_TYPE = {
step_type: SpanKind.CLIENT if step_type is StepType.llm else SpanKind.INTERNAL for step_type in _NAME_PARTS
}


def _step_type(span: BaseStep) -> StepType:
raw_type = getattr(span, "type", None)
Expand All @@ -48,9 +44,21 @@ def _step_type(span: BaseStep) -> StepType:
def _span_name(span: BaseStep, step_type: StepType) -> str:
prefix, field_name = _NAME_PARTS[step_type]
detail = getattr(span, field_name, None)
if step_type is StepType.retriever:
if detail is not None:
return f"{prefix} {detail}"
detail = getattr(span, "name", None)
return " ".join(part for part in (prefix, str(detail).strip() if detail is not None else "") if part)
Comment on lines +47 to 51

@fercor-cisco fercor-cisco Aug 6, 2026

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🟡 minor (bug): The retriever branch returns f"{prefix} {detail}" without stripping, while the shared fallback path below it does str(detail).strip(). Combined with normalize_data_source_id, which only maps the exact empty string to None, a whitespace-only or padded ID leaks straight through:

  • data_source_id=" " → span name "retrieval " (trailing whitespace) and attribute gen_ai.data_source.id == " ".
  • data_source_id=" kb " → span name "retrieval kb ".

That is the exact class of malformed name test_missing_optional_name_parts_do_not_leave_whitespace exists to prevent — it just never exercises the new data_source_id path. Two options:

  1. Strip in the converter only (below). Cheap, but the attribute still carries the untrimmed value, so name and attribute disagree.
  2. Preferred: strip in LoggedRetrieverSpan.normalize_data_source_id and treat the result as absent when empty, so name and attribute stay consistent and the empty-vs-whitespace cases collapse into one rule:
@field_validator("data_source_id", mode="before")
@classmethod
def normalize_data_source_id(cls, value: object) -> object:
    if isinstance(value, str):
        stripped = value.strip()
        return stripped or None
    return value

Note option 2 changes the intent of test_retriever_name_and_attribute_use_explicit_data_source_id_byte_for_byte; that test's case ("knowledge base/v1") still passes since it has no leading/trailing whitespace, but please add a whitespace-only case either way.

Suggested change
if step_type is StepType.retriever:
if detail is not None:
return f"{prefix} {detail}"
detail = getattr(span, "name", None)
return " ".join(part for part in (prefix, str(detail).strip() if detail is not None else "") if part)
if step_type is StepType.retriever:
if detail is not None:
return f"{prefix} {str(detail).strip()}"
detail = getattr(span, "name", None)

🤖 Generated by the Astra agent

@pradystar pradystar Aug 6, 2026

Copy link
Copy Markdown
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Agreed whitespace only IDs are now absent. Padded IDs are trimmed consistently before use in both the span name and attribute.

Comment on lines 46 to 51

@fercor-cisco fercor-cisco Aug 7, 2026

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🟡 minor (bug): The retriever branch reads data_source_id with a bare getattr, but set_retriever_attributes reads the same field via _field() (attribute_mapping.py:426), which falls back to model_extra. For any span carrying data_source_id only as a model extra — or a duck-typed source object like the SimpleNamespace(**fields, model_extra={...}) pattern already exercised in test_control_mapping_accepts_schema_compatible_control_span — the attribute gen_ai.data_source.id is emitted while the span name silently falls back to retrieval {display_name}. Name and identity then disagree for the same span, which is exactly the coupling this PR is trying to establish.

Not reachable through add_retriever_span today (it always constructs a real LoggedRetrieverSpan), so minor. Using the same accessor in both places removes the divergence by construction. Note this also means the returned value is no longer guaranteed to be a normalized string, so keep the .strip()/str() handling on it.

Suggested change
detail = getattr(span, field_name, None)
if step_type is StepType.retriever:
if detail is not None:
return f"{prefix} {detail}"
detail = getattr(span, "name", None)
return " ".join(part for part in (prefix, str(detail).strip() if detail is not None else "") if part)
def _span_name(span: BaseStep, step_type: StepType) -> str:
prefix, field_name = _NAME_PARTS[step_type]
detail = getattr(span, field_name, None)
if step_type is StepType.retriever:
detail = _field(span, field_name)
if detail is None:
detail = getattr(span, "name", None)
return " ".join(part for part in (prefix, str(detail).strip() if detail is not None else "") if part)

🤖 Generated by the Astra agent



def _span_kind(span: BaseStep, step_type: StepType) -> SpanKind:
if step_type in (StepType.llm, StepType.retriever):
return SpanKind.CLIENT
if step_type is StepType.agent and getattr(span, "span_kind", None) is SpanKind.CLIENT:
return SpanKind.CLIENT
return SpanKind.INTERNAL


def _to_unix_ns(value: datetime) -> int:
if value.tzinfo is None:
value = value.replace(tzinfo=UTC)
Expand Down Expand Up @@ -98,7 +106,7 @@ def convert_span(
attributes=build_span_attributes(span, session_id),
events=(),
links=(),
kind=_KIND_BY_STEP_TYPE[step_type],
kind=_span_kind(span, step_type),
instrumentation_scope=_INSTRUMENTATION_SCOPE,
status=_span_status(span),
start_time=start_time_ns,
Expand Down
64 changes: 57 additions & 7 deletions src/splunk_ao/decorator.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -55,6 +55,7 @@ def call_llm(prompt, temperature=0.7):
from types import TracebackType
from typing import Any, TypeVar, cast, overload

from opentelemetry.trace import SpanKind
from typing_extensions import ParamSpec

from galileo_core.schemas.logging.span import WorkflowSpan
Expand Down Expand Up @@ -276,6 +277,8 @@ def log(
name: str | None = None,
span_type: SPAN_TYPE | None = None,
params: dict[str, str | Callable] | None = None,
data_source_id: str | None = None,
span_kind: SpanKind | None = None,
) -> Callable[[Callable[P, R]], Callable[P, R]]: ...

def log(
Expand All @@ -286,6 +289,8 @@ def log(
span_type: SPAN_TYPE | None = None,
params: dict[str, str | Callable] | None = None,
dataset_record: DatasetRecord | None = None,
data_source_id: str | None = None,
span_kind: SpanKind | None = None,
) -> Callable[[Callable[P, R]], Callable[P, R]]:
"""
Main decorator function for logging function calls.
Expand All @@ -304,33 +309,66 @@ def log(
Optional span type ("llm", "retriever", "tool", "workflow", "agent")
params
Optional parameter mapping for extracting specific values
data_source_id
Optional authoritative retrieval data-source ID for ``span_type="retriever"``.
span_kind
Optional OTel kind for ``span_type="agent"``. Only ``SpanKind.CLIENT`` marks a remote agent call.
dataset_record
Optional parameter for dataset values. This is used by the local experiment module to set the dataset fields on the trace/spans and not generally provided for logging to log streams.

Returns
-------
A decorated function that logs its execution
"""
explicit_span_params = {
key: value
for key, value in {"data_source_id": data_source_id, "span_kind": span_kind}.items()
if value is not None
}

def decorator(func: Callable[P, R]) -> Callable[P, R]:
if inspect.isasyncgenfunction(func):
return cast(
Callable[P, R],
self._async_generator_log(
func, name=name, span_type=span_type, params=params, dataset_record=dataset_record
func,
name=name,
span_type=span_type,
params=params,
dataset_record=dataset_record,
explicit_span_params=explicit_span_params,
),
)
if inspect.isgeneratorfunction(func):
return cast(
Callable[P, R],
self._sync_generator_log(
func, name=name, span_type=span_type, params=params, dataset_record=dataset_record
func,
name=name,
span_type=span_type,
params=params,
dataset_record=dataset_record,
explicit_span_params=explicit_span_params,
),
)
wrapped = (
self._async_log(func, name=name, span_type=span_type, params=params, dataset_record=dataset_record)
self._async_log(
func,
name=name,
span_type=span_type,
params=params,
dataset_record=dataset_record,
explicit_span_params=explicit_span_params,
)
if asyncio.iscoroutinefunction(func)
else self._sync_log(func, name=name, span_type=span_type, params=params, dataset_record=dataset_record)
else self._sync_log(
func,
name=name,
span_type=span_type,
params=params,
dataset_record=dataset_record,
explicit_span_params=explicit_span_params,
)
)
return cast(Callable[P, R], wrapped)

Expand All @@ -348,6 +386,7 @@ def _async_log(
span_type: SPAN_TYPE | None,
params: dict[str, str | Callable] | None = None,
dataset_record: DatasetRecord | None = None,
explicit_span_params: dict[str, Any] | None = None,
) -> F:
"""
Internal method to handle logging for async functions.
Expand Down Expand Up @@ -384,6 +423,7 @@ async def async_wrapper(*args: Any, **kwargs: Any) -> Any:
name=name or func.__name__,
span_type=span_type,
params=params,
explicit_span_params=explicit_span_params,
is_method=self._is_method(func),
func_args=args,
func_kwargs=kwargs,
Expand Down Expand Up @@ -415,6 +455,7 @@ def _sync_log(
span_type: SPAN_TYPE | None,
params: dict[str, str | Callable] | None = None,
dataset_record: DatasetRecord | None = None,
explicit_span_params: dict[str, Any] | None = None,
) -> F:
"""
Internal method to handle logging for synchronous functions.
Expand Down Expand Up @@ -442,6 +483,7 @@ def sync_wrapper(*args: Any, **kwargs: Any) -> Any:
name=name or func.__name__,
span_type=span_type,
params=params,
explicit_span_params=explicit_span_params,
is_method=self._is_method(func),
func_args=args,
func_kwargs=kwargs,
Expand Down Expand Up @@ -473,6 +515,7 @@ def _sync_generator_log(
span_type: SPAN_TYPE | None,
params: dict[str, str | Callable] | None = None,
dataset_record: DatasetRecord | None = None,
explicit_span_params: dict[str, Any] | None = None,
) -> F:
@wraps(func)
def generator_wrapper(*args: Any, **kwargs: Any) -> Generator:
Expand All @@ -481,6 +524,7 @@ def generator_wrapper(*args: Any, **kwargs: Any) -> Generator:
name=name or func.__name__,
span_type=span_type,
params=params,
explicit_span_params=explicit_span_params,
is_method=self._is_method(func),
func_args=args,
func_kwargs=kwargs,
Expand All @@ -500,6 +544,7 @@ def _async_generator_log(
span_type: SPAN_TYPE | None,
params: dict[str, str | Callable] | None = None,
dataset_record: DatasetRecord | None = None,
explicit_span_params: dict[str, Any] | None = None,
) -> F:
@wraps(func)
def async_generator_wrapper(*args: Any, **kwargs: Any) -> AsyncGenerator:
Expand All @@ -508,6 +553,7 @@ def async_generator_wrapper(*args: Any, **kwargs: Any) -> AsyncGenerator:
name=name or func.__name__,
span_type=span_type,
params=params,
explicit_span_params=explicit_span_params,
is_method=self._is_method(func),
func_args=args,
func_kwargs=kwargs,
Expand Down Expand Up @@ -543,6 +589,7 @@ def _prepare_input(
name: str,
span_type: SPAN_TYPE | None,
params: dict[str, str | Callable] | None = None,
explicit_span_params: dict[str, Any] | None = None,
is_method: bool = False,
func_args: tuple = (),
func_kwargs: dict | None = None,
Expand Down Expand Up @@ -606,6 +653,8 @@ def _prepare_input(
if param_name in input_ and param_name not in span_params:
span_params[param_name] = input_[param_name]

span_params.update(explicit_span_params or {})

if "name" not in span_params:
span_params["name"] = name

Expand Down Expand Up @@ -685,10 +734,10 @@ def _get_span_param_names(self, span_type: SPAN_TYPE) -> list[str]:
common_params = ["name", "input", "metadata", "tags"]
span_params = {
"llm": [*common_params, "model", "temperature", "tools"],
"retriever": common_params,
"retriever": [*common_params, "data_source_id"],
"tool": [*common_params, "tool_call_id"],
"workflow": common_params,
"agent": [*common_params, "agent_type"],
"agent": [*common_params, "agent_type", "span_kind"],
Comment on lines +737 to +740

@fercor-cisco fercor-cisco Aug 7, 2026

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🟡 minor (bug): Adding data_source_id and span_kind to the auto-mapped lists means these are now populated from any decorated function parameter with a matching name (_prepare_input copies input_[param_name] into span_params at line 653-654), not just from the new decorator kwargs.

Two consequences worth weighing:

  1. data_source_id: a coincidentally-named parameter of the wrong type — e.g. @log(span_type="retriever") def search(q: str, data_source_id: UUID) — makes LoggedRetrieverSpan(...) raise ValidationError (the mode="before" validator passes non-str through untouched, and Pydantic won't coerce UUID/int to str). add_retriever_span is wrapped in @warn_catch_exception(exceptions=(Exception,)), so it returns None and the entire span is dropped with only a warning. Previously retriever auto-mapped only common_params, so a same-named argument was simply ignored.
  2. span_kind: a user parameter named span_kind holding anything other than the literal SpanKind.CLIENT enum member is silently coerced to INTERNAL (see the validator thread).

This mirrors how model/temperature/agent_type already behave, so it's consistent rather than novel — but those are descriptive metadata, whereas these two drive OTel span identity and kind. Either coerce defensively in the validator (str(value) for non-None non-str), or restrict these two hints to explicit_span_params only and drop them from _get_span_param_names, so a name collision can't degrade or destroy a span.

🤖 Generated by the Astra agent

}
return span_params.get(span_type, common_params)

Expand Down Expand Up @@ -791,8 +840,9 @@ def _prepare_call(
created_at = span_params.get("created_at", _get_timestamp())
if span_type == "agent":
agent_type = span_params.get("agent_type")
span_kind = span_params.get("span_kind", SpanKind.INTERNAL)
span = client_instance.add_agent_span(
input=input_, name=name, agent_type=agent_type, created_at=created_at
input=input_, name=name, agent_type=agent_type, span_kind=span_kind, created_at=created_at
)
else:
span = client_instance.add_workflow_span(input=input_, name=name, created_at=created_at)
Expand Down
Loading