feat(security): emit SecurityAnalysisEvent with the analyzer's verdict per action batch

.github/agent-server-openapi-weak-schema-allowlist.json

@@ -592,5 +592,11 @@
     "kind": "empty-object-schema",
     "reason": "This existing endpoint has an opaque or non-JSON response contract and is tracked by the weak-type ratchet.",
     "owner": "OpenHands OSS"
+  },
+  {
+    "pointer": "/components/schemas/SecurityAnalysisEvent/properties/details/additionalProperties/additionalProperties",
+    "kind": "unrestricted-additional-properties",
+    "reason": "Per-action analyzer detail (probabilities, confidence, rationale) is analyzer-specific by design; the shape belongs to each SecurityAnalyzer implementation.",
+    "owner": "OpenHands OSS"
   }
 ]

openhands-sdk/openhands/sdk/agent/agent.py

@@ -50,6 +50,7 @@
     CondensationRequest,
 )
 from openhands.sdk.event.error_classification import AGENT_OUTCOME
+from openhands.sdk.event.security import SecurityAnalysisEvent
 from openhands.sdk.llm import (
     LLM,
     ImageContent,
@@ -1044,7 +1045,10 @@ async def _astep(
                 )
 
     def _requires_user_confirmation(
-        self, state: ConversationState, action_events: list[ActionEvent]
+        self,
+        state: ConversationState,
+        action_events: list[ActionEvent],
+        on_event: ConversationCallbackType | None = None,
     ) -> bool:
         """
         Decide whether user confirmation is needed to proceed.
@@ -1068,12 +1072,25 @@ def _requires_user_confirmation(
         # If a security analyzer is registered, use it to grab the risks of the actions
         # involved. If not, we'll set the risks to UNKNOWN.
         if state.security_analyzer is not None:
-            risks = [
-                risk
-                for _, risk in state.security_analyzer.analyze_pending_actions(
-                    action_events
+            analyses = state.security_analyzer.analyze_actions(action_events)
+            risks = [analysis.risk for _, analysis in analyses]
+            # Record the verdict regardless of what the policy does with it, so
+            # the log and UI show what the analyzer said even under NeverConfirm.
+            if on_event is not None:
+                on_event(
+                    SecurityAnalysisEvent(
+                        analyzer=state.security_analyzer.__class__.__name__,
+                        policy=state.confirmation_policy.__class__.__name__,
+                        risks={
+                            action.id: analysis.risk for action, analysis in analyses
+                        },
+                        details={
+                            action.id: analysis.details
+                            for action, analysis in analyses
+                            if analysis.details is not None
+                        },
+                    )
                 )
-            ]
         else:
             risks = [risk.SecurityRisk.UNKNOWN] * len(action_events)
 

openhands-sdk/openhands/sdk/agent/response_dispatch.py

@@ -128,6 +128,7 @@ def _requires_user_confirmation(
             self,
             state: ConversationState,
             action_events: list[ActionEvent],
+            on_event: ConversationCallbackType | None = None,
         ) -> bool: ...
 
         def _maybe_emit_vllm_tokens(
@@ -183,7 +184,7 @@ def _handle_tool_calls(
                 continue
             action_events.append(action_event)
 
-        if self._requires_user_confirmation(state, action_events):
+        if self._requires_user_confirmation(state, action_events, on_event):
             return
 
         if action_events:
@@ -237,7 +238,7 @@ async def _ahandle_tool_calls(
                 continue
             action_events.append(action_event)
 
-        if self._requires_user_confirmation(state, action_events):
+        if self._requires_user_confirmation(state, action_events, on_event):
             return
 
         if action_events:

openhands-sdk/openhands/sdk/event/__init__.py

@@ -22,6 +22,7 @@
     RESUME_CONTEXT_MARKER,
     render_resume_transcript,
 )
+from openhands.sdk.event.security import SecurityAnalysisEvent
 from openhands.sdk.event.streaming_delta import StreamingDeltaEvent
 from openhands.sdk.event.token import TokenEvent
 from openhands.sdk.event.types import EventID, ToolCallID
@@ -46,6 +47,7 @@
     "StreamingDeltaEvent",
     "Condensation",
     "CondensationRequest",
+    "SecurityAnalysisEvent",
     "CondensationSummaryEvent",
     "ConversationStateUpdateEvent",
     "HookExecutionEvent",

openhands-sdk/openhands/sdk/event/security.py

@@ -0,0 +1,48 @@
+from typing import Any
+
+from pydantic import Field
+from rich.text import Text
+
+from openhands.sdk.event.base import Event
+from openhands.sdk.event.types import EventID, SourceType
+from openhands.sdk.security.risk import SecurityRisk
+
+
+class SecurityAnalysisEvent(Event):
+    """Risk levels the configured security analyzer assigned to pending actions.
+
+    Emitted once per batch of actions the agent is about to execute, whenever a
+    security analyzer is configured on the conversation, regardless of whether
+    the confirmation policy then asks for confirmation. This makes the
+    analyzer's verdict visible in the event log and UI even when the policy
+    (for example ``NeverConfirm``) never consults it.
+
+    This event is not shown to the LLM.
+    """
+
+    source: SourceType = "environment"
+
+    analyzer: str = Field(
+        description="Kind of the security analyzer that produced these risks."
+    )
+    policy: str = Field(
+        description="Kind of the confirmation policy in force when the actions "
+        "were analyzed."
+    )
+    risks: dict[EventID, SecurityRisk] = Field(
+        description="Risk level per analyzed action, keyed by ActionEvent id."
+    )
+    details: dict[EventID, dict[str, Any]] = Field(
+        default_factory=dict,
+        description="Optional analyzer-specific detail per action, keyed by "
+        "ActionEvent id (for example probabilities, confidence, or a rationale).",
+    )
+
+    @property
+    def visualize(self) -> Text:
+        text = Text()
+        text.append("Security Analysis\n", style="bold")
+        text.append(f"analyzer={self.analyzer} policy={self.policy}\n")
+        for action_id, risk in self.risks.items():
+            text.append(f"  {action_id}: {risk.value}\n")
+        return text

openhands-sdk/openhands/sdk/security/__init__.py

@@ -1,4 +1,4 @@
-from openhands.sdk.security.analyzer import SecurityAnalyzerBase
+from openhands.sdk.security.analyzer import SecurityAnalysis, SecurityAnalyzerBase
 from openhands.sdk.security.confirmation_policy import (
     AlwaysConfirm,
     ConfirmationPolicyBase,
@@ -28,6 +28,7 @@
 
 __all__ = [
     "SecurityRisk",
+    "SecurityAnalysis",
     "SecurityAnalyzerBase",
     "LLMSecurityAnalyzer",
     "ToolShieldLLMSecurityAnalyzer",

openhands-sdk/openhands/sdk/security/analyzer.py

@@ -1,4 +1,7 @@
 from abc import ABC, abstractmethod
+from typing import Any
+
+from pydantic import BaseModel, Field
 
 from openhands.sdk.event.base import Event
 from openhands.sdk.event.llm_convertible import ActionEvent
@@ -12,6 +15,17 @@
 logger = get_logger(__name__)
 
 
+class SecurityAnalysis(BaseModel):
+    """Result of analyzing one action: a risk level plus optional detail."""
+
+    risk: SecurityRisk
+    details: dict[str, Any] | None = Field(
+        default=None,
+        description="Analyzer-specific detail behind the risk (probabilities, "
+        "confidence, rationale, ...). Surfaced in SecurityAnalysisEvent.",
+    )
+
+
 class SecurityAnalyzerBase(DiscriminatedUnionMixin, ABC):
     """Abstract base class for security analyzers.
 
@@ -82,30 +96,54 @@ def should_require_confirmation(
             # LOW and MEDIUM risk actions don't require confirmation by default
             return False
 
+    def analyze_action(self, action: ActionEvent) -> SecurityAnalysis:
+        """Analyze one action, returning its risk and optional detail.
+
+        The default implementation wraps :meth:`security_risk` with no detail.
+        Analyzers that can explain their verdict (probabilities, confidence, a
+        rationale) should override this instead of ``security_risk``.
+        """
+        return SecurityAnalysis(risk=self.security_risk(action))
+
+    def analyze_actions(
+        self, pending_actions: list[ActionEvent]
+    ) -> list[tuple[ActionEvent, SecurityAnalysis]]:
+        """Analyze pending actions, returning (action, analysis) pairs.
+
+        An analyzer error defaults that action to HIGH risk, with the error
+        recorded in the analysis detail.
+        """
+        analyzed: list[tuple[ActionEvent, SecurityAnalysis]] = []
+        for action_event in pending_actions:
+            try:
+                analysis = self.analyze_action(action_event)
+                logger.debug(
+                    f"Action {action_event} analyzed with risk level: {analysis.risk}"
+                )
+            except Exception as e:
+                logger.error(f"Error analyzing action {action_event}: {e}")
+                # Default to HIGH risk on analysis error for safety
+                analysis = SecurityAnalysis(
+                    risk=SecurityRisk.HIGH, details={"error": str(e)}
+                )
+            analyzed.append((action_event, analysis))
+        return analyzed
+
     def analyze_pending_actions(
         self, pending_actions: list[ActionEvent]
     ) -> list[tuple[ActionEvent, SecurityRisk]]:
         """Analyze all pending actions in a conversation.
 
-        This method gets all unmatched actions from the conversation state
-        and analyzes each one for security risks.
+        Compatibility form of :meth:`analyze_actions` that returns only the
+        risk level for each action.
 
         Args:
-            conversation: The conversation to analyze
+            pending_actions: The unmatched actions to analyze
 
         Returns:
             List of tuples containing (action, risk_level) for each pending action
         """
-        analyzed_actions = []
-
-        for action_event in pending_actions:
-            try:
-                risk = self.security_risk(action_event)
-                analyzed_actions.append((action_event, risk))
-                logger.debug(f"Action {action_event} analyzed with risk level: {risk}")
-            except Exception as e:
-                logger.error(f"Error analyzing action {action_event}: {e}")
-                # Default to HIGH risk on analysis error for safety
-                analyzed_actions.append((action_event, SecurityRisk.HIGH))
-
-        return analyzed_actions
+        return [
+            (action, analysis.risk)
+            for action, analysis in self.analyze_actions(pending_actions)
+        ]