Add ApplyGuardrailScope for security guardrail evaluation tracing

libraries/microsoft-agents-a365-observability-core/microsoft_agents_a365/observability/core/__init__.py

@@ -4,6 +4,7 @@
 # Microsoft Agent 365 Python SDK for OpenTelemetry tracing.
 
 from .agent_details import AgentDetails
+from .apply_guardrail_scope import ApplyGuardrailScope
 from .config import (
     configure,
     get_tracer,
@@ -19,6 +20,11 @@
     unregister_span_enricher,
 )
 from .exporters.spectra_exporter_options import SpectraExporterOptions
+from .guardrail_decision_type import GuardrailDecisionType
+from .guardrail_details import GuardrailDetails
+from .guardrail_finding import GuardrailFinding
+from .guardrail_risk_severity import GuardrailRiskSeverity
+from .guardrail_target_type import GuardrailTargetType
 from .inference_call_details import InferenceCallDetails
 from .models.service_endpoint import ServiceEndpoint
 from .inference_operation_type import InferenceOperationType
@@ -80,6 +86,7 @@
     # Base scope class
     "OpenTelemetryScope",
     # Specific scope classes
+    "ApplyGuardrailScope",
     "ExecuteToolScope",
     "InvokeAgentScope",
     "InferenceScope",
@@ -98,7 +105,12 @@
     "SpanDetails",
     "InferenceCallDetails",
     "ServiceEndpoint",
+    "GuardrailDetails",
+    "GuardrailFinding",
     # Enums
+    "GuardrailDecisionType",
+    "GuardrailRiskSeverity",
+    "GuardrailTargetType",
     "InferenceOperationType",
     "ToolType",
     # OTEL gen-ai message format types

libraries/microsoft-agents-a365-observability-core/microsoft_agents_a365/observability/core/apply_guardrail_scope.py

@@ -0,0 +1,224 @@
+# Copyright (c) Microsoft Corporation.
+# Licensed under the MIT License.
+
+from opentelemetry.trace import SpanKind
+
+from .agent_details import AgentDetails
+from .constants import (
+    APPLY_GUARDRAIL_OPERATION_NAME,
+    CHANNEL_LINK_KEY,
+    CHANNEL_NAME_KEY,
+    GEN_AI_CONVERSATION_ID_KEY,
+    GUARDIAN_ID_KEY,
+    GUARDIAN_NAME_KEY,
+    GUARDIAN_PROVIDER_NAME_KEY,
+    GUARDIAN_VERSION_KEY,
+    SECURITY_CONTENT_INPUT_HASH_KEY,
+    SECURITY_CONTENT_INPUT_VALUE_KEY,
+    SECURITY_CONTENT_MODIFIED_KEY,
+    SECURITY_CONTENT_OUTPUT_VALUE_KEY,
+    SECURITY_DECISION_CODE_KEY,
+    SECURITY_DECISION_REASON_KEY,
+    SECURITY_DECISION_TYPE_KEY,
+    SECURITY_EXTERNAL_EVENT_ID_KEY,
+    SECURITY_FINDING_EVENT_NAME,
+    SECURITY_POLICY_DECISION_TYPE_KEY,
+    SECURITY_POLICY_ID_KEY,
+    SECURITY_POLICY_NAME_KEY,
+    SECURITY_POLICY_VERSION_KEY,
+    SECURITY_RISK_CATEGORY_KEY,
+    SECURITY_RISK_METADATA_KEY,
+    SECURITY_RISK_SCORE_KEY,
+    SECURITY_RISK_SEVERITY_KEY,
+    SECURITY_TARGET_ID_KEY,
+    SECURITY_TARGET_TYPE_KEY,
+)
+from .guardrail_decision_type import GuardrailDecisionType
+from .guardrail_details import GuardrailDetails
+from .guardrail_finding import GuardrailFinding
+from .models.user_details import UserDetails
+from .opentelemetry_scope import OpenTelemetryScope
+from .request import Request
+from .span_details import SpanDetails
+
+
+class ApplyGuardrailScope(OpenTelemetryScope):
+    """Provides OpenTelemetry tracing scope for security guardrail evaluation operations.
+
+    Describes a security guardian evaluation. Multiple guardian spans MAY exist under a
+    single operation span if multiple guardians are chained.
+
+    Guardian spans SHOULD be children of the operation span they are protecting
+    (e.g., inference or execute_tool spans).
+    """
+
+    @staticmethod
+    def start(
+        details: GuardrailDetails,
+        agent_details: AgentDetails,
+        request: Request | None = None,
+        user_details: UserDetails | None = None,
+        span_details: SpanDetails | None = None,
+    ) -> "ApplyGuardrailScope":
+        """Creates and starts a new scope for guardrail evaluation tracing.
+
+        Args:
+            details: Details of the guardrail evaluation (target, decision, guardian
+                info, policy).
+            agent_details: Information about the agent being guarded.
+            request: Optional request details for conversation context.
+            user_details: Optional human user details.
+            span_details: Optional span configuration (parent context, timing, kind,
+                span links).
+
+        Returns:
+            A new ApplyGuardrailScope instance.
+        """
+        return ApplyGuardrailScope(details, agent_details, request, user_details, span_details)
+
+    def __init__(
+        self,
+        details: GuardrailDetails,
+        agent_details: AgentDetails,
+        request: Request | None = None,
+        user_details: UserDetails | None = None,
+        span_details: SpanDetails | None = None,
+    ):
+        """Initialize the guardrail evaluation scope.
+
+        Args:
+            details: Details of the guardrail evaluation.
+            agent_details: Information about the agent being guarded.
+            request: Optional request details for conversation context.
+            user_details: Optional human user details.
+            span_details: Optional span configuration.
+        """
+        resolved_span_details = SpanDetails(
+            span_kind=(
+                span_details.span_kind
+                if span_details and span_details.span_kind
+                else SpanKind.INTERNAL
+            ),
+            parent_context=span_details.parent_context if span_details else None,
+            start_time=span_details.start_time if span_details else None,
+            end_time=span_details.end_time if span_details else None,
+            span_links=span_details.span_links if span_details else None,
+        )
+
+        activity_name = self._build_activity_name(details)
+
+        super().__init__(
+            operation_name=APPLY_GUARDRAIL_OPERATION_NAME,
+            activity_name=activity_name,
+            agent_details=agent_details,
+            span_details=resolved_span_details,
+        )
+
+        # Required attributes
+        self.set_tag_maybe(SECURITY_DECISION_TYPE_KEY, details.decision_type.value)
+        self.set_tag_maybe(SECURITY_TARGET_TYPE_KEY, details.target_type)
+
+        # Guardian attributes
+        self.set_tag_maybe(GUARDIAN_ID_KEY, details.guardian_id)
+        self.set_tag_maybe(GUARDIAN_NAME_KEY, details.guardian_name)
+        self.set_tag_maybe(GUARDIAN_PROVIDER_NAME_KEY, details.guardian_provider_name)
+        self.set_tag_maybe(GUARDIAN_VERSION_KEY, details.guardian_version)
+
+        # Target attributes
+        self.set_tag_maybe(SECURITY_TARGET_ID_KEY, details.target_id)
+
+        # Decision attributes
+        self.set_tag_maybe(SECURITY_DECISION_REASON_KEY, details.decision_reason)
+        self.set_tag_maybe(SECURITY_DECISION_CODE_KEY, details.decision_code)
+
+        # Policy attributes
+        self.set_tag_maybe(SECURITY_POLICY_ID_KEY, details.policy_id)
+        self.set_tag_maybe(SECURITY_POLICY_NAME_KEY, details.policy_name)
+        self.set_tag_maybe(SECURITY_POLICY_VERSION_KEY, details.policy_version)
+
+        # Content attributes
+        self.set_tag_maybe(SECURITY_CONTENT_INPUT_HASH_KEY, details.content_input_hash)
+        if details.content_modified is not None:
+            self.set_tag_maybe(SECURITY_CONTENT_MODIFIED_KEY, details.content_modified)
+
+        # Correlation
+        self.set_tag_maybe(SECURITY_EXTERNAL_EVENT_ID_KEY, details.external_event_id)
+
+        # Request context
+        if request is not None:
+            self.set_tag_maybe(SECURITY_CONTENT_INPUT_VALUE_KEY, request.content)
+            self.set_tag_maybe(GEN_AI_CONVERSATION_ID_KEY, request.conversation_id)
+            if request.channel is not None:
+                self.set_tag_maybe(CHANNEL_NAME_KEY, request.channel.name)
+                self.set_tag_maybe(CHANNEL_LINK_KEY, request.channel.link)
+
+    def record_decision(
+        self, decision_type: GuardrailDecisionType, reason: str | None = None
+    ) -> None:
+        """Records an updated decision on the guardrail span.
+
+        Use this when the guardrail decision is determined after span creation.
+
+        Args:
+            decision_type: The decision type made by the guardian.
+            reason: Optional human-readable explanation for the decision.
+        """
+        self.set_tag_maybe(SECURITY_DECISION_TYPE_KEY, decision_type.value)
+        self.set_tag_maybe(SECURITY_DECISION_REASON_KEY, reason)
+
+    def record_content_output(self, output_value: str) -> None:
+        """Records the output content value for the guardrail evaluation (opt-in).
+
+        Args:
+            output_value: The output content after guardrail processing.
+        """
+        self.set_tag_maybe(SECURITY_CONTENT_OUTPUT_VALUE_KEY, output_value)
+
+    def record_finding(self, finding: GuardrailFinding) -> None:
+        """Records a security finding event on the current span.
+
+        Multiple findings may be recorded per guardrail evaluation.
+
+        Args:
+            finding: The security finding to record.
+
+        Raises:
+            ValueError: When finding is None.
+        """
+        if finding is None:
+            raise ValueError("finding must not be None")
+
+        if self._span is None or not self._is_telemetry_enabled():
+            return
+
+        attributes: dict[str, str | float | list[str]] = {
+            SECURITY_RISK_CATEGORY_KEY: finding.risk_category,
+            SECURITY_RISK_SEVERITY_KEY: finding.risk_severity,
+        }
+
+        if finding.policy_decision_type is not None:
+            attributes[SECURITY_POLICY_DECISION_TYPE_KEY] = finding.policy_decision_type
+
+        if finding.policy_id is not None:
+            attributes[SECURITY_POLICY_ID_KEY] = finding.policy_id
+
+        if finding.policy_name is not None:
+            attributes[SECURITY_POLICY_NAME_KEY] = finding.policy_name
+
+        if finding.policy_version is not None:
+            attributes[SECURITY_POLICY_VERSION_KEY] = finding.policy_version
+
+        if finding.risk_score is not None:
+            attributes[SECURITY_RISK_SCORE_KEY] = finding.risk_score
+
+        if finding.risk_metadata is not None:
+            attributes[SECURITY_RISK_METADATA_KEY] = finding.risk_metadata
+
+        self._span.add_event(SECURITY_FINDING_EVENT_NAME, attributes=attributes)
+
+    @staticmethod
+    def _build_activity_name(details: GuardrailDetails) -> str:
+        """Build the display name for the span."""
+        if details.guardian_name:
+            return f"{APPLY_GUARDRAIL_OPERATION_NAME} {details.guardian_name} {details.target_type}"
+        return f"{APPLY_GUARDRAIL_OPERATION_NAME} {details.target_type}"

libraries/microsoft-agents-a365-observability-core/microsoft_agents_a365/observability/core/constants.py

@@ -7,6 +7,7 @@
 INVOKE_AGENT_OPERATION_NAME = "invoke_agent"
 EXECUTE_TOOL_OPERATION_NAME = "execute_tool"
 OUTPUT_MESSAGES_OPERATION_NAME = "output_messages"
+APPLY_GUARDRAIL_OPERATION_NAME = "apply_guardrail"
 CHAT_OPERATION_NAME = "chat"
 
 # OpenTelemetry semantic conventions
@@ -112,3 +113,32 @@
 TELEMETRY_SDK_VERSION_KEY = "telemetry.sdk.version"
 TELEMETRY_SDK_NAME_VALUE = "A365ObservabilitySDK"
 TELEMETRY_SDK_LANGUAGE_VALUE = "python"
+
+# Guardian attributes
+GUARDIAN_ID_KEY = "microsoft.guardian.id"
+GUARDIAN_NAME_KEY = "microsoft.guardian.name"
+GUARDIAN_PROVIDER_NAME_KEY = "microsoft.guardian.provider.name"
+GUARDIAN_VERSION_KEY = "microsoft.guardian.version"
+
+# Security attributes
+SECURITY_DECISION_TYPE_KEY = "microsoft.security.decision.type"
+SECURITY_DECISION_REASON_KEY = "microsoft.security.decision.reason"
+SECURITY_DECISION_CODE_KEY = "microsoft.security.decision.code"
+SECURITY_TARGET_TYPE_KEY = "microsoft.security.target.type"
+SECURITY_TARGET_ID_KEY = "microsoft.security.target.id"
+SECURITY_POLICY_ID_KEY = "microsoft.security.policy.id"
+SECURITY_POLICY_NAME_KEY = "microsoft.security.policy.name"
+SECURITY_POLICY_VERSION_KEY = "microsoft.security.policy.version"
+SECURITY_CONTENT_INPUT_HASH_KEY = "microsoft.security.content.input.hash"
+SECURITY_CONTENT_MODIFIED_KEY = "microsoft.security.content.modified"
+SECURITY_EXTERNAL_EVENT_ID_KEY = "microsoft.security.external_event_id"
+SECURITY_CONTENT_INPUT_VALUE_KEY = "microsoft.security.content.input.value"
+SECURITY_CONTENT_OUTPUT_VALUE_KEY = "microsoft.security.content.output.value"
+
+# Security finding event
+SECURITY_FINDING_EVENT_NAME = "microsoft.security.finding"
+SECURITY_RISK_CATEGORY_KEY = "microsoft.security.risk.category"
+SECURITY_RISK_SEVERITY_KEY = "microsoft.security.risk.severity"
+SECURITY_RISK_SCORE_KEY = "microsoft.security.risk.score"
+SECURITY_RISK_METADATA_KEY = "microsoft.security.risk.metadata"
+SECURITY_POLICY_DECISION_TYPE_KEY = "microsoft.security.policy.decision.type"

libraries/microsoft-agents-a365-observability-core/microsoft_agents_a365/observability/core/exporters/utils.py

@@ -14,6 +14,7 @@
 from opentelemetry.trace import SpanKind, StatusCode
 
 from ..constants import (
+    APPLY_GUARDRAIL_OPERATION_NAME,
     CHAT_OPERATION_NAME,
     ENABLE_A365_OBSERVABILITY_EXPORTER,
     EXECUTE_TOOL_OPERATION_NAME,
@@ -38,6 +39,7 @@
         INVOKE_AGENT_OPERATION_NAME,
         EXECUTE_TOOL_OPERATION_NAME,
         OUTPUT_MESSAGES_OPERATION_NAME,
+        APPLY_GUARDRAIL_OPERATION_NAME,
         CHAT_OPERATION_NAME,
         InferenceOperationType.CHAT.value,
     }

libraries/microsoft-agents-a365-observability-core/microsoft_agents_a365/observability/core/guardrail_decision_type.py

@@ -0,0 +1,23 @@
+# Copyright (c) Microsoft Corporation.
+# Licensed under the MIT License.
+
+from enum import Enum
+
+
+class GuardrailDecisionType(Enum):
+    """The decision made by a security guardian during guardrail evaluation."""
+
+    ALLOW = "allow"
+    """Content or action is allowed to proceed."""
+
+    AUDIT = "audit"
+    """Content or action is logged for review but allowed to proceed."""
+
+    DENY = "deny"
+    """Content or action is denied/blocked."""
+
+    MODIFY = "modify"
+    """Content was modified (e.g., redacted, sanitized, rewritten)."""
+
+    WARN = "warn"
+    """Content or action triggered a warning but is allowed to proceed."""

libraries/microsoft-agents-a365-observability-core/microsoft_agents_a365/observability/core/guardrail_details.py

@@ -0,0 +1,47 @@
+# Copyright (c) Microsoft Corporation.
+# Licensed under the MIT License.
+
+from dataclasses import dataclass
+
+from .guardrail_decision_type import GuardrailDecisionType
+
+
+@dataclass(frozen=True)
+class GuardrailDetails:
+    """Details of a guardrail evaluation for security operations tracing.
+
+    Args:
+        target_type: The type of content or action the guardrail is applied to (required).
+            See GuardrailTargetType for well-known values.
+        decision_type: The decision made by the guardian (required).
+        guardian_name: Human-readable name of the guardian.
+        guardian_id: Unique identifier of the guardian.
+        guardian_provider_name: Provider of the guardian service
+            (e.g., azure.ai.content_safety).
+        guardian_version: Version of the guardian.
+        target_id: Identifier of the target being guarded.
+        decision_reason: Human-readable explanation for the decision.
+        decision_code: Machine-readable decision code.
+        policy_id: Identifier of the policy that triggered the decision.
+        policy_name: Human-readable name of the policy.
+        policy_version: Version of the policy.
+        content_input_hash: Hash of the input content for forensic correlation.
+        content_modified: Whether content was modified by the guardrail.
+        external_event_id: External correlation identifier for SIEM systems.
+    """
+
+    target_type: str
+    decision_type: GuardrailDecisionType
+    guardian_name: str | None = None
+    guardian_id: str | None = None
+    guardian_provider_name: str | None = None
+    guardian_version: str | None = None
+    target_id: str | None = None
+    decision_reason: str | None = None
+    decision_code: str | None = None
+    policy_id: str | None = None
+    policy_name: str | None = None
+    policy_version: str | None = None
+    content_input_hash: str | None = None
+    content_modified: bool | None = None
+    external_event_id: str | None = None