fix: resolve all ruff lint and format errors for CI

- Fix type comparisons using `is` instead of `==` (E721)
- Combine nested if statements in conversation.py and sanitizer.py (SIM102)
- Remove unused variables in sandbox.py (F841)
- Add per-file-ignores in pyproject.toml for test patterns:
  - ARG001: Unused function arguments (common in test stubs)
  - ARG005: Unused lambda arguments (common in mock side_effects)
  - SIM117: Nested with statements (clearer for multiple mocks)
- Run ruff format on all source and test files

All 616 tests pass with 99.26% coverage.

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>

Author: sattyamj-attri

pyproject.toml

@@ -110,6 +110,13 @@ ignore = [
     "E501",   # line too long (handled by formatter)
 ]
 
+[tool.ruff.lint.per-file-ignores]
+"tests/**/*.py" = [
+    "ARG001",  # Unused function arguments common in test stubs
+    "ARG005",  # Unused lambda arguments common in mock side_effects
+    "SIM117",  # Nested with statements clearer for multiple mocks
+]
+
 [tool.ruff.lint.isort]
 known-first-party = ["agent_airlock"]
 

src/agent_airlock/__init__.py

@@ -28,6 +28,14 @@ def run_code(code: str) -> str:
     reset_context,
     set_current_context,
 )
+from .conversation import (
+    ConversationConstraints,
+    ConversationState,
+    ConversationTracker,
+    ToolCall,
+    get_conversation_tracker,
+    reset_conversation_tracker,
+)
 from .core import Airlock, SandboxExecutionError, SandboxUnavailableError, airlock
 from .policy import (
     BUSINESS_HOURS_POLICY,
@@ -60,14 +68,6 @@ def run_code(code: str) -> str:
     is_async_generator_function,
     is_generator_function,
 )
-from .conversation import (
-    ConversationConstraints,
-    ConversationState,
-    ConversationTracker,
-    ToolCall,
-    get_conversation_tracker,
-    reset_conversation_tracker,
-)
 from .validator import GhostArgumentError
 
 __version__ = "0.1.5"

src/agent_airlock/conversation.py

@@ -161,9 +161,7 @@ def _maybe_cleanup(self) -> None:
         self._last_cleanup = now
         cutoff = datetime.now(timezone.utc).timestamp() - self._ttl_seconds
         expired = [
-            sid
-            for sid, state in self._sessions.items()
-            if state.created_at.timestamp() < cutoff
+            sid for sid, state in self._sessions.items() if state.created_at.timestamp() < cutoff
         ]
 
         for sid in expired:
@@ -264,7 +262,10 @@ def should_block(
 
             # Check if conversation is blocked
             if state.is_blocked:
-                return True, f"Conversation blocked until {state.blocked_until}: {state.block_reason}"
+                return (
+                    True,
+                    f"Conversation blocked until {state.blocked_until}: {state.block_reason}",
+                )
 
             # Check cooldown_after constraints
             for trigger_tool, cooldown_seconds in constraints.cooldown_after.items():
@@ -287,8 +288,7 @@ def should_block(
                     if elapsed < cooldown:
                         remaining = int(cooldown - elapsed)
                         return True, (
-                            f"Tool '{tool_name}' cooldown: "
-                            f"{remaining}s remaining of {cooldown}s"
+                            f"Tool '{tool_name}' cooldown: {remaining}s remaining of {cooldown}s"
                         )
 
             # Check max calls per tool
@@ -297,8 +297,7 @@ def should_block(
                 current_calls = len(state.get_successful_calls_for_tool(tool_name))
                 if current_calls >= max_calls:
                     return True, (
-                        f"Tool '{tool_name}' limit reached: "
-                        f"{current_calls}/{max_calls} calls used"
+                        f"Tool '{tool_name}' limit reached: {current_calls}/{max_calls} calls used"
                     )
 
             # Check required_before constraints
@@ -308,17 +307,18 @@ def should_block(
                 missing = set(required_tools) - called_tools
                 if missing:
                     return True, (
-                        f"Tool '{tool_name}' requires prior calls to: "
-                        f"{', '.join(sorted(missing))}"
+                        f"Tool '{tool_name}' requires prior calls to: {', '.join(sorted(missing))}"
                     )
 
             # Check max total calls
-            if constraints.max_total_calls is not None:
-                if state.success_count >= constraints.max_total_calls:
-                    return True, (
-                        f"Conversation call limit reached: "
-                        f"{state.success_count}/{constraints.max_total_calls}"
-                    )
+            if (
+                constraints.max_total_calls is not None
+                and state.success_count >= constraints.max_total_calls
+            ):
+                return True, (
+                    f"Conversation call limit reached: "
+                    f"{state.success_count}/{constraints.max_total_calls}"
+                )
 
             # Check blocked ratio
             if constraints.max_blocked_ratio is not None and state.call_count > 0:

src/agent_airlock/core.py

@@ -448,9 +448,7 @@ async def async_wrapper(*args: P.args, **kwargs: P.kwargs) -> R | dict[str, Any]
                 token = set_current_context(context)
                 try:
                     if self.sandbox:
-                        result = await self._execute_in_sandbox_async(
-                            func, *args, **cleaned_kwargs
-                        )
+                        result = await self._execute_in_sandbox_async(func, *args, **cleaned_kwargs)
                     else:
                         # Await the async validated function
                         # Type ignore: validated_func preserves async nature of func

src/agent_airlock/sandbox.py

@@ -430,8 +430,6 @@ def execute_in_sandbox(
 
     # Execute in sandbox
     pool = get_sandbox_pool(config)
-    stdout_lines: list[str] = []
-    stderr_lines: list[str] = []
 
     try:
         with pool.sandbox() as sandbox:

src/agent_airlock/sanitizer.py

@@ -616,13 +616,15 @@ def sanitize_with_workspace_config(
         data_type = SensitiveDataType(detection["type"])
 
         # Apply workspace-specific filtering
-        if data_type == SensitiveDataType.EMAIL:
-            if not workspace_config.should_mask_email(detection["value"]):
-                continue
+        if data_type == SensitiveDataType.EMAIL and not workspace_config.should_mask_email(
+            detection["value"]
+        ):
+            continue
 
-        if data_type == SensitiveDataType.PHONE:
-            if not workspace_config.should_mask_phone(detection["value"]):
-                continue
+        if data_type == SensitiveDataType.PHONE and not workspace_config.should_mask_phone(
+            detection["value"]
+        ):
+            continue
 
         filtered_detections.append(detection)
 
@@ -650,10 +652,12 @@ def sanitize_with_workspace_config(
         # Handle custom patterns
         if type_str.startswith("custom:"):
             pattern_name = type_str[7:]
-            strategy = workspace_config.custom_strategies.get(
-                pattern_name, MaskingStrategy.FULL
+            strategy = workspace_config.custom_strategies.get(pattern_name, MaskingStrategy.FULL)
+            masked = (
+                f"[{pattern_name.upper()}]"
+                if strategy == MaskingStrategy.TYPE_ONLY
+                else "[REDACTED]"
             )
-            masked = f"[{pattern_name.upper()}]" if strategy == MaskingStrategy.TYPE_ONLY else "[REDACTED]"
         else:
             data_type = SensitiveDataType(type_str)
             strategy = mask_config.get(data_type, MaskingStrategy.FULL)
@@ -664,21 +668,15 @@ def sanitize_with_workspace_config(
                 masked_value = _mask_value(password_value, data_type, strategy)
                 masked = full_match.replace(password_value, masked_value)
                 result_text = (
-                    result_text[: detection["start"]]
-                    + masked
-                    + result_text[detection["end"] :]
+                    result_text[: detection["start"]] + masked + result_text[detection["end"] :]
                 )
                 detection["masked_as"] = "[REDACTED]"
                 detections.append(detection)
                 continue
             else:
                 masked = _mask_value(detection["value"], data_type, strategy)
 
-        result_text = (
-            result_text[: detection["start"]]
-            + masked
-            + result_text[detection["end"] :]
-        )
+        result_text = result_text[: detection["start"]] + masked + result_text[detection["end"] :]
         detection["masked_as"] = masked
         detections.append(detection)
 

src/agent_airlock/streaming.py

@@ -361,6 +361,5 @@ def sync_wrapper(*args: Any, **kwargs: Any) -> Generator[Any, None, None]:
 
     else:
         raise TypeError(
-            f"create_streaming_wrapper requires a generator function, "
-            f"got {type(func).__name__}"
+            f"create_streaming_wrapper requires a generator function, got {type(func).__name__}"
         )

tests/test_azure_compatibility.py

@@ -14,8 +14,6 @@
 import json
 from typing import Any, Literal, get_type_hints
 
-import pytest
-
 from agent_airlock import Airlock, AirlockConfig, SecurityPolicy
 
 
@@ -58,13 +56,13 @@ def get_weather(
             prop: dict[str, Any] = {}
 
             # Map Python types to JSON Schema types
-            if hint == str or str(hint) == "str":
+            if hint is str or str(hint) == "str":
                 prop["type"] = "string"
-            elif hint == int or str(hint) == "int":
+            elif hint is int or str(hint) == "int":
                 prop["type"] = "integer"
-            elif hint == float or str(hint) == "float":
+            elif hint is float or str(hint) == "float":
                 prop["type"] = "number"
-            elif hint == bool or str(hint) == "bool":
+            elif hint is bool or str(hint) == "bool":
                 prop["type"] = "boolean"
             elif hasattr(hint, "__origin__") and hint.__origin__ is Literal:
                 prop["type"] = "string"
@@ -387,13 +385,13 @@ def _python_type_to_json(hint: Any) -> str:
         """Convert Python type hint to JSON schema type."""
         if hint is None:
             return "string"
-        if hint == str or str(hint) == "str":
+        if hint is str or str(hint) == "str":
             return "string"
-        if hint == int or str(hint) == "int":
+        if hint is int or str(hint) == "int":
             return "integer"
-        if hint == float or str(hint) == "float":
+        if hint is float or str(hint) == "float":
             return "number"
-        if hint == bool or str(hint) == "bool":
+        if hint is bool or str(hint) == "bool":
             return "boolean"
         return "string"
 
@@ -420,7 +418,9 @@ def rate_limited_tool(x: int) -> int:
         result = rate_limited_tool(x=2)
         assert isinstance(result, dict)
         assert result.get("status") == "blocked"
-        assert "rate" in result.get("reason", "").lower() or "limit" in result.get("error", "").lower()
+        assert (
+            "rate" in result.get("reason", "").lower() or "limit" in result.get("error", "").lower()
+        )
 
 
 class TestAzureGovCloud: