Skip to content

fix: correct Observe telemetry attribution #461

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

Merged
Paulo Lacerda (placerda) merged 1 commit into from
Aug 27, 2026
Merged

fix: correct Observe telemetry attribution #461

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
7 changes: 7 additions & 0 deletions CHANGELOG.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -70,6 +70,13 @@ This format follows [Keep a Changelog](https://keepachangelog.com/) and adheres
WAF checklist row are gone.

### Fixed
- **Observe now attributes Foundry agents and presents usable telemetry tables.**
Hosted and prompt agents are classified from their emitted provider metadata,
project attribution recognizes the Foundry project dimension, and Runs returns
one correlated execution instead of separate rows for each internal operation.
Overview counts only agent invocations, token usage is split into sortable
columns, and internal connector diagnostics and the redundant Telemetry
coverage tab no longer appear in the user-facing dashboard.
- **Foundry project links now open the configured project instead of the tenant
landing page.** In project-observability-only mode there is no cloud evaluation
report from which to recover a portal URL, so Cockpit previously fell back to
Expand Down
65 changes: 52 additions & 13 deletions src/agentops/agent/observe/queries.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -39,13 +39,15 @@
_APPGENAI_TABLE = "AppGenAIContent"
_PROJECT_RESOURCE_ID = (
'tostring(coalesce(Properties["gen_ai.project.id"], '
'Properties["gen_ai.azure_ai_project.id"]))'
'Properties["gen_ai.azure_ai_project.id"], '
'Properties["microsoft.foundry.project.id"]))'
)

TOKEN_CLASS_ALIASES: dict[str, tuple[str, ...]] = {
"cache_read": (
"gen_ai.usage.cache_read.input_tokens",
"gen_ai.usage.cache_read_input_tokens",
"gen_ai.usage.cached_tokens",
),
"cache_write": (
"gen_ai.usage.cache_write.input_tokens",
Expand Down Expand Up @@ -169,6 +171,7 @@ def _agent_extend_clauses() -> list[str]:
'| extend agent_name = tostring(Properties["gen_ai.agent.name"])',
'| extend provider_name = tostring(Properties["gen_ai.provider.name"])',
'| extend system = tostring(Properties["gen_ai.system"])',
'| extend operation_name = tostring(Properties["gen_ai.operation.name"])',
'| extend model = tostring(coalesce(Properties["gen_ai.request.model"], '
'Properties["gen_ai.response.model"]))',
'| extend input_tokens = toint(Properties["gen_ai.usage.input_tokens"])',
Expand Down Expand Up @@ -259,16 +262,32 @@ def _token_class_extend_clauses() -> list[str]:
def build_overview_query(
filters: ObserveFilterState, *, scope_source: TelemetrySource | None = None
) -> str:
"""Bounded aggregate invocation/failure/latency query for the overview view."""
"""Aggregate agent invocations without counting internal HTTP/model spans."""
lines = [
_TELEMETRY_TABLES,
_time_window_clause(filters),
*_dimension_filters(filters, scope_source),
"| where isnotempty(Name)",
"| summarize invocations = count(), "
"failures = countif(Success == false), "
"avg_latency_ms = avg(DurationMs), "
"p95_latency_ms = percentile(DurationMs, 95)",
*_agent_extend_clauses(),
'| extend is_request_invocation = Type == "AppRequests" and '
'operation_name == "invoke_agent", '
'is_dependency_invocation = Type == "AppDependencies" and '
'operation_name == "invoke_agent"',
"| summarize request_invocations = countif(is_request_invocation), "
"dependency_invocations = countif(is_dependency_invocation), "
"request_failures = countif(is_request_invocation and Success == false), "
"dependency_failures = countif(is_dependency_invocation and Success == false), "
"request_avg_latency_ms = avgif(DurationMs, is_request_invocation), "
"dependency_avg_latency_ms = avgif(DurationMs, is_dependency_invocation), "
"request_p95_latency_ms = percentileif(DurationMs, 95, is_request_invocation), "
"dependency_p95_latency_ms = percentileif(DurationMs, 95, is_dependency_invocation)",
"| extend invocations = iff(request_invocations > 0, "
"request_invocations, dependency_invocations), "
"failures = iff(request_invocations > 0, request_failures, dependency_failures), "
"avg_latency_ms = iff(request_invocations > 0, "
"request_avg_latency_ms, dependency_avg_latency_ms), "
"p95_latency_ms = iff(request_invocations > 0, "
"request_p95_latency_ms, dependency_p95_latency_ms)",
"| project invocations, failures, avg_latency_ms, p95_latency_ms",
]
return "\n".join(lines)

Expand All @@ -282,9 +301,16 @@ def build_agents_query(
_time_window_clause(filters),
*_dimension_filters(filters, scope_source),
*_agent_extend_clauses(),
"| summarize invocations = count(), "
"failures = countif(Success == false), "
"p95_latency_ms = percentile(DurationMs, 95), "
'| extend is_request_invocation = Type == "AppRequests" and '
'operation_name == "invoke_agent", '
'is_dependency_invocation = Type == "AppDependencies" and '
'operation_name == "invoke_agent"',
"| summarize request_invocations = countif(is_request_invocation), "
"dependency_invocations = countif(is_dependency_invocation), "
"request_failures = countif(is_request_invocation and Success == false), "
"dependency_failures = countif(is_dependency_invocation and Success == false), "
"request_p95_latency_ms = percentileif(DurationMs, 95, is_request_invocation), "
"dependency_p95_latency_ms = percentileif(DurationMs, 95, is_dependency_invocation), "
"input_tokens = sum(input_tokens), "
"output_tokens = sum(output_tokens), "
"last_seen = max(TimeGenerated), "
Expand All @@ -294,6 +320,15 @@ def build_agents_query(
"system = take_anyif(system, isnotempty(system)), "
"model = take_anyif(model, isnotempty(model)) "
"by project_resource_id, agent_key",
"| extend invocations = iff(request_invocations > 0, "
"request_invocations, dependency_invocations), "
"failures = iff(request_invocations > 0, request_failures, dependency_failures), "
"p95_latency_ms = iff(request_invocations > 0, "
"request_p95_latency_ms, dependency_p95_latency_ms)",
"| project-away request_invocations, dependency_invocations, "
"request_failures, dependency_failures, request_p95_latency_ms, "
"dependency_p95_latency_ms",
"| where invocations > 0",
]
return _bounded_aggregate(aggregate_lines, order_by="invocations")

Expand All @@ -309,6 +344,7 @@ def build_models_query(
*_agent_extend_clauses(),
'| extend deployment = tostring(Properties["gen_ai.request.deployment"])',
*_token_class_extend_clauses(),
"| where isnotempty(model) or isnotempty(deployment)",
]
summary_lines = [
"| summarize requests = count(), "
Expand Down Expand Up @@ -488,9 +524,12 @@ def build_runs_query(
"cache_write_token_reports = countif(isnotnull(cache_write_tokens)), "
"reasoning_token_reports = countif(isnotnull(reasoning_tokens)), "
"credit_reports = countif(isnotnull(credits)), "
"credit_event_reports = countif(isnotnull(credit_event)) "
"by project_resource_id, agent_key, agent_id, agent_name, provider_name, system, "
"run_key, run_key_kind, operation_name",
"credit_event_reports = countif(isnotnull(credit_event)), "
"agent_id = take_anyif(agent_id, isnotempty(agent_id)), "
"agent_name = take_anyif(agent_name, isnotempty(agent_name)), "
"provider_name = take_anyif(provider_name, isnotempty(provider_name)), "
"system = take_anyif(system, isnotempty(system)) "
"by project_resource_id, agent_key, run_key, run_key_kind",
"| extend input_tokens = iff(input_token_reports == 0, long(null), input_tokens), "
"output_tokens = iff(output_token_reports == 0, long(null), output_tokens), "
"cache_read_tokens = iff(cache_read_token_reports == 0, long(null), "
Expand Down
52 changes: 45 additions & 7 deletions src/agentops/agent/observe/service.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -227,6 +227,10 @@ class _CachedView:

_HOSTED_AGENT_KINDS = frozenset({"hosted", "container", "foundry_hosted"})
_PROMPT_AGENT_KINDS = frozenset({"prompt", "foundry_prompt"})
_HOSTED_AGENT_PROVIDERS = frozenset(
{"azure.ai.foundry", "microsoft.agent_framework", "microsoft agent framework"}
)
_PROMPT_AGENT_PROVIDERS = frozenset({"microsoft.foundry", "microsoft foundry"})


def _normalized_runtime_value(value: Any) -> str | None:
Expand Down Expand Up @@ -345,6 +349,14 @@ def classify_runtime(
if _is_copilot_studio_provider(provider_name, system):
return "copilot_studio"

provider = _normalized_runtime_value(provider_name)
if provider is not None:
readable_provider = provider.replace("_", " ")
if provider in _HOSTED_AGENT_PROVIDERS or readable_provider in _HOSTED_AGENT_PROVIDERS:
return "foundry_hosted"
if provider in _PROMPT_AGENT_PROVIDERS or readable_provider in _PROMPT_AGENT_PROVIDERS:
return "foundry_prompt"

if agent_id:
return _inventory_agent_kind(inventory, agent_id=agent_id) or "unknown"

Expand Down Expand Up @@ -1808,7 +1820,9 @@ async def query_view(
cache_status="hit",
)

inventory = await self.get_inventory(scope, refresh=refresh)
# Refresh telemetry without repeating the slower control-plane discovery.
# Inventory has its own cache and explicit discover endpoint for forced refreshes.
inventory = await self.get_inventory(scope)
available_sources = [
source
for source in inventory.telemetry_sources
Expand Down Expand Up @@ -1927,7 +1941,7 @@ async def query_attribution(
scope=scope,
).id

inventory = await self.get_inventory(scope, refresh=request.refresh)
inventory = await self.get_inventory(scope)
available_sources = [
source
for source in inventory.telemetry_sources
Expand Down Expand Up @@ -2748,7 +2762,7 @@ async def _query_user_usage_attribution(
principal_group_ids=groups,
)

inventory = await self.get_inventory(scope, refresh=request.refresh)
inventory = await self.get_inventory(scope)
sources = [
source
for source in inventory.telemetry_sources
Expand Down Expand Up @@ -3121,7 +3135,7 @@ async def query_cost(
cache_status="hit",
)

inventory = await self.get_inventory(scope, refresh=refresh)
inventory = await self.get_inventory(scope)
available_sources = [
source
for source in inventory.telemetry_sources
Expand Down Expand Up @@ -3671,12 +3685,30 @@ def _normalize_view(
return runs, coverage

# "overview": aggregate totals only, never inferring a zero as failure.
totals = {"invocations": 0, "failures": 0}
totals: dict[str, int | float | None] = {
"invocations": 0,
"failures": 0,
"avg_latency_ms": None,
"p95_latency_ms": None,
}
weighted_latency = 0.0
latency_invocations = 0
source_p95_values: list[float] = []
for result in results:
rows = list(result.tables or [])
for row in rows:
totals["invocations"] += int(row.get("invocations") or 0)
totals["failures"] += int(row.get("failures") or 0)
invocations = int(row.get("invocations") or 0)
totals["invocations"] = int(totals["invocations"] or 0) + invocations
totals["failures"] = int(totals["failures"] or 0) + int(
row.get("failures") or 0
)
average = row.get("avg_latency_ms")
if average is not None and invocations > 0:
weighted_latency += float(average) * invocations
latency_invocations += invocations
p95 = row.get("p95_latency_ms")
if p95 is not None:
source_p95_values.append(float(p95))
coverage.append(
classify_query_coverage(
source_id=result.source_id,
Expand All @@ -3687,6 +3719,12 @@ def _normalize_view(
refreshed_at=refreshed_at,
)
)
if latency_invocations:
totals["avg_latency_ms"] = weighted_latency / latency_invocations
if source_p95_values:
# A percentile cannot be recomputed from per-source aggregates. The
# maximum is a conservative cross-source operational signal.
totals["p95_latency_ms"] = max(source_p95_values)
return totals, coverage

def _build_diagnostics(
Expand Down
Loading
Loading