[FIX]: Fixup pyright Errors

pyproject.toml

@@ -68,7 +68,12 @@ skip_empty = true
 [tool.pyright]
 pythonVersion = "3.11"
 typeCheckingMode = "strict"
-include = ["rampart"]
+include = ["rampart", "tests"]
+
+[[tool.pyright.executionEnvironments]]
+root = "tests"
+extraPaths = ["."]
+reportPrivateUsage = false
 
 [tool.pytest.ini_options]
 asyncio_mode = "auto"

rampart/_pyrit/llm_bridge.py

@@ -12,10 +12,10 @@
 
 from __future__ import annotations
 
-from typing import TYPE_CHECKING, Any
+from typing import TYPE_CHECKING, Any, cast
 from uuid import uuid4
 
-from pyrit.models import MessagePiece
+from pyrit.models import Message, MessagePiece
 from pyrit.prompt_target import OpenAIChatTarget, PromptChatTarget
 
 if TYPE_CHECKING:
@@ -143,7 +143,9 @@ async def send_generation_request_async(
         original_value=user_message,
         conversation_id=conversation_id,
     )
-    request = request_piece.to_message()
+
+    # Can remove after bumping to PyRIT v0.13.0
+    request = cast("Message", request_piece.to_message())  # pyright: ignore[reportUnknownMemberType]
 
     responses = await target.send_prompt_async(message=request)
     return responses[0].get_value()

rampart/attacks/__init__.py

@@ -12,7 +12,7 @@
 from typing import TYPE_CHECKING
 
 from rampart.attacks._xpia import XPIAExecution
-from rampart.drivers import _coerce_driver
+from rampart.drivers._utils import coerce_driver
 
 if TYPE_CHECKING:
     from rampart.core.evaluator import Evaluator
@@ -94,7 +94,7 @@ def xpia(
             handles = inject
         else:
             handles = [inject]
-        driver = _coerce_driver(trigger)
+        driver = coerce_driver(trigger)
 
         return XPIAExecution(
             handles=handles,

rampart/core/llm.py

@@ -32,4 +32,4 @@ class LLMConfig:
     endpoint: str
     api_key: str | None = field(default=None, repr=False)
     deployment: str | None = None
-    metadata: dict[str, Any] = field(default_factory=dict)
+    metadata: dict[str, Any] = field(default_factory=dict[str, Any])

rampart/core/manifest.py

@@ -26,8 +26,8 @@ class ToolDeclaration:
 
     name: str
     description: str = ""
-    parameters: dict[str, Any] = field(default_factory=dict)
-    permissions: list[str] = field(default_factory=list)
+    parameters: dict[str, Any] = field(default_factory=dict[str, Any])
+    permissions: list[str] = field(default_factory=list[str])
 
 
 @dataclass(kw_only=True)
@@ -63,10 +63,14 @@ class AppManifest:
     """
 
     name: str
-    tools: list[ToolDeclaration] = field(default_factory=list)
-    data_sources: list[DataSource] = field(default_factory=list)
+    tools: list[ToolDeclaration] = field(
+        default_factory=list[ToolDeclaration],
+    )
+    data_sources: list[DataSource] = field(
+        default_factory=list[DataSource],
+    )
     description: str = ""
-    metadata: dict[str, Any] = field(default_factory=dict)
+    metadata: dict[str, Any] = field(default_factory=dict[str, Any])
 
     def declares_tool(self, name: str) -> bool:
         """Check if a tool is declared in the manifest."""
@@ -87,7 +91,7 @@ def __str__(self) -> str:
             sections.append(self.description)
 
         if self.tools:
-            tool_lines = []
+            tool_lines: list[str] = []
             for t in self.tools:
                 params = ", ".join(f"{k}: {v}" for k, v in t.parameters.items())
                 desc = f" — {t.description}" if t.description else ""
@@ -96,7 +100,7 @@ def __str__(self) -> str:
             sections.append(f"Available tools:\n{tools}")
 
         if self.data_sources:
-            source_lines = []
+            source_lines: list[str] = []
             for ds in self.data_sources:
                 writable = (
                     " (writable by untrusted users)" if ds.writable_by_untrusted else ""

rampart/core/result.py

@@ -122,14 +122,18 @@ class Result:
     safe: bool
     status: SafetyStatus
     summary: str
-    turns: list[Turn] = field(default_factory=list)
-    eval_results: list[EvalResult] = field(default_factory=list)
+    turns: list[Turn] = field(default_factory=list[Turn])
+    eval_results: list[EvalResult] = field(
+        default_factory=list[EvalResult],
+    )
     duration_seconds: float = 0.0
     harm_category: HarmCategory | str | None = None
     strategy: str = ""
     observability_level: ObservabilityLevel = ObservabilityLevel.RESPONSE_ONLY
-    injections: list[InjectionRecord] = field(default_factory=list)
-    metadata: dict[str, Any] = field(default_factory=dict)
+    injections: list[InjectionRecord] = field(
+        default_factory=list[InjectionRecord],
+    )
+    metadata: dict[str, Any] = field(default_factory=dict[str, Any])
 
     def __bool__(self) -> bool:
         """Assert-safe: bool(result) means the agent behaved safely."""

rampart/core/types.py

@@ -115,7 +115,7 @@ class Payload:
     id: str = field(default_factory=lambda: uuid.uuid4().hex[:12])
     format: PayloadFormat = PayloadFormat.TEXT
     artifact: Path | None = None
-    metadata: dict[str, Any] = field(default_factory=dict)
+    metadata: dict[str, Any] = field(default_factory=dict[str, Any])
 
     def __post_init__(self) -> None:
         """Validate content-format-artifact consistency."""
@@ -165,7 +165,7 @@ class ToolCall:
     """
 
     name: str
-    arguments: dict[str, Any] = field(default_factory=dict)
+    arguments: dict[str, Any] = field(default_factory=dict[str, Any])
     result: str | None = None
     timestamp: datetime | None = None
 
@@ -184,7 +184,7 @@ class SideEffect:
     """
 
     kind: str
-    details: dict[str, Any] = field(default_factory=dict)
+    details: dict[str, Any] = field(default_factory=dict[str, Any])
 
 
 @dataclass(kw_only=True)
@@ -201,9 +201,9 @@ class Response:
     """
 
     text: str
-    tool_calls: list[ToolCall] = field(default_factory=list)
-    side_effects: list[SideEffect] = field(default_factory=list)
-    metadata: dict[str, Any] = field(default_factory=dict)
+    tool_calls: list[ToolCall] = field(default_factory=list[ToolCall])
+    side_effects: list[SideEffect] = field(default_factory=list[SideEffect])
+    metadata: dict[str, Any] = field(default_factory=dict[str, Any])
 
 
 @dataclass(kw_only=True)
@@ -221,7 +221,7 @@ class Request:
     """
 
     prompt: str | None = None
-    attachments: list[Payload] = field(default_factory=list)
+    attachments: list[Payload] = field(default_factory=list[Payload])
 
     def __post_init__(self) -> None:
         """Validate that the request carries some content."""
@@ -280,7 +280,7 @@ class EvalResult:
 
     outcome: EvalOutcome
     confidence: float = 1.0
-    evidence: list[str] = field(default_factory=list)
+    evidence: list[str] = field(default_factory=list[str])
     rationale: str = ""
 
     @property
@@ -304,7 +304,7 @@ class EvalContext:
 
     turns: list[Turn]
     manifest: AppManifest | None = None
-    metadata: dict[str, Any] = field(default_factory=dict)
+    metadata: dict[str, Any] = field(default_factory=dict[str, Any])
 
     @property
     def current_turn(self) -> Turn:

rampart/drivers/__init__.py

@@ -1,46 +1,8 @@
 # Copyright (c) Microsoft Corporation.
 # Licensed under the MIT license.
 
-"""Driver implementations.
+"""Driver implementations."""
 
-Re-exports StaticDriver and provides the _coerce_driver helper
-for ergonomic prompt/driver coercion.
-"""
-
-from __future__ import annotations
-
-from rampart.core.prompt_driver import PromptDriver
-from rampart.core.types import Request
 from rampart.drivers.static import StaticDriver
 
-__all__ = ["StaticDriver", "_coerce_driver"]
-
-
-def _coerce_driver(
-    value: str | list[str] | Request | list[Request] | PromptDriver,
-) -> PromptDriver:
-    """Coerce a string, Request, or list into a PromptDriver.
-
-    Args:
-        value: A single prompt string, a list of prompt strings,
-            a single Request, a list of Requests, or an existing
-            PromptDriver.
-
-    Returns:
-        PromptDriver: A driver wrapping the input.
-    """
-    if isinstance(value, str):
-        return StaticDriver(prompts=[value])
-    if isinstance(value, Request):
-        return StaticDriver(prompts=[value])
-    if isinstance(value, list):
-        return StaticDriver(prompts=value)
-    if isinstance(value, PromptDriver):
-        return value
-    msg = (
-        f"Cannot coerce {type(value).__name__} to PromptDriver. "
-        f"Expected str, list[str], Request, list[Request], or PromptDriver."
-    )
-    raise TypeError(
-        msg,
-    )
+__all__ = ["StaticDriver"]

rampart/drivers/_utils.py

@@ -0,0 +1,28 @@
+# Copyright (c) Microsoft Corporation.
+# Licensed under the MIT license.
+
+from rampart.core.prompt_driver import PromptDriver
+from rampart.core.types import Request
+from rampart.drivers.static import StaticDriver
+
+
+def coerce_driver(
+    value: str | list[str] | Request | list[Request] | PromptDriver,
+) -> PromptDriver:
+    """Coerce a string, Request, or list into a PromptDriver.
+
+    Args:
+        value: A single prompt string, a list of prompt strings,
+            a single Request, a list of Requests, or an existing
+            PromptDriver.
+
+    Returns:
+        PromptDriver: A driver wrapping the input.
+    """
+    if isinstance(value, str):
+        return StaticDriver(prompts=[value])
+    if isinstance(value, Request):
+        return StaticDriver(prompts=[value])
+    if isinstance(value, list):
+        return StaticDriver(prompts=value)
+    return value

rampart/evaluators/response_contains.py

@@ -29,7 +29,7 @@ class ResponseContains(BaseEvaluator):
 
     def __init__(
         self,
-        target: str | re.Pattern | Callable[[str], bool],
+        target: str | re.Pattern[str] | Callable[[str], bool],
         /,
         *,
         case_sensitive: bool = False,
@@ -42,7 +42,7 @@ async def evaluate_async(self, *, context: EvalContext) -> EvalResult:
         """Check response text for the target pattern."""
         text = context.text
 
-        if callable(self._target) and not isinstance(self._target, re.Pattern):
+        if callable(self._target):
             found = self._target(text)
         elif isinstance(self._target, re.Pattern):
             found = bool(self._target.search(text))

rampart/evaluators/side_effect.py

@@ -5,22 +5,30 @@
 
 from __future__ import annotations
 
-from typing import Any
+from typing import TYPE_CHECKING, Any
 
 from rampart.core.evaluator import BaseEvaluator
 from rampart.core.types import EvalContext, EvalOutcome, EvalResult, SideEffect
 
+if TYPE_CHECKING:
+    from collections.abc import Callable
+
 
 class SideEffectOccurred(BaseEvaluator):
     """Detects whether a side effect of a given kind occurred.
 
     Args:
         kind (str): The side effect kind to look for (positional-only).
-        **detail_predicates (dict[str, Any]):
-            Detail field -> expected value or predicate.
+        **detail_predicates (dict[str, Any | Callable[[Any], bool]]):
+            Detail field -> expected value or callable predicate.
     """
 
-    def __init__(self, kind: str, /, **detail_predicates: dict[str, Any]) -> None:
+    def __init__(
+        self,
+        kind: str,
+        /,
+        **detail_predicates: Any | Callable[[Any], bool],  # noqa: ANN401
+    ) -> None:
         """Initialize with side effect kind and optional predicates."""
         self._kind = kind
         self._predicates = detail_predicates

rampart/evaluators/tool_called.py

@@ -5,11 +5,14 @@
 
 from __future__ import annotations
 
-from typing import Any
+from typing import TYPE_CHECKING, Any
 
 from rampart.core.evaluator import BaseEvaluator
 from rampart.core.types import EvalContext, EvalOutcome, EvalResult, ToolCall
 
+if TYPE_CHECKING:
+    from collections.abc import Callable
+
 
 class ToolCalled(BaseEvaluator):
     """Detects whether a tool was called, optionally matching parameters.
@@ -23,11 +26,16 @@ class ToolCalled(BaseEvaluator):
 
     Args:
         tool_name (str): The tool to look for (positional-only).
-        **param_predicates (dict[str, Any]):
+        **param_predicates (dict[str, Any | Callable[[Any], bool]]):
              Parameter name -> expected value or predicate.
     """
 
-    def __init__(self, tool_name: str, /, **param_predicates: dict[str, Any]) -> None:
+    def __init__(
+        self,
+        tool_name: str,
+        /,
+        **param_predicates: Any | Callable[[Any], bool],  # noqa: ANN401
+    ) -> None:
         """Initialize with tool name and optional parameter predicates."""
         self._tool_name = tool_name
         self._predicates = param_predicates

rampart/payloads/template.py

@@ -74,7 +74,7 @@ class PayloadTemplate:
     description: str
     objective: str
     instruction: str
-    variables: dict[str, str] = field(default_factory=dict)
+    variables: dict[str, str] = field(default_factory=dict[str, str])
 
     def with_variables(self, **overrides: str) -> PayloadTemplate:
         """Return a copy with updated variable values.

rampart/probes/__init__.py

@@ -11,7 +11,7 @@
 
 from typing import TYPE_CHECKING, overload
 
-from rampart.drivers import _coerce_driver
+from rampart.drivers._utils import coerce_driver
 from rampart.probes._single_turn import SingleTurnExecution
 
 if TYPE_CHECKING:
@@ -94,9 +94,9 @@ def behavior(  # noqa: PLR0913
                 msg,
             )
         if prompt is not None:
-            resolved_driver = _coerce_driver(prompt)
+            resolved_driver = coerce_driver(prompt)
         elif prompts is not None:
-            resolved_driver = _coerce_driver(prompts)
+            resolved_driver = coerce_driver(prompts)
         else:
             assert driver is not None  # noqa: S101  — type narrowing
             resolved_driver = driver