fix(#365): fix ruff, mypy, pip, and bandit issues in coordinator

- Fix 20 ruff errors: UP035 (Callable import), UP042 (StrEnum), E501
  (line length), F401 (unused imports), UP045 (Optional -> X | None),
  I001 (import sorting)
- Fix mypy error: wrap slowapi rate limit handler with
  Exception-compatible signature for add_exception_handler
- Pin pip >= 25.3 in Dockerfile (CVE-2025-8869, CVE-2026-1703)
- Add nosec B104 to config.py (container-bound 0.0.0.0 is acceptable)
- Add nosec B101 to telemetry.py (assert for type narrowing)
- Create bandit.yaml to suppress B404/B607/B603 in gates/ tooling

Fixes #365

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

apps/coordinator/Dockerfile

@@ -16,7 +16,7 @@ COPY pyproject.toml .
 RUN python -m venv /opt/venv
 ENV PATH="/opt/venv/bin:$PATH"
 COPY src/ ./src/
-RUN pip install --no-cache-dir --upgrade pip && \
+RUN pip install --no-cache-dir "pip>=25.3" && \
     pip install --no-cache-dir .
 
 # Production stage

apps/coordinator/bandit.yaml

@@ -0,0 +1,23 @@
+# Bandit security linting configuration for mosaic-coordinator
+#
+# Suppressions documented below. All are intentional and reviewed.
+#
+# B104 (bind to 0.0.0.0): Inline nosec in src/config.py.
+#       Container-bound service — must listen on all interfaces inside Docker.
+#
+# B101 (assert usage): Inline nosec in src/telemetry.py.
+#       Assert used for type narrowing after None guard (satisfies mypy).
+#
+# B404, B607, B603 (subprocess usage): Skipped globally.
+#       Only triggered in src/gates/ quality gate tooling, which intentionally
+#       invokes external tools (pytest, ruff, mypy) via subprocess as its
+#       core functionality. No other source files use subprocess.
+
+skips:
+  - B404 # import subprocess — only in gates/ (intentional)
+  - B607 # start process with partial path — only in gates/ (intentional)
+  - B603 # subprocess call without shell=True — only in gates/ (intentional)
+
+exclude_dirs:
+  - tests
+  - venv

apps/coordinator/src/circuit_breaker.py

@@ -13,13 +13,14 @@ Reference: SEC-ORCH-7 from security review
 
 import logging
 import time
-from enum import Enum
+from collections.abc import Callable
-from typing import Any, Callable
+from enum import StrEnum
+from typing import Any
 
 logger = logging.getLogger(__name__)
 
 
-class CircuitState(str, Enum):
+class CircuitState(StrEnum):
     """States for the circuit breaker."""
 
     CLOSED = "closed"  # Normal operation

apps/coordinator/src/config.py

@@ -21,7 +21,7 @@ class Settings(BaseSettings):
     anthropic_api_key: str
 
     # Server Configuration
-    host: str = "0.0.0.0"
+    host: str = "0.0.0.0"  # nosec B104 — Container-bound: listen on all interfaces inside Docker
     port: int = 8000
 
     # Logging

apps/coordinator/src/context_monitor.py

@@ -192,7 +192,8 @@ class ContextMonitor:
                 logger.error(
                     f"Error monitoring agent {agent_id}: {e} "
                     f"(circuit breaker: {circuit_breaker.state.value}, "
-                    f"failures: {circuit_breaker.failure_count}/{circuit_breaker.failure_threshold})"
+                    f"failures: {circuit_breaker.failure_count}"
+                    f"/{circuit_breaker.failure_threshold})"
                 )
 
             # Wait for next poll (or until stopped)

apps/coordinator/src/coordinator.py

@@ -4,7 +4,7 @@ import asyncio
 import logging
 from typing import TYPE_CHECKING, Any
 
-from src.circuit_breaker import CircuitBreaker, CircuitBreakerError, CircuitState
+from src.circuit_breaker import CircuitBreaker, CircuitBreakerError
 from src.context_monitor import ContextMonitor
 from src.forced_continuation import ForcedContinuationService
 from src.models import ContextAction
@@ -142,7 +142,8 @@ class Coordinator:
                     logger.error(
                         f"Error in process_queue: {e} "
                         f"(circuit breaker: {self._circuit_breaker.state.value}, "
-                        f"failures: {self._circuit_breaker.failure_count}/{self._circuit_breaker.failure_threshold})"
+                        f"failures: {self._circuit_breaker.failure_count}"
+                        f"/{self._circuit_breaker.failure_threshold})"
                     )
 
                 # Wait for poll interval or stop signal
@@ -432,7 +433,8 @@ class OrchestrationLoop:
                     logger.error(
                         f"Error in process_next_issue: {e} "
                         f"(circuit breaker: {self._circuit_breaker.state.value}, "
-                        f"failures: {self._circuit_breaker.failure_count}/{self._circuit_breaker.failure_threshold})"
+                        f"failures: {self._circuit_breaker.failure_count}"
+                        f"/{self._circuit_breaker.failure_threshold})"
                     )
 
                 # Wait for poll interval or stop signal

apps/coordinator/src/gates/coverage_gate.py

@@ -1,7 +1,6 @@
 """CoverageGate - Enforces 85% minimum test coverage via pytest-cov."""
 
 import json
-import os
 import subprocess
 from pathlib import Path
 

apps/coordinator/src/main.py

@@ -8,11 +8,13 @@ from contextlib import asynccontextmanager
 from pathlib import Path
 from typing import Any
 
-from fastapi import FastAPI, Request
+from fastapi import FastAPI
 from pydantic import BaseModel
 from slowapi import Limiter, _rate_limit_exceeded_handler
 from slowapi.errors import RateLimitExceeded
 from slowapi.util import get_remote_address
+from starlette.requests import Request
+from starlette.responses import Response
 
 from .config import settings
 from .coordinator import Coordinator
@@ -141,7 +143,16 @@ if os.getenv("OTEL_ENABLED", "true").lower() != "false":
 
 # Register rate limiter
 app.state.limiter = limiter
-app.add_exception_handler(RateLimitExceeded, _rate_limit_exceeded_handler)
+
+
+def _rate_limit_handler(request: Request, exc: Exception) -> Response:
+    """Wrapper for slowapi handler with Exception-compatible signature."""
+    if not isinstance(exc, RateLimitExceeded):
+        return Response(content="Rate limit error", status_code=429)
+    return _rate_limit_exceeded_handler(request, exc)
+
+
+app.add_exception_handler(RateLimitExceeded, _rate_limit_handler)
 
 
 class HealthResponse(BaseModel):

apps/coordinator/src/models.py

@@ -1,12 +1,12 @@
 """Data models for mosaic-coordinator."""
 
-from enum import Enum
+from enum import StrEnum
 from typing import Literal
 
 from pydantic import BaseModel, Field, field_validator
 
 
-class Capability(str, Enum):
+class Capability(StrEnum):
     """Agent capability levels."""
 
     HIGH = "high"
@@ -14,7 +14,7 @@ class Capability(str, Enum):
     LOW = "low"
 
 
-class AgentName(str, Enum):
+class AgentName(StrEnum):
     """Available AI agents."""
 
     OPUS = "opus"
@@ -24,7 +24,7 @@ class AgentName(str, Enum):
     MINIMAX = "minimax"
 
 
-class ContextAction(str, Enum):
+class ContextAction(StrEnum):
     """Actions to take based on context usage thresholds."""
 
     CONTINUE = "continue"  # Below compact threshold, keep working

apps/coordinator/src/queue.py

@@ -5,7 +5,7 @@ import logging
 import shutil
 from dataclasses import dataclass, field
 from datetime import datetime
-from enum import Enum
+from enum import StrEnum
 from pathlib import Path
 from typing import Any
 
@@ -14,7 +14,7 @@ from src.models import IssueMetadata
 logger = logging.getLogger(__name__)
 
 
-class QueueItemStatus(str, Enum):
+class QueueItemStatus(StrEnum):
     """Status of a queue item."""
 
     PENDING = "pending"

apps/coordinator/src/security.py

@@ -4,7 +4,6 @@ import hashlib
 import hmac
 import logging
 import re
-from typing import Optional
 
 logger = logging.getLogger(__name__)
 
@@ -33,11 +32,14 @@ INJECTION_PATTERNS = [
 ]
 
 # XML-like tags that could be used for injection
-DANGEROUS_TAG_PATTERN = re.compile(r"<\s*(instructions?|prompt|context|system|user|assistant)\s*>", re.IGNORECASE)
+DANGEROUS_TAG_PATTERN = re.compile(
+    r"<\s*(instructions?|prompt|context|system|user|assistant)\s*>",
+    re.IGNORECASE,
+)
 
 
 def sanitize_for_prompt(
-    content: Optional[str],
+    content: str | None,
     max_length: int = DEFAULT_MAX_PROMPT_LENGTH
 ) -> str:
     """

apps/coordinator/src/telemetry.py

@@ -139,7 +139,7 @@ class TelemetryService:
         if self._tracer is None:
             # Initialize if not already done
             self.initialize()
-        assert self._tracer is not None
+        assert self._tracer is not None  # nosec B101 — Type narrowing after None guard
         return self._tracer
 
     def shutdown(self) -> None:

apps/coordinator/tests/test_telemetry.py

@@ -1,7 +1,9 @@
 """Tests for OpenTelemetry telemetry initialization."""
 
+from unittest.mock import MagicMock, patch
+
 import pytest
-from unittest.mock import MagicMock, patch, ANY
+
 from src.telemetry import TelemetryService, get_tracer
 
 
@@ -171,7 +173,10 @@ class TestGetTracer:
         self, mock_set_provider: MagicMock, mock_get_tracer_func: MagicMock, reset_telemetry
     ) -> None:
         """Test that get_tracer uses the correct service name."""
-        with patch.dict("os.environ", {"OTEL_SERVICE_NAME": "test-service", "OTEL_ENABLED": "true"}):
+        with patch.dict(
+            "os.environ",
+            {"OTEL_SERVICE_NAME": "test-service", "OTEL_ENABLED": "true"},
+        ):
             # Reset global state
             import src.telemetry
             src.telemetry._telemetry_service = None

apps/coordinator/tests/test_tracing_decorators.py

@@ -1,8 +1,10 @@
 """Tests for tracing decorators."""
 
+from unittest.mock import MagicMock, patch
+
 import pytest
-from unittest.mock import MagicMock, patch, AsyncMock
 from opentelemetry.trace import SpanKind
+
 from src.tracing_decorators import trace_agent_operation, trace_tool_execution
 
 
@@ -130,7 +132,9 @@ class TestTraceToolExecution:
             result = await test_func(param="value")
 
             assert result == "result-value"
-            mock_tracer.start_as_current_span.assert_called_once_with("tool.test_tool", kind=SpanKind.CLIENT)
+            mock_tracer.start_as_current_span.assert_called_once_with(
+                "tool.test_tool", kind=SpanKind.CLIENT
+            )
             mock_span.set_attribute.assert_any_call("tool.name", "test_tool")
 
     @pytest.mark.asyncio