[STYLE]: Fixup Linting of Initial Files

pyproject.toml

@@ -35,6 +35,24 @@ markers = [
 [tool.ruff.lint]
 select = ["ALL"]
 
+[tool.ruff.lint.per-file-ignores]
+"tests/**" = [
+    "S101",                                    # assert is pytest's API
+    "D100", "D101", "D102", "D104", "D107",    # no docstrings needed
+    "ANN001", "ANN201", "ANN202", "ANN401",    # no type annotations needed
+    "PLR2004",                                 # magic values in assertions are fine
+    "ARG001", "ARG002",                        # unused args (fixtures, stubs)
+    "PLC0415",                                 # imports inside functions for isolation
+    "SLF001",                                  # testing private members is valid
+    "TRY003", "EM101", "EM102",                # exception message style
+    "BLE001",                                  # catching Exception in tests
+    "TRY301",                                  # raise in try blocks
+    "S108",                                    # /tmp usage
+    "PT017", "PT018",                          # assertion style
+    "ASYNC240",                                # pathlib in async tests
+    "PERF401", "RUF015", "PTH123",             # micro-optimizations / style
+]
+
 [tool.ruff.lint.flake8-copyright]
 notice-rgx = "Copyright \\(c\\) Microsoft Corporation\\.\\s*\\n.*Licensed under the MIT license"
 

rampart/__init__.py

@@ -6,6 +6,7 @@
 Public API re-exports for convenient top-level access.
 """
 
+from rampart.attacks import Attacks
 from rampart.core.adapter import AgentAdapter, Session
 from rampart.core.errors import InfrastructureError
 from rampart.core.evaluator import BaseEvaluator, Evaluator
@@ -40,7 +41,6 @@
     ToolCall,
     Turn,
 )
-from rampart.attacks import Attacks
 from rampart.probes import Probes
 from rampart.pytest_plugin._collection import record_result
 
@@ -82,4 +82,4 @@
     "record_result",
     "resolve_as_attack",
     "resolve_as_probe",
-]
+]

rampart/_pyrit/llm_bridge.py

@@ -12,26 +12,31 @@
 
 from __future__ import annotations
 
-from typing import Any
+from typing import TYPE_CHECKING, Any
+from uuid import uuid4
 
+from pyrit.models import MessagePiece
 from pyrit.prompt_target import OpenAIChatTarget, PromptChatTarget
 
-from rampart.core.llm import LLMConfig
+if TYPE_CHECKING:
+    from rampart.core.llm import LLMConfig
 
 # OpenAIChatTarget constructor parameters that can be forwarded
 # from LLMConfig.metadata.  Kept explicit so unrecognised keys
 # are silently ignored rather than causing PyRIT TypeErrors.
-_FORWARDED_MODEL_PARAMS: frozenset[str] = frozenset({
-    "frequency_penalty",
-    "max_completion_tokens",
-    "max_requests_per_minute",
-    "max_tokens",
-    "n",
-    "presence_penalty",
-    "seed",
-    "temperature",
-    "top_p",
-})
+_FORWARDED_MODEL_PARAMS: frozenset[str] = frozenset(
+    {
+        "frequency_penalty",
+        "max_completion_tokens",
+        "max_requests_per_minute",
+        "max_tokens",
+        "n",
+        "presence_penalty",
+        "seed",
+        "temperature",
+        "top_p",
+    },
+)
 
 
 def create_prompt_target(config: LLMConfig) -> PromptChatTarget:
@@ -84,12 +89,14 @@ def create_prompt_target(config: LLMConfig) -> PromptChatTarget:
 def _validate(config: LLMConfig) -> None:
     """Raise early with clear messages for missing required fields."""
     if not config.model:
+        msg = "LLMConfig.model is required (e.g. 'gpt-4o')."
         raise ValueError(
-            "LLMConfig.model is required (e.g. 'gpt-4o')."
+            msg,
         )
     if not config.endpoint:
+        msg = "LLMConfig.endpoint is required (e.g. 'https://api.openai.com/v1')."
         raise ValueError(
-            "LLMConfig.endpoint is required (e.g. 'https://api.openai.com/v1')."
+            msg,
         )
 
 
@@ -123,12 +130,8 @@ async def send_generation_request_async(
     Returns:
         str: The LLM's text response.
     """
-    import uuid
-
-    from pyrit.models import MessagePiece
-
     target = create_prompt_target(config)
-    conversation_id = str(uuid.uuid4())
+    conversation_id = str(uuid4())
 
     target.set_system_prompt(
         system_prompt=system_message,

rampart/attacks/__init__.py

@@ -9,20 +9,23 @@
 
 from __future__ import annotations
 
+from typing import TYPE_CHECKING
+
 from rampart.attacks._xpia import XPIAExecution
-from rampart.core.evaluator import Evaluator
-from rampart.core.execution import BaseExecution, ExecutionEventHandler
-from rampart.core.injection import InjectionHandle
-from rampart.core.prompt_driver import PromptDriver
-from rampart.core.types import Request
 from rampart.drivers import _coerce_driver
 
+if TYPE_CHECKING:
+    from rampart.core.evaluator import Evaluator
+    from rampart.core.execution import BaseExecution, ExecutionEventHandler
+    from rampart.core.injection import InjectionHandle
+    from rampart.core.prompt_driver import PromptDriver
+    from rampart.core.types import Request
+
 __all__ = ["Attacks", "XPIAExecution"]
 
 
 class Attacks:
-    """
-    Factory methods for attack test executions.
+    """Factory methods for attack test executions.
 
     Each method returns a BaseExecution.  The test calls
     ``execute_async`` and asserts the result.
@@ -47,8 +50,7 @@ def xpia(
         max_turns: int = 25,
         event_handlers: list[ExecutionEventHandler] | None = None,
     ) -> BaseExecution:
-        """
-        Create an XPIA attack execution.
+        """Create an XPIA attack execution.
 
         Orchestrates the full XPIA flow: inject payloads into surfaces,
         wait for indexing, create a session, drive the trigger

rampart/attacks/_xpia.py

@@ -28,7 +28,6 @@
     InjectionRecord,
     ObservabilityLevel,
     PromptDriver,
-    Request,
     Result,
     SafetyStatus,
     Turn,
@@ -39,8 +38,7 @@
 
 
 class XPIAExecution(BaseExecution):
-    """
-    Executes the full XPIA attack lifecycle.
+    """Executes the full XPIA attack lifecycle.
 
     Inherits BaseExecution.  Implements ``_execute_async`` with XPIA's
     specific phase structure.  The lifecycle skeleton (event dispatch,
@@ -95,8 +93,7 @@ def strategy_name(self) -> str:
         return "xpia"
 
     async def _execute_async(self, *, adapter: AgentAdapter) -> Result:
-        """
-        Orchestrate the XPIA lifecycle and return a safety Result.
+        """Orchestrate the XPIA lifecycle and return a safety Result.
 
         Delegates phase execution to ``_run_phases_async`` and result
         construction to ``_build_attack_result`` or
@@ -116,17 +113,22 @@ async def _execute_async(self, *, adapter: AgentAdapter) -> Result:
         )
         if max_turns_hit:
             return self._max_turns_error_result(
-                adapter=adapter, turns=turns, eval_results=eval_results,
+                adapter=adapter,
+                turns=turns,
+                eval_results=eval_results,
             )
         return self._build_attack_result(
-            adapter=adapter, turns=turns, eval_results=eval_results,
+            adapter=adapter,
+            turns=turns,
+            eval_results=eval_results,
         )
 
     async def _run_phases_async(
-        self, *, adapter: AgentAdapter,
+        self,
+        *,
+        adapter: AgentAdapter,
     ) -> tuple[list[Turn], list[EvalResult], bool]:
-        """
-        Run XPIA phases 1-5 inside a cleanup-guaranteed context.
+        """Run XPIA phases 1-5 inside a cleanup-guaranteed context.
 
         Args:
             adapter (AgentAdapter): The agent adapter.
@@ -140,7 +142,7 @@ async def _run_phases_async(
         async with AsyncExitStack() as stack:
             await self._activate_handles_async(stack=stack)
             session = await stack.enter_async_context(
-                await adapter.create_session_async()
+                await adapter.create_session_async(),
             )
 
             for turn_index in range(self._max_turns):
@@ -150,16 +152,19 @@ async def _run_phases_async(
 
                 request = decision.request
                 response = await session.send_async(request)
-                turns.append(Turn(
-                    request=request,
-                    response=response,
-                    turn_number=turn_index,
-                    driver_reasoning=decision.reasoning,
-                ))
+                turns.append(
+                    Turn(
+                        request=request,
+                        response=response,
+                        turn_number=turn_index,
+                        driver_reasoning=decision.reasoning,
+                    ),
+                )
 
                 eval_result = await self._evaluator.evaluate_async(
                     context=EvalContext(
-                        turns=list(turns), manifest=adapter.manifest,
+                        turns=list(turns),
+                        manifest=adapter.manifest,
                     ),
                 )
                 eval_results.append(eval_result)
@@ -172,10 +177,11 @@ async def _run_phases_async(
         return turns, eval_results, False
 
     async def _activate_handles_async(
-        self, *, stack: AsyncExitStack,
+        self,
+        *,
+        stack: AsyncExitStack,
     ) -> None:
-        """
-        Activate all injection handles and wait for indexing.
+        """Activate all injection handles and wait for indexing.
 
         Args:
             stack (AsyncExitStack): The exit stack managing cleanup.
@@ -184,7 +190,8 @@ async def _activate_handles_async(
             await stack.enter_async_context(handle)
 
         delay = max(
-            (h.indexing_delay_seconds for h in self._handles), default=0.0,
+            (h.indexing_delay_seconds for h in self._handles),
+            default=0.0,
         )
         if delay > 0:
             await asyncio.sleep(delay)
@@ -196,8 +203,7 @@ def _build_attack_result(
         turns: list[Turn],
         eval_results: list[EvalResult],
     ) -> Result:
-        """
-        Resolve eval results into a final attack Result.
+        """Resolve eval results into a final attack Result.
 
         Applies observability adjustment when the initial verdict is SAFE.
 
@@ -213,7 +219,10 @@ def _build_attack_result(
 
         if status == SafetyStatus.SAFE:
             safe, status = self._adjust_for_observability(
-                adapter=adapter, turns=turns, safe=safe, status=status,
+                adapter=adapter,
+                turns=turns,
+                safe=safe,
+                status=status,
             )
 
         return Result(
@@ -235,8 +244,7 @@ def _max_turns_error_result(
         turns: list[Turn],
         eval_results: list[EvalResult],
     ) -> Result:
-        """
-        Build an ERROR result when the driver exceeds max_turns.
+        """Build an ERROR result when the driver exceeds max_turns.
 
         Args:
             adapter (AgentAdapter): The adapter under test.
@@ -255,8 +263,7 @@ def _max_turns_error_result(
             safe=False,
             status=SafetyStatus.ERROR,
             summary=(
-                f"Max turns ({self._max_turns}) reached"
-                " — driver did not terminate"
+                f"Max turns ({self._max_turns}) reached — driver did not terminate"
             ),
             turns=turns,
             eval_results=eval_results,
@@ -284,8 +291,7 @@ def _adjust_for_observability(
         safe: bool,
         status: SafetyStatus,
     ) -> tuple[bool, SafetyStatus]:
-        """
-        Downgrade SAFE to UNDETERMINED when the adapter lacks observability.
+        """Downgrade SAFE to UNDETERMINED when the adapter lacks observability.
 
         If the evaluator returned NOT_DETECTED (and ``resolve_as_attack``
         produced SAFE), but the adapter has RESPONSE_ONLY observability
@@ -323,8 +329,7 @@ def _collect_response_metadata(
     *,
     turns: list[Turn],
 ) -> dict[str, Any]:
-    """
-    Aggregate response metadata from all turns.
+    """Aggregate response metadata from all turns.
 
     Adapters can attach diagnostic data (e.g. conversation_id, session_id)
     to Response.metadata. This function collects it into a single dict
@@ -355,8 +360,7 @@ def _build_summary(
     status: SafetyStatus,
     eval_results: list[EvalResult],
 ) -> str:
-    """
-    Build a human-readable one-line summary for XPIA results.
+    """Build a human-readable one-line summary for XPIA results.
 
     Args:
         status (SafetyStatus): The resolved safety status.
@@ -374,7 +378,9 @@ def _build_summary(
         return f"Attack objective detected: {'; '.join(evidence[:3])}"
     if status == SafetyStatus.UNDETERMINED:
         rationales = [er.rationale for er in eval_results if er.rationale]
-        detail = "; ".join(rationales[:2]) if rationales else "Insufficient observability"
+        detail = (
+            "; ".join(rationales[:2]) if rationales else "Insufficient observability"
+        )
         return f"Evaluation undetermined: {detail}"
     if status == SafetyStatus.ERROR:
         return "Infrastructure error during execution"