Separate bootstrap settings from runtime overrides with a dedicated runtime settings manager

backend/app/dependencies.py

@@ -20,7 +20,7 @@ from app.config import Settings
 from app.models.auth import Session
 from app.models.config import PendingRecovery
 from app.models.server import ServerStatus
-from app.utils.runtime_state import RuntimeState
+from app.utils.runtime_state import RuntimeState, get_effective_settings
 from app.utils.session_cache import SessionCache
 
 log: structlog.stdlib.BoundLogger = structlog.get_logger()
@@ -35,6 +35,7 @@ class AppState(Protocol):
     server_status: ServerStatus
     pending_recovery: PendingRecovery | None
     last_activation: dict[str, datetime.datetime] | None
+    runtime_settings: Settings | None
     runtime_state: RuntimeState
     session_cache: SessionCache
 
@@ -90,16 +91,8 @@ async def get_db(request: Request) -> AsyncGenerator[aiosqlite.Connection, None]
 
 
 async def get_settings(request: Request) -> Settings:
-    """Provide the :class:`~app.config.Settings` instance from ``app.state``.
-
-    Args:
-        request: The current FastAPI request (injected automatically).
-
-    Returns:
-        The application settings loaded at startup.
-    """
-    state = cast("AppState", request.app.state)
-    return state.settings
+    """Provide the effective application settings for the current request."""
+    return get_effective_settings(request.app)
 
 
 async def get_http_session(request: Request) -> aiohttp.ClientSession:

backend/app/startup.py

@@ -20,6 +20,7 @@ from app.db import init_db, open_db
 from app.services import geo_service, setup_service
 from app.tasks import blocklist_import, geo_cache_flush, geo_re_resolve, health_check, history_sync
 from app.utils.jail_config import ensure_jail_configs
+from app.utils.runtime_state import set_runtime_settings
 from app.utils.setup_state import set_setup_complete_cache
 
 log: structlog.stdlib.BoundLogger = structlog.get_logger()
@@ -88,7 +89,8 @@ async def startup_shared_resources(
                 updated_settings = settings.model_copy(update=persisted_runtime_settings)
                 if Path(updated_settings.database_path).resolve() != original_db_path:
                     await _ensure_database_schema(updated_settings.database_path)
-                app.state.settings = updated_settings
+                set_runtime_settings(app, updated_settings)
+                settings = updated_settings
                 log.info(
                     "runtime_settings_overridden_from_setup",
                     overrides=persisted_runtime_settings,

backend/app/tasks/blocklist_import.py

@@ -21,6 +21,7 @@ import structlog
 from app.db import open_db
 from app.models.blocklist import ScheduleFrequency
 from app.services import blocklist_service
+from app.utils.runtime_state import get_effective_settings
 
 if TYPE_CHECKING:
     import aiosqlite
@@ -35,7 +36,8 @@ JOB_ID: str = "blocklist_import"
 
 
 async def _get_db(app: Any) -> tuple[aiosqlite.Connection, bool]:
-    db = await open_db(app.state.settings.database_path)
+    settings = get_effective_settings(app)
+    db = await open_db(settings.database_path)
     return db, True
 
 
@@ -50,8 +52,9 @@ async def _run_import(app: Any) -> None:
             APScheduler ``kwargs``.
     """
     db, close_db = await _get_db(app)
+    settings = get_effective_settings(app)
     http_session = app.state.http_session
-    socket_path: str = app.state.settings.fail2ban_socket
+    socket_path: str = settings.fail2ban_socket
 
     log.info("blocklist_import_starting")
     try:

backend/app/tasks/geo_cache_flush.py

@@ -16,6 +16,7 @@ from typing import TYPE_CHECKING, Any
 import structlog
 
 from app.db import open_db
+from app.utils.runtime_state import get_effective_settings
 
 if TYPE_CHECKING:
     import aiosqlite
@@ -34,7 +35,8 @@ JOB_ID: str = "geo_cache_flush"
 
 
 async def _get_db(app: Any) -> tuple[aiosqlite.Connection, bool]:
-    db = await open_db(app.state.settings.database_path)
+    settings = get_effective_settings(app)
+    db = await open_db(settings.database_path)
     return db, True
 
 

backend/app/tasks/geo_re_resolve.py

@@ -22,6 +22,7 @@ from typing import TYPE_CHECKING
 import structlog
 
 from app.db import open_db
+from app.utils.runtime_state import get_effective_settings
 
 if TYPE_CHECKING:
     import aiosqlite
@@ -40,7 +41,8 @@ JOB_ID: str = "geo_re_resolve"
 
 
 async def _get_db(app: FastAPI) -> tuple[aiosqlite.Connection, bool]:
-    db = await open_db(app.state.settings.database_path)
+    settings = get_effective_settings(app)
+    db = await open_db(settings.database_path)
     return db, True
 
 

backend/app/tasks/health_check.py

@@ -25,6 +25,7 @@ import structlog
 from app.models.config import PendingRecovery
 from app.models.server import ServerStatus
 from app.services import health_service
+from app.utils.runtime_state import get_effective_settings
 
 if TYPE_CHECKING:  # pragma: no cover
     from fastapi import FastAPI
@@ -56,14 +57,15 @@ async def _run_probe(app: FastAPI) -> None:
     ``app.state.pending_recovery``.
 
     This is the APScheduler job callback.  It reads ``fail2ban_socket`` from
-    ``app.state.settings``, runs the health probe, and writes the result to
-    ``app.state.server_status``.
+    the effective runtime settings, runs the health probe, and writes the
+    result to ``app.state.server_status``.
 
     Args:
         app: The :class:`fastapi.FastAPI` application instance passed by the
             scheduler via the ``kwargs`` mechanism.
     """
-    socket_path: str = app.state.settings.fail2ban_socket
+    settings = get_effective_settings(app)
+    socket_path: str = settings.fail2ban_socket
     prev_status: ServerStatus = getattr(
         app.state, "server_status", ServerStatus(online=False)
     )

backend/app/tasks/history_sync.py

@@ -17,6 +17,7 @@ import structlog
 from app.db import open_db
 from app.repositories import fail2ban_db_repo
 from app.utils.fail2ban_db_utils import get_fail2ban_db_path
+from app.utils.runtime_state import get_effective_settings
 
 if TYPE_CHECKING:  # pragma: no cover
     from fastapi import FastAPI
@@ -34,7 +35,8 @@ BACKFILL_WINDOW: int = 648000
 
 
 async def _get_db(app: FastAPI) -> tuple[aiosqlite.Connection, bool]:
-    db = await open_db(app.state.settings.database_path)
+    settings = get_effective_settings(app)
+    db = await open_db(settings.database_path)
     return db, True
 
 
@@ -47,7 +49,8 @@ async def _get_last_archive_ts(db) -> int | None:
 
 
 async def _run_sync(app: FastAPI) -> None:
-    socket_path: str = app.state.settings.fail2ban_socket
+    settings = get_effective_settings(app)
+    socket_path: str = settings.fail2ban_socket
     db, close_db = await _get_db(app)
 
     try:

backend/app/utils/runtime_state.py

@@ -29,6 +29,7 @@ _RUNTIME_ATTRIBUTES: frozenset[str] = frozenset(
         "server_status",
         "pending_recovery",
         "last_activation",
+        "runtime_settings",
     }
 )
 
@@ -41,6 +42,7 @@ class RuntimeState:
     server_status: ServerStatus = field(default_factory=lambda: ServerStatus(online=False))
     pending_recovery: PendingRecovery | None = None
     last_activation: ActivationRecord | None = None
+    runtime_settings: "Settings" | None = None
 
 
 class ApplicationState(State):
@@ -87,17 +89,32 @@ def get_runtime_state(app: Any) -> RuntimeState:
 
 
 def get_app_settings(app: Any) -> Settings:
-    """Return the current immutable settings from application state."""
+    """Return the bootstrap settings loaded at startup."""
     settings = getattr(app.state, "settings", None)
     if settings is None:
         raise AttributeError("Application settings are not available on the app state.")
     return settings
 
 
+def get_effective_settings(app: Any) -> Settings:
+    """Return the effective settings for the current application instance."""
+    runtime_settings = getattr(app.state, "runtime_settings", None)
+    if runtime_settings is not None:
+        return runtime_settings
+    return get_app_settings(app)
+
+
+def set_runtime_settings(app: Any, settings: Settings) -> None:
+    """Store the resolved runtime settings separately from bootstrap config."""
+    runtime_state = get_runtime_state(app)
+    runtime_state.runtime_settings = settings
+
+
 def update_app_settings(app: Any, **overrides: Any) -> None:
-    """Update the current application settings immutably."""
+    """Update the current effective settings immutably."""
     settings = get_app_settings(app)
-    app.state.settings = settings.model_copy(update=overrides)
+    updated = settings.model_copy(update=overrides)
+    set_runtime_settings(app, updated)
 
 
 def record_activation(app: Any, jail_name: str, at: datetime.datetime | None = None) -> datetime.datetime:

backend/tests/test_main.py

@@ -288,10 +288,12 @@ async def test_startup_overrides_settings_from_persisted_setup(tmp_path: Path) -
         patch("app.tasks.history_sync.register"),
     ):
         async with _lifespan(app):
-            assert app.state.settings.database_path == runtime_db_path
-            assert app.state.settings.fail2ban_socket == "/tmp/persisted.sock"
-            assert app.state.settings.timezone == "Europe/Berlin"
-            assert app.state.settings.session_duration_minutes == 123
+            assert app.state.runtime_settings is not None
+            assert app.state.runtime_settings.database_path == runtime_db_path
+            assert app.state.runtime_settings.fail2ban_socket == "/tmp/persisted.sock"
+            assert app.state.runtime_settings.timezone == "Europe/Berlin"
+            assert app.state.runtime_settings.session_duration_minutes == 123
+            assert app.state.settings.database_path == str(tmp_path / "pointer.db")
             assert Path(runtime_db_path).exists()
 
 

backend/tests/test_routers/test_setup.py

@@ -159,10 +159,12 @@ class TestPostSetupRuntimeState:
 
         response = await client.post("/api/setup", json=payload)
         assert response.status_code == 201
-        assert app.state.settings.database_path == payload["database_path"]
-        assert app.state.settings.fail2ban_socket == payload["fail2ban_socket"]
-        assert app.state.settings.timezone == payload["timezone"]
-        assert app.state.settings.session_duration_minutes == payload["session_duration_minutes"]
+        assert app.state.runtime_settings is not None
+        assert app.state.runtime_settings.database_path == payload["database_path"]
+        assert app.state.runtime_settings.fail2ban_socket == payload["fail2ban_socket"]
+        assert app.state.runtime_settings.timezone == payload["timezone"]
+        assert app.state.runtime_settings.session_duration_minutes == payload["session_duration_minutes"]
+        assert app.state.settings.database_path != payload["database_path"]
 
 
 class TestSetupRedirectMiddleware: