fix(tracing): narrow Agentex platform ownership

Ignore generic wrappers and database drivers while reserving platform attribution for managed tracing export and Redis stream persistence paths.

Co-authored-by: Cursor <cursoragent@cursor.com>

Author: jshaikScale

src/agentex/lib/core/tracing/__init__.py

@@ -4,6 +4,8 @@
 from agentex.lib.core.tracing.span_error import (
     ERROR_CLASSIFIER_VERSION,
     AGENTEX_ERROR_CLASSIFIER_CONFIG,
+    AGENTEX_IGNORED_MODULE_PREFIXES,
+    AGENTEX_PLATFORM_MODULE_PREFIXES,
     ErrorCategory,
     PlatformError,
     ApplicationError,
@@ -34,6 +36,8 @@
     "ErrorClassifierConfig",
     "TracebackOwnershipPolicy",
     "ERROR_CLASSIFIER_VERSION",
+    "AGENTEX_IGNORED_MODULE_PREFIXES",
+    "AGENTEX_PLATFORM_MODULE_PREFIXES",
     "AGENTEX_ERROR_CLASSIFIER_CONFIG",
     "AsyncSpanQueue",
     "get_default_span_queue",

src/agentex/lib/core/tracing/span_error.py

@@ -1,6 +1,5 @@
 from __future__ import annotations
 
-import os
 from typing import Any, cast
 
 from scale_gp_beta.lib.tracing import (
@@ -27,11 +26,28 @@
 # SGP and agentex-native span stores.
 SPAN_ERROR_KEY = "__error__"
 
-_AGENTEX_PACKAGE_ROOT = os.path.normcase(os.path.normpath(os.path.join(os.path.dirname(__file__), "..", "..", "..")))
+AGENTEX_PLATFORM_MODULE_PREFIXES = (
+    # Trace export is Agentex-owned telemetry delivery, not agent execution.
+    "agentex.lib.core.tracing.processors",
+    "agentex.lib.core.tracing.span_queue",
+    # This repository is the managed Redis stream transport/persistence layer.
+    "agentex.lib.core.adapters.streams.adapter_redis",
+)
+AGENTEX_IGNORED_MODULE_PREFIXES = (
+    # Generic Agentex orchestration/wrappers provide context, not ownership.
+    "agentex",
+    # Database/client packages never decide ownership by exception identity.
+    "sqlalchemy",
+    "pyodbc",
+    "psycopg",
+    "psycopg2",
+    "asyncpg",
+    "redis",
+)
 AGENTEX_ERROR_CLASSIFIER_CONFIG = ErrorClassifierConfig(
     policy=TracebackOwnershipPolicy(
-        platform_module_prefixes=("agentex",),
-        platform_file_roots=(_AGENTEX_PACKAGE_ROOT,),
+        platform_module_prefixes=AGENTEX_PLATFORM_MODULE_PREFIXES,
+        ignored_module_prefixes=AGENTEX_IGNORED_MODULE_PREFIXES,
         infer_application_from_unowned_absolute_paths=True,
     )
 )

tests/lib/core/tracing/test_span_error.py

@@ -1,6 +1,5 @@
 from __future__ import annotations
 
-import json
 import uuid
 from typing import Any
 from datetime import UTC, datetime
@@ -44,6 +43,23 @@ def _synthetic_function(module_name: str, filename: str, body: str, **values: An
     return namespace["run"]
 
 
+def _record_synthetic(function: Any, *, innermost_only: bool = False) -> dict[str, Any]:
+    span = _make_span()
+    try:
+        function()
+    except Exception as exc:
+        if innermost_only:
+            traceback = exc.__traceback__
+            assert traceback is not None
+            while traceback.tb_next is not None:
+                traceback = traceback.tb_next
+            exc = exc.with_traceback(traceback)
+        set_span_error(span, exc)
+    error = get_span_error(span)
+    assert error is not None
+    return error
+
+
 # ---------------------------------------------------------------------------
 # Helpers: set_span_error / get_span_error
 # ---------------------------------------------------------------------------
@@ -102,53 +118,102 @@ class ImplicitlyCategorizedError(RuntimeError):
         set_span_error(span, ImplicitlyCategorizedError("boom"))
         assert get_span_error(span)["category"] == "unknown"  # type: ignore[index]
 
-    def test_agentex_internal_origin_is_platform(self):
-        platform = _synthetic_function(
-            "agentex.lib.synthetic_runtime",
-            "/synthetic/agentex/runtime.py",
+    def test_generic_agentex_wrapper_around_user_failure_is_application(self):
+        application = _synthetic_function(
+            "customer_agent.tool",
+            "/synthetic/application/tool.py",
             "raise RuntimeError('boom')",
         )
-        span = _make_span()
-        try:
-            platform()
-        except Exception as exc:
-            set_span_error(span, exc)
+        wrapper = _synthetic_function(
+            "agentex.lib.core.temporal.workflows.workflow",
+            "/synthetic/agentex/workflow.py",
+            "target()",
+            target=application,
+        )
 
-        error = get_span_error(span)
-        assert error is not None
-        assert error["category"] == "platform"
+        error = _record_synthetic(wrapper)
+
+        assert error["category"] == "application"
         assert error["category_source"] == "stack_trace"
-        assert error["category_reason"] == "stack_rule:platform_module"
+        assert error["category_reason"] == "stack_rule:unowned_absolute_source"
 
-    def test_stdlib_dependency_under_application_is_application(self):
+    def test_application_database_failure_is_application(self):
+        driver = _synthetic_function(
+            "sqlalchemy.engine",
+            "/venv/site-packages/sqlalchemy/engine.py",
+            "raise RuntimeError('db failed')",
+        )
         application = _synthetic_function(
             "customer_agent.main",
             "/synthetic/application/main.py",
-            "parse('{')",
-            parse=json.loads,
+            "query()",
+            query=driver,
         )
-        span = _make_span()
-        try:
-            application()
-        except Exception as exc:
-            set_span_error(span, exc)
 
-        assert get_span_error(span)["category"] == "application"  # type: ignore[index]
+        assert _record_synthetic(application)["category"] == "application"
+
+    def test_managed_stream_database_failure_is_platform(self):
+        driver = _synthetic_function(
+            "redis.asyncio.client",
+            "/venv/site-packages/redis/client.py",
+            "raise RuntimeError('db failed')",
+        )
+        managed_store = _synthetic_function(
+            "agentex.lib.core.adapters.streams.adapter_redis",
+            "/synthetic/agentex/adapter_redis.py",
+            "query()",
+            query=driver,
+        )
 
-    def test_stdlib_dependency_under_agentex_is_platform(self):
-        platform = _synthetic_function(
-            "agentex.lib.synthetic_runtime",
-            "/synthetic/agentex/runtime.py",
-            "parse('{')",
-            parse=json.loads,
+        assert _record_synthetic(managed_store)["category"] == "platform"
+
+    def test_application_validation_inside_platform_operation_is_application(self):
+        validation = _synthetic_function(
+            "customer_agent.query",
+            "/synthetic/application/query.py",
+            "raise ValueError('invalid configuration')",
+        )
+        managed_store = _synthetic_function(
+            "agentex.lib.core.adapters.streams.adapter_redis",
+            "/synthetic/agentex/adapter_redis.py",
+            "validate()",
+            validate=validation,
+        )
+
+        assert _record_synthetic(managed_store)["category"] == "application"
+
+    def test_driver_only_database_failure_is_unknown(self):
+        driver = _synthetic_function(
+            "pyodbc",
+            "/venv/site-packages/pyodbc.py",
+            "raise RuntimeError('db failed')",
+        )
+
+        error = _record_synthetic(driver, innermost_only=True)
+
+        assert error["category"] == "unknown"
+        assert error["category_reason"] == "stack_no_owned_frame"
+
+    def test_nested_wrapper_dependency_propagation_uses_application(self):
+        driver = _synthetic_function(
+            "sqlalchemy.engine",
+            "/venv/site-packages/sqlalchemy/engine.py",
+            "raise RuntimeError('db failed')",
+        )
+        application = _synthetic_function(
+            "customer_agent.repository",
+            "/synthetic/application/repository.py",
+            "query()",
+            query=driver,
+        )
+        wrapper = _synthetic_function(
+            "agentex.lib.core.temporal.activities.activity_helpers",
+            "/synthetic/agentex/activity_helpers.py",
+            "target()",
+            target=application,
         )
-        span = _make_span()
-        try:
-            platform()
-        except Exception as exc:
-            set_span_error(span, exc)
 
-        assert get_span_error(span)["category"] == "platform"  # type: ignore[index]
+        assert _record_synthetic(wrapper)["category"] == "application"
 
     def test_set_preserves_existing_dict_keys(self):
         span = _make_span(data={"__span_type__": "LLM"})