diff --git a/.idea/workspace.xml b/.idea/workspace.xml
index baef0cf3..aeb81c2f 100644
--- a/.idea/workspace.xml
+++ b/.idea/workspace.xml
@@ -47,35 +47,35 @@
- {
+ "keyToString": {
+ "ASKED_SHARE_PROJECT_CONFIGURATION_FILES": "true",
+ "ModuleVcsDetector.initialDetectionPerformed": "true",
+ "RunOnceActivity.ShowReadmeOnStart": "true",
+ "RunOnceActivity.TerminalTabsStorage.copyFrom.TerminalArrangementManager": "true",
+ "RunOnceActivity.TerminalTabsStorage.copyFrom.TerminalArrangementManager.252": "true",
+ "RunOnceActivity.git.unshallow": "true",
+ "RunOnceActivity.typescript.service.memoryLimit.init": "true",
+ "SHARE_PROJECT_CONFIGURATION_FILES": "true",
+ "Shell Script.agent_frontend.executor": "Run",
+ "Shell Script.agent_template_backend.executor": "Run",
+ "Shell Script.mcp_gateway.executor": "Run",
+ "Shell Script.mcp_server.executor": "Run",
+ "git-widget-placeholder": "main",
+ "kotlin-language-version-configured": "true",
+ "last_opened_file_path": "D:/Dropbox/ORACLE/TIM/FY27/Commits/agent_platform_oci",
+ "node.js.detected.package.eslint": "true",
+ "node.js.detected.package.tslint": "true",
+ "node.js.selected.package.eslint": "(autodetect)",
+ "node.js.selected.package.tslint": "(autodetect)",
+ "nodejs_package_manager_path": "npm",
+ "project.structure.last.edited": "Project",
+ "project.structure.proportion": "0.15",
+ "project.structure.side.proportion": "0.2",
+ "settings.editor.selected.configurable": "advanced.settings",
+ "vue.rearranger.settings.migration": "true"
}
-}]]>
+}
@@ -178,6 +178,8 @@
+
+
diff --git a/OBSERVABILITY_FIX_NOTES.md b/OBSERVABILITY_FIX_NOTES.md
new file mode 100644
index 00000000..c7b2767f
--- /dev/null
+++ b/OBSERVABILITY_FIX_NOTES.md
@@ -0,0 +1,17 @@
+# Observability severity patch
+
+- Central `severity_policy.py` classifies explicit MCP failures, guardrail blocks, judge rejections, RAG failures and memory/cache errors.
+- `Telemetry.event()` accepts `level` and `status_message` and preserves WARNING/ERROR events as individual observations in compact mode.
+- `Telemetry.span()` handle supports `set_status(level, message)`; failed MCP results mark their span ERROR without raising exceptions.
+- Long-term-memory read/write failures emit telemetry while preserving existing fallback behavior.
+- Guardrail blocked and negative judge evaluation emit WARNING when these helper methods are invoked.
+- Normal cache misses and empty RAG results are not automatically failures.
+
+## Scope / limitations
+- Existing call sites must invoke `GuardrailTelemetry.blocked` for guardrail blocking events; this patch does not rewrite every pipeline call site.
+- Redis cache currently has no injected Telemetry instance, so its existing logger errors are not yet Langfuse observations. A separate dependency-injection change is required to cover that path safely.
+- RAG database connection errors that are caught by external KBDB integrations without emitting a telemetry event cannot be classified automatically.
+- No live Langfuse ingestion test was possible in this environment. Verify both compact and verbose modes against your server.
+
+## Validation
+`PYTHONPATH=libs/agent_framework/src python -m pytest -q tests/test_observability_severity_policy.py`
diff --git a/libs/agent_framework/src/agent_framework/mcp/tool_router.py b/libs/agent_framework/src/agent_framework/mcp/tool_router.py
index 48c5d222..66618e95 100644
--- a/libs/agent_framework/src/agent_framework/mcp/tool_router.py
+++ b/libs/agent_framework/src/agent_framework/mcp/tool_router.py
@@ -236,8 +236,10 @@ class MCPToolRouter:
mcp_server=("mcp_gateway" if self.gateway_enabled else server.name),
input=mapped_arguments,
tags=["mcp", "tool", "mcp_gateway" if self.gateway_enabled else "mcp_server"],
- ):
+ ) as span:
result = await _execute()
+ if not result.ok:
+ span.set_status("ERROR", result.error or "MCP tool failed")
await self.telemetry.event(
"mcp.tool_call.completed",
{
diff --git a/libs/agent_framework/src/agent_framework/memory/long_term_memory.py b/libs/agent_framework/src/agent_framework/memory/long_term_memory.py
index 82bdf19e..56b51c86 100644
--- a/libs/agent_framework/src/agent_framework/memory/long_term_memory.py
+++ b/libs/agent_framework/src/agent_framework/memory/long_term_memory.py
@@ -33,7 +33,9 @@ class LongTermMemoryManager:
return []
try:
return await self.store.search(tenant_id=tenant, agent_id=agent, subject_key=subject, limit=int(getattr(self.settings, 'LONG_TERM_MEMORY_MAX_CONTEXT_ITEMS', 20)))
- except Exception:
+ except Exception as exc:
+ if self.telemetry:
+ await self.telemetry.event('memory.long_term.load.failed', {'error': str(exc), 'fallback': True}, kind='memory')
logger.exception('Falha não crítica ao carregar LTM')
return []
@@ -49,6 +51,8 @@ class LongTermMemoryManager:
saved = await self.store.upsert_many(tenant_id=tenant, agent_id=agent, subject_key=subject, items=candidates, source_session_id=str(state.get('conversation_key') or state.get('session_id') or ''), source_message_id=str((state.get('context') or {}).get('message_id') or ''))
return {'saved': len(saved), 'items': [item.to_dict() for item in saved]}
except Exception as exc:
+ if self.telemetry:
+ await self.telemetry.event('memory.long_term.persist.failed', {'error': str(exc)}, kind='memory')
logger.exception('Falha não crítica ao persistir LTM')
return {'saved': 0, 'error': str(exc)}
diff --git a/libs/agent_framework/src/agent_framework/observability/guardrail_events.py b/libs/agent_framework/src/agent_framework/observability/guardrail_events.py
index 8ad75dac..73277dfa 100644
--- a/libs/agent_framework/src/agent_framework/observability/guardrail_events.py
+++ b/libs/agent_framework/src/agent_framework/observability/guardrail_events.py
@@ -10,4 +10,4 @@ class GuardrailTelemetry:
async def blocked(self, stage: str, decision: Any):
payload = decision.model_dump() if hasattr(decision, "model_dump") else dict(decision or {})
payload.update({"stage": stage})
- await self.telemetry.event(f"guardrail.{payload.get('code', 'unknown')}.blocked", payload, kind="guardrail")
+ await self.telemetry.event(f"guardrail.{payload.get('code', 'unknown')}.blocked", payload, kind="guardrail", level="WARNING")
diff --git a/libs/agent_framework/src/agent_framework/observability/judge_events.py b/libs/agent_framework/src/agent_framework/observability/judge_events.py
index 25f43d11..2753afc2 100644
--- a/libs/agent_framework/src/agent_framework/observability/judge_events.py
+++ b/libs/agent_framework/src/agent_framework/observability/judge_events.py
@@ -6,4 +6,4 @@ class JudgeTelemetry:
async def evaluated(self, result: Any, latency_ms: int | None = None):
payload = result.model_dump() if hasattr(result, "model_dump") else dict(result or {})
payload.update({"latency_ms": latency_ms})
- await self.telemetry.event(f"judge.{payload.get('name', 'unknown')}.evaluated", payload, kind="judge")
+ await self.telemetry.event(f"judge.{payload.get('name', 'unknown')}.evaluated", payload, kind="judge", level="WARNING" if any(payload.get(k) is False for k in ("approved", "passed", "accepted")) else None)
diff --git a/libs/agent_framework/src/agent_framework/observability/severity_policy.py b/libs/agent_framework/src/agent_framework/observability/severity_policy.py
new file mode 100644
index 00000000..6368c0a3
--- /dev/null
+++ b/libs/agent_framework/src/agent_framework/observability/severity_policy.py
@@ -0,0 +1,30 @@
+"""Conservative operational severity policy for Langfuse observations.
+
+Only failures explicitly evidenced by event names/payloads are elevated.
+Normal cache misses, empty memory reads and normal retrievals are not warnings.
+"""
+from dataclasses import dataclass
+from typing import Any
+
+@dataclass(frozen=True)
+class Severity:
+ level: str
+ message: str | None = None
+
+def classify_event(name: str, payload: dict[str, Any], *, explicit_level: str | None = None, explicit_message: str | None = None) -> Severity:
+ name = str(name or "").lower()
+ payload = payload or {}
+ message = explicit_message or str(payload.get("error") or payload.get("reason") or "") or None
+ if explicit_level is not None:
+ return Severity(explicit_level, message)
+ if name.startswith("guardrail.") and name.endswith(".blocked"):
+ return Severity("WARNING", message or "Guardrail blocked the request or response")
+ if name.startswith("judge.") and (payload.get("approved") is False or payload.get("passed") is False or payload.get("accepted") is False):
+ return Severity("WARNING", message or "Judge rejected the response")
+ if name.startswith("mcp.") and payload.get("ok") is False:
+ return Severity("ERROR", message or "MCP tool failed")
+ if name.startswith("rag.") and (name.endswith(".failed") or payload.get("error")):
+ return Severity("WARNING" if name.endswith(("rewrite.failed", "compress.failed")) else "ERROR", message or name)
+ if name.startswith(("memory.", "cache.", "checkpoint.")) and (name.endswith(".failed") or payload.get("error")):
+ return Severity("WARNING" if payload.get("fallback") else "ERROR", message or name)
+ return Severity("DEFAULT")
diff --git a/libs/agent_framework/src/agent_framework/observability/telemetry.py b/libs/agent_framework/src/agent_framework/observability/telemetry.py
index 209f6d30..37f4cf32 100644
--- a/libs/agent_framework/src/agent_framework/observability/telemetry.py
+++ b/libs/agent_framework/src/agent_framework/observability/telemetry.py
@@ -206,6 +206,15 @@ class _SpanHandle:
self.output: Any = None
self.has_output = False
self.metadata: dict[str, Any] = {}
+ self.level: str | None = None
+ self.status_message: str | None = None
+
+ def set_status(self, level: str, message: str | None = None) -> None:
+ """Mark an operational failure without raising an exception."""
+ if level not in {"DEFAULT", "WARNING", "ERROR", "DEBUG"}:
+ raise ValueError(f"Invalid Langfuse level: {level}")
+ self.level = level
+ self.status_message = message
def set_observation(self, observation: Any | None) -> None:
self.observation = observation
@@ -442,22 +451,26 @@ class Telemetry:
await self.event_bus.publish(f"{name}.started", attrs, kind="span")
yield handle
duration_ms = int((time.time() - start) * 1000)
- status = {"status": "ok", "duration_ms": duration_ms}
+ status = {"status": "error" if handle.level == "ERROR" else "degraded" if handle.level == "WARNING" else "ok", "duration_ms": duration_ms}
out = handle.output if handle.has_output else status
metadata = {**observation_metadata, **status, **handle.metadata}
if span_events is not None:
metadata["aggregated_event_count"] = len(span_events)
metadata["aggregated_events"] = span_events
- self._update_observation(observation, input=attrs.get("input"), output=out, metadata=metadata)
+ self._update_observation(observation, input=attrs.get("input"), output=out, metadata=metadata, level=handle.level, status_message=handle.status_message)
if is_root_span:
self._set_trace_io(observation, input=attrs.get("input"), output=out)
legacy_io_update = {
"input": attrs.get("input"),
"output": out,
"metadata": metadata,
+ "level": handle.level,
+ "status_message": handle.status_message,
}
if otel_span is not None:
otel_span.set_attribute("duration_ms", duration_ms)
+ if handle.level in ("WARNING", "ERROR"):
+ otel_span.set_attribute("operational.level", handle.level)
completed_payload = {**attrs, **status}
if handle.has_output:
completed_payload["output"] = out
@@ -510,7 +523,11 @@ class Telemetry:
try: otel_cm.__exit__(None, None, None)
except Exception: logger.debug("Falha ao fechar span OTEL", exc_info=True)
- async def event(self, name: str, payload: dict[str, Any] | None = None, *, kind: str = "event"):
+ async def event(self, name: str, payload: dict[str, Any] | None = None, *, kind: str = "event", level: str | None = None, status_message: str | None = None):
+ from .severity_policy import classify_event
+ if level is not None and level not in {"DEBUG", "DEFAULT", "WARNING", "ERROR"}:
+ raise ValueError(f"Invalid Langfuse level: {level}")
+ severity = classify_event(name, payload or {}, explicit_level=level, explicit_message=status_message)
name, payload, mapping_metadata = self.code_mapper.normalize_payload(name, payload, None)
if mapping_metadata:
payload = {**payload, **mapping_metadata}
@@ -524,14 +541,14 @@ class Telemetry:
"kind": kind,
"payload": payload,
})
- if not _is_compact_visible_event(name) or not self.is_enabled():
+ if (not _is_compact_visible_event(name) and severity.level not in ("WARNING", "ERROR")) or not self.is_enabled():
return
try:
metadata = {**payload, "event_kind": kind}
- cm = self._start_observation(name=name, as_type="span", input=payload, metadata=metadata)
+ cm = self._start_observation(name=name, as_type="span", input=payload, metadata=metadata, level=severity.level, status_message=severity.message)
if cm is not None:
with cm as obs:
- self._update_observation(obs, input=payload, output={"status": "ok"}, metadata=metadata)
+ self._update_observation(obs, input=payload, output={"status": "error" if severity.level == "ERROR" else "warning" if severity.level == "WARNING" else "ok"}, metadata=metadata, level=severity.level, status_message=severity.message)
except Exception:
logger.exception("Falha ao enviar event compacto via observation")
return
@@ -545,7 +562,7 @@ class Telemetry:
metadata = {**payload, "event_kind": kind}
if self.is_compact_mode():
metadata["_ignore_current_parent"] = True
- cm = self._start_observation(name=name, as_type=_langfuse_type(kind), metadata=metadata)
+ cm = self._start_observation(name=name, as_type=_langfuse_type(kind), metadata=metadata, level=severity.level, status_message=severity.message)
if cm is not None:
with cm: pass
except Exception:
diff --git a/tests/test_observability_severity_policy.py b/tests/test_observability_severity_policy.py
new file mode 100644
index 00000000..29d79144
--- /dev/null
+++ b/tests/test_observability_severity_policy.py
@@ -0,0 +1,23 @@
+import sys
+from pathlib import Path
+sys.path.insert(0, str(Path(__file__).resolve().parents[1] / 'libs' / 'agent_framework' / 'src'))
+from agent_framework.observability.severity_policy import classify_event
+
+def test_mcp_failure():
+ s = classify_event('mcp.tool_call.completed', {'ok': False, 'error': 'All connection attempts failed'})
+ assert s.level == 'ERROR'
+
+def test_guardrail_block():
+ assert classify_event('guardrail.DLEX_OUT.blocked', {'allowed': False}).level == 'WARNING'
+
+def test_judge_reject():
+ assert classify_event('judge.quality.evaluated', {'passed': False}).level == 'WARNING'
+
+def test_rag_fallback():
+ assert classify_event('rag.rewrite.failed', {}).level == 'WARNING'
+
+def test_normal_cache_miss():
+ assert classify_event('cache.get', {'hit': False}).level == 'DEFAULT'
+
+def test_memory_fallback():
+ assert classify_event('memory.long_term.load.failed', {'error': 'timeout', 'fallback': True}).level == 'WARNING'