From 0d986e025a842a26deabab74211ef030f636c8c5 Mon Sep 17 00:00:00 2001
From: T3782834 <T3782834@timbrasil.com.br>
Date: Wed, 28 Oct 2026 13:05:53 -0300
Subject: [PATCH] adding DB Error in Observability

---
 .idea/workspace.xml                           | 58 ++++++++++---------
 OBSERVABILITY_FIX_NOTES.md                    | 17 ++++++
 .../src/agent_framework/mcp/tool_router.py    |  4 +-
 .../memory/long_term_memory.py                |  6 +-
 .../observability/guardrail_events.py         |  2 +-
 .../observability/judge_events.py             |  2 +-
 .../observability/severity_policy.py          | 30 ++++++++++
 .../observability/telemetry.py                | 31 +++++++---
 tests/test_observability_severity_policy.py   | 23 ++++++++
 9 files changed, 134 insertions(+), 39 deletions(-)
 create mode 100644 OBSERVABILITY_FIX_NOTES.md
 create mode 100644 libs/agent_framework/src/agent_framework/observability/severity_policy.py
 create mode 100644 tests/test_observability_severity_policy.py

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 @@
     <option name="hideEmptyMiddlePackages" value="true" />
     <option name="showLibraryContents" value="true" />
   </component>
-  <component name="PropertiesComponent"><![CDATA[{
-  "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"
+  <component name="PropertiesComponent">{
+  &quot;keyToString&quot;: {
+    &quot;ASKED_SHARE_PROJECT_CONFIGURATION_FILES&quot;: &quot;true&quot;,
+    &quot;ModuleVcsDetector.initialDetectionPerformed&quot;: &quot;true&quot;,
+    &quot;RunOnceActivity.ShowReadmeOnStart&quot;: &quot;true&quot;,
+    &quot;RunOnceActivity.TerminalTabsStorage.copyFrom.TerminalArrangementManager&quot;: &quot;true&quot;,
+    &quot;RunOnceActivity.TerminalTabsStorage.copyFrom.TerminalArrangementManager.252&quot;: &quot;true&quot;,
+    &quot;RunOnceActivity.git.unshallow&quot;: &quot;true&quot;,
+    &quot;RunOnceActivity.typescript.service.memoryLimit.init&quot;: &quot;true&quot;,
+    &quot;SHARE_PROJECT_CONFIGURATION_FILES&quot;: &quot;true&quot;,
+    &quot;Shell Script.agent_frontend.executor&quot;: &quot;Run&quot;,
+    &quot;Shell Script.agent_template_backend.executor&quot;: &quot;Run&quot;,
+    &quot;Shell Script.mcp_gateway.executor&quot;: &quot;Run&quot;,
+    &quot;Shell Script.mcp_server.executor&quot;: &quot;Run&quot;,
+    &quot;git-widget-placeholder&quot;: &quot;main&quot;,
+    &quot;kotlin-language-version-configured&quot;: &quot;true&quot;,
+    &quot;last_opened_file_path&quot;: &quot;D:/Dropbox/ORACLE/TIM/FY27/Commits/agent_platform_oci&quot;,
+    &quot;node.js.detected.package.eslint&quot;: &quot;true&quot;,
+    &quot;node.js.detected.package.tslint&quot;: &quot;true&quot;,
+    &quot;node.js.selected.package.eslint&quot;: &quot;(autodetect)&quot;,
+    &quot;node.js.selected.package.tslint&quot;: &quot;(autodetect)&quot;,
+    &quot;nodejs_package_manager_path&quot;: &quot;npm&quot;,
+    &quot;project.structure.last.edited&quot;: &quot;Project&quot;,
+    &quot;project.structure.proportion&quot;: &quot;0.15&quot;,
+    &quot;project.structure.side.proportion&quot;: &quot;0.2&quot;,
+    &quot;settings.editor.selected.configurable&quot;: &quot;advanced.settings&quot;,
+    &quot;vue.rearranger.settings.migration&quot;: &quot;true&quot;
   }
-}]]></component>
+}</component>
   <component name="RecentsManager">
     <key name="CopyFile.RECENT_KEYS">
       <recent name="C:\Asus_Projects\kbdb\agent_framework_oci\templates\agent_template_backend" />
@@ -178,6 +178,8 @@
       <workItem from="1785452072692" duration="784000" />
       <workItem from="1785453511270" duration="1006000" />
       <workItem from="1792665278602" duration="734000" />
+      <workItem from="1793193595355" duration="436000" />
+      <workItem from="1793194079956" duration="4230000" />
     </task>
     <task id="LOCAL-00001" summary="Ajustes na documentação e remanejamento dos folders">
       <option name="closed" value="true" />
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'