refactor: drain no_magic_numbers baseline to zero via Final hoists (#1856 phase 2)

src/synthorg/a2a/connection_types/a2a_peer.py

@@ -4,15 +4,19 @@
 verifier registration for the unified webhook receiver.
 """
 
+from typing import Final
+
 from synthorg.a2a.push_verifier import A2APushVerifier
 from synthorg.integrations.connections.models import ConnectionType
 from synthorg.integrations.webhooks.verifiers.protocol import (
     SignatureVerifier,  # noqa: TC001
 )
 
+_DEFAULT_CLOCK_SKEW_SECONDS: Final[int] = 300
+
 
 def get_a2a_push_verifier(
-    clock_skew_seconds: int = 300,
+    clock_skew_seconds: int = _DEFAULT_CLOCK_SKEW_SECONDS,
 ) -> SignatureVerifier:
     """Create an A2A push notification verifier.
 

src/synthorg/a2a/gateway.py

@@ -9,7 +9,7 @@
 import asyncio
 import hmac
 import json
-from typing import Any, ClassVar
+from typing import Any, ClassVar, Final
 
 from litestar import Controller, Request, post
 from litestar.datastructures import State  # noqa: TC002
@@ -57,6 +57,7 @@
 from synthorg.observability.events.settings import SETTINGS_FETCH_FAILED
 
 logger = get_logger(__name__)
+_DEFAULT_HTTP_STATUS: Final[int] = 400
 
 _SUPPORTED_METHODS = frozenset(
     {
@@ -72,7 +73,7 @@
 # ``a2a.max_message_parts``.  This constant mirrors that registry
 # default so a test harness or a boot path that bypasses
 # :class:`AppState` still enforces the documented ceiling.
-_MAX_MESSAGE_PARTS_FALLBACK = 100
+_MAX_MESSAGE_PARTS_FALLBACK: Final[int] = 100
 
 
 async def _resolve_max_message_parts(app_state: Any) -> int:
@@ -627,7 +628,7 @@ def __init__(
         code: int,
         message: str,
         *,
-        http_status: int = 400,
+        http_status: int = _DEFAULT_HTTP_STATUS,
     ) -> None:
         super().__init__(message)
         self.code = code

src/synthorg/a2a/push_verifier.py

@@ -8,6 +8,7 @@
 import hashlib
 import hmac
 import math
+from typing import Final
 
 from synthorg.core.clock import Clock, SystemClock
 from synthorg.observability import get_logger
@@ -19,7 +20,7 @@
 
 logger = get_logger(__name__)
 
-_DEFAULT_CLOCK_SKEW_SECONDS = 300
+_DEFAULT_CLOCK_SKEW_SECONDS: Final[int] = 300
 
 
 class A2APushVerifier:

src/synthorg/api/app.py

@@ -9,7 +9,7 @@
 import sys
 import time
 from pathlib import Path
-from typing import TYPE_CHECKING
+from typing import TYPE_CHECKING, Final
 
 from litestar import Controller, Litestar, Router
 from litestar.config.compression import CompressionConfig
@@ -142,9 +142,7 @@
 # Update both sites together if the default ever changes; otherwise a
 # bootstrap value will silently disagree with operator-editable
 # overrides resolved through ``ConfigResolver``.
-_DEFAULT_TIMEOUT_CHECK_INTERVAL_SECONDS = (
-    60.0  # lint-allow: magic-numbers -- bootstrap default mirrored by ConfigResolver
-)
+_DEFAULT_TIMEOUT_CHECK_INTERVAL_SECONDS: Final[float] = 60.0
 
 
 def _build_default_approval_timeout_scheduler(

src/synthorg/api/auth/middleware.py

@@ -3,7 +3,7 @@
 import hashlib
 import hmac as _hmac
 from datetime import UTC, datetime
-from typing import TYPE_CHECKING, Any
+from typing import TYPE_CHECKING, Any, Final
 
 import jwt
 from litestar.enums import ScopeType
@@ -45,7 +45,7 @@
 
 logger = get_logger(__name__)
 
-_BEARER_PARTS = 2
+_BEARER_PARTS: Final[int] = 2
 _DEFAULT_COOKIE_NAME = "session"
 
 

src/synthorg/api/auth/ticket_store.py

@@ -24,7 +24,7 @@
 import math
 import secrets
 import threading
-from typing import ClassVar
+from typing import ClassVar, Final
 
 from pydantic import BaseModel, ConfigDict
 
@@ -44,6 +44,7 @@
 )
 
 logger = get_logger(__name__)
+_DEFAULT_TTL_SECONDS: Final[float] = 30.0
 
 
 class TicketLimitExceededError(PerOperationRateLimitError):
@@ -108,7 +109,7 @@ class WsTicketStore:
 
     def __init__(
         self,
-        ttl_seconds: float = 30.0,
+        ttl_seconds: float = _DEFAULT_TTL_SECONDS,
         *,
         max_pending_per_user: int = _MAX_PENDING_PER_USER,
         clock: Clock | None = None,

src/synthorg/api/boundary.py

@@ -34,7 +34,7 @@
 from collections.abc import (
     Mapping,  # noqa: TC003 -- runtime-needed via annotation introspection
 )
-from typing import LiteralString, overload
+from typing import Final, LiteralString, overload
 
 from pydantic import BaseModel, TypeAdapter, ValidationError
 
@@ -43,7 +43,7 @@
 
 logger = get_logger(__name__)
 
-_MAX_LOGGED_LOCATIONS = 5
+_MAX_LOGGED_LOCATIONS: Final[int] = 5
 
 
 @overload

src/synthorg/api/bus_bridge.py

@@ -35,9 +35,9 @@
 _SUBSCRIBER_ID: Final[str] = "__api_bridge__"
 _POLL_TIMEOUT: Final[float] = 1.0
 """Fallback poll timeout used when no resolver is wired in."""
-_MAX_CONSECUTIVE_ERRORS: Final[int] = 30  # lint-allow: magic-numbers -- bootstrap
+_MAX_CONSECUTIVE_ERRORS: Final[int] = 30
 """Fallback error budget used when no resolver is wired in."""
-_STOP_DRAIN_TIMEOUT_SECONDS: Final[float] = 10.0  # lint-allow: magic-numbers -- boot
+_STOP_DRAIN_TIMEOUT_SECONDS: Final[float] = 10.0
 """Fallback hard deadline for the ``stop()`` drain.
 
 10 seconds is the standard graceful-shutdown grace period across

src/synthorg/api/controllers/activities.py

@@ -3,7 +3,7 @@
 import asyncio
 from datetime import UTC, datetime, timedelta
 from enum import IntEnum
-from typing import TYPE_CHECKING, Annotated, Any
+from typing import TYPE_CHECKING, Annotated, Any, Final
 
 if TYPE_CHECKING:
     from collections.abc import Awaitable
@@ -39,6 +39,7 @@
 from synthorg.tools.invocation_record import ToolInvocationRecord  # noqa: TC001
 
 logger = get_logger(__name__)
+_DEFAULT_LIMIT: Final[int] = 50
 
 
 # Degraded source names -- used in responses and tests.
@@ -315,7 +316,7 @@ async def list_activities(  # noqa: PLR0913
         request: Request[Any, Any, Any],
         state: State,
         cursor: CursorParam = None,
-        limit: CursorLimit = 50,
+        limit: CursorLimit = _DEFAULT_LIMIT,
         event_type: Annotated[
             ActivityEventType | None,
             Parameter(

src/synthorg/api/controllers/agent_identity_versions.py

@@ -1,7 +1,7 @@
 """Agent identity version history API -- list, get, diff, rollback."""
 
 import asyncio
-from typing import Annotated
+from typing import Annotated, Final
 
 from litestar import Controller, get, post
 from litestar.datastructures import State  # noqa: TC002
@@ -41,6 +41,7 @@
 from synthorg.versioning import VersionSnapshot
 
 logger = get_logger(__name__)
+_DEFAULT_LIMIT: Final[int] = 20
 
 SnapshotT = VersionSnapshot[AgentIdentity]
 
@@ -108,7 +109,7 @@ async def list_versions(
         state: State,
         agent_id: PathId,
         cursor: CursorParam = None,
-        limit: CursorLimit = 20,
+        limit: CursorLimit = _DEFAULT_LIMIT,
     ) -> PaginatedResponse[SnapshotT]:
         """List version history for an agent identity."""
         secret = state.app_state.cursor_secret

src/synthorg/api/controllers/agents.py

@@ -1,7 +1,7 @@
 """Agent configuration, performance, activity, history, and CRUD mutations."""
 
 import json
-from typing import Any, Self
+from typing import Any, Final, Self
 
 from litestar import Controller, Request, Response, delete, get, patch, post
 from litestar.datastructures import State  # noqa: TC002
@@ -59,11 +59,12 @@
 )
 
 logger = get_logger(__name__)
+_DEFAULT_LIMIT: Final[int] = 50
 
 # Safety cap for lifecycle event queries to prevent unbounded memory
 # allocation.  The paginate() helper already caps the returned page
 # to MAX_LIMIT, but the underlying fetch is uncapped without this.
-_MAX_LIFECYCLE_EVENTS = 10_000
+_MAX_LIFECYCLE_EVENTS: Final[int] = 10_000
 
 
 async def _resolve_agent_id(
@@ -187,7 +188,7 @@ async def list_agents(
         self,
         state: State,
         cursor: CursorParam = None,
-        limit: CursorLimit = 50,
+        limit: CursorLimit = _DEFAULT_LIMIT,
     ) -> PaginatedResponse[AgentConfig]:
         """List all configured agents.
 
@@ -423,7 +424,7 @@ async def get_agent_activity(
         state: State,
         agent_name: PathName,
         cursor: CursorParam = None,
-        limit: CursorLimit = 50,
+        limit: CursorLimit = _DEFAULT_LIMIT,
     ) -> PaginatedResponse[ActivityEvent]:
         """Get an agent's activity timeline (paginated).
 

src/synthorg/api/controllers/analytics.py

@@ -3,7 +3,7 @@
 import asyncio
 from collections import Counter
 from datetime import UTC, datetime, timedelta
-from typing import TYPE_CHECKING, Annotated, NamedTuple
+from typing import TYPE_CHECKING, Annotated, Final, NamedTuple
 
 from litestar import Controller, get
 from litestar.datastructures import State  # noqa: TC002
@@ -48,6 +48,7 @@
     from synthorg.hr.performance.models import TaskMetricRecord
 
 logger = get_logger(__name__)
+_DEFAULT_HORIZON_DAYS: Final[int] = 14
 
 
 # ── Response models ────────────────────────────────────────────
@@ -642,7 +643,7 @@ async def get_forecast(
                 le=90,
                 description="Projection horizon in days",
             ),
-        ] = 14,
+        ] = _DEFAULT_HORIZON_DAYS,
     ) -> ApiResponse[ForecastResponse]:
         """Return budget spend projection.
 

src/synthorg/api/controllers/approvals.py

@@ -4,7 +4,7 @@
 import math
 from datetime import UTC, datetime, timedelta
 from enum import StrEnum
-from typing import Annotated, Any
+from typing import Annotated, Any, Final
 from uuid import uuid4
 
 from litestar import Controller, Request, get, post
@@ -70,6 +70,7 @@
 from synthorg.settings.enums import SettingNamespace
 
 logger = get_logger(__name__)
+_DEFAULT_LIMIT: Final[int] = 50
 
 _URGENCY_CRITICAL_FALLBACK_SECONDS: float = 3600.0
 _URGENCY_HIGH_FALLBACK_SECONDS: float = 14400.0
@@ -516,7 +517,7 @@ async def list_approvals(  # noqa: PLR0913
         action_type: Annotated[str, Parameter(max_length=QUERY_MAX_LENGTH)]
         | None = None,
         cursor: CursorParam = None,
-        limit: CursorLimit = 50,
+        limit: CursorLimit = _DEFAULT_LIMIT,
     ) -> PaginatedResponse[ApprovalResponse]:
         """List approval items with optional filters.
 

src/synthorg/api/controllers/artifacts.py

@@ -1,6 +1,6 @@
 """Artifact controller -- endpoints for artifact management, storage, and retrieval."""
 
-from typing import Annotated, Any
+from typing import Annotated, Any, Final
 
 from litestar import Controller, Request, Response, delete, get, post, put
 from litestar.datastructures import State  # noqa: TC002
@@ -43,6 +43,7 @@
 )
 
 logger = get_logger(__name__)
+_DEFAULT_LIMIT: Final[int] = 50
 
 
 def _service(state: State) -> ArtifactService:
@@ -178,7 +179,7 @@ async def list_artifacts(  # noqa: PLR0913
         self,
         state: State,
         cursor: CursorParam = None,
-        limit: CursorLimit = 50,
+        limit: CursorLimit = _DEFAULT_LIMIT,
         task_id: TaskIdFilter = None,
         created_by: CreatedByFilter = None,
         type: TypeFilter = None,  # noqa: A002

src/synthorg/api/controllers/audit.py

@@ -10,7 +10,7 @@
 import asyncio
 import json
 from datetime import datetime  # noqa: TC003
-from typing import Annotated
+from typing import Annotated, Final
 
 from litestar import Controller, get
 from litestar.datastructures import State  # noqa: TC002
@@ -35,8 +35,9 @@
 from synthorg.settings.enums import SettingNamespace
 
 logger = get_logger(__name__)
+_DEFAULT_LIMIT: Final[int] = 50
 
-_MAX_AUDIT_QUERY = 10_000
+_MAX_AUDIT_QUERY: Final[int] = 10_000
 """Fallback cap applied when no settings resolver is wired in."""
 
 # Module-level log-once guard for the settings-resolution fallback:
@@ -101,7 +102,7 @@ async def list_audit_entries(  # noqa: PLR0913
         since: datetime | None = None,
         until: datetime | None = None,
         cursor: CursorParam = None,
-        limit: CursorLimit = 50,
+        limit: CursorLimit = _DEFAULT_LIMIT,
         jsonb_contains: Annotated[str, Parameter(max_length=2048)] | None = None,
         jsonb_key_exists: Annotated[str, Parameter(max_length=256)] | None = None,
     ) -> PaginatedResponse[AuditEntry]:

src/synthorg/api/controllers/backup.py

@@ -4,7 +4,7 @@
 (used by the CLI for ``synthorg backup`` / ``synthorg wipe``).
 """
 
-from typing import TYPE_CHECKING, Annotated
+from typing import TYPE_CHECKING, Annotated, Final
 
 if TYPE_CHECKING:
     from collections.abc import Awaitable, Callable
@@ -54,6 +54,7 @@
 from synthorg.observability.events.idempotency import IDEMPOTENCY_CLAIM_IN_FLIGHT
 
 logger = get_logger(__name__)
+_DEFAULT_LIMIT: Final[int] = 50
 
 
 async def _do_backup_as_dict(
@@ -184,7 +185,7 @@ async def list_backups(
         self,
         state: State,
         cursor: CursorParam = None,
-        limit: CursorLimit = 50,
+        limit: CursorLimit = _DEFAULT_LIMIT,
     ) -> PaginatedResponse[BackupInfo]:
         """List available backups (paginated, newest first).