diff --git a/src/transcription/app.py b/src/transcription/app.py
index ecd4198..02f5cdf 100644
--- a/src/transcription/app.py
+++ b/src/transcription/app.py
@@ -16,16 +16,9 @@ from .db import dispose_database_runtime
 from .db import initialize_database_runtime
 from .services import ServiceBundle
 from .ui import register_pages
-from .worker import WorkerNotifier
-from .worker import resolve_worker_notifier
 from .worker import worker_consumer_lifespan
 
 
-def get_worker_notifier(app: FastAPI) -> WorkerNotifier:
-    """Return app worker notifier, or a no-op fallback when unavailable."""
-    return resolve_worker_notifier(app.state)
-
-
 @asynccontextmanager
 async def _lifespan(app: FastAPI):
     configure_logging()
diff --git a/src/transcription/app_state.py b/src/transcription/app_state.py
new file mode 100644
index 0000000..110f641
--- /dev/null
+++ b/src/transcription/app_state.py
@@ -0,0 +1,39 @@
+"""Helpers for accessing lifespan-owned application state resources."""
+
+from __future__ import annotations
+
+from fastapi import FastAPI
+from sqlalchemy.ext.asyncio import async_sessionmaker
+from sqlmodel.ext.asyncio.session import AsyncSession
+
+from transcription.db.runtime import DatabaseRuntime
+from transcription.db.runtime import get_session_factory
+from transcription.worker import WorkerNotifier
+from transcription.worker import resolve_worker_notifier
+
+
+def resolve_database_runtime(state: object) -> DatabaseRuntime | None:
+    """Return database runtime from app-like state objects when available."""
+    runtime = getattr(state, "runtime", None)
+    return runtime if isinstance(runtime, DatabaseRuntime) else None
+
+
+def require_database_runtime(state: object) -> DatabaseRuntime:
+    """Return database runtime or raise when app lifespan has not initialized it."""
+    runtime = resolve_database_runtime(state)
+    if runtime is None:
+        raise RuntimeError("Database runtime is not initialized on application state")
+    return runtime
+
+
+def resolve_session_factory(state: object) -> async_sessionmaker[AsyncSession]:
+    """Return DB session factory from state when available, otherwise shared runtime."""
+    runtime = resolve_database_runtime(state)
+    if runtime is not None:
+        return runtime.session_factory
+    return get_session_factory()
+
+
+def get_worker_notifier(app: FastAPI) -> WorkerNotifier:
+    """Return app worker notifier, or a no-op fallback when unavailable."""
+    return resolve_worker_notifier(app.state)
diff --git a/src/transcription/ui/pages/upload_page.py b/src/transcription/ui/pages/upload_page.py
index 1b94622..4f3109b 100644
--- a/src/transcription/ui/pages/upload_page.py
+++ b/src/transcription/ui/pages/upload_page.py
@@ -5,6 +5,7 @@ from __future__ import annotations
 from fastapi import Request
 from nicegui import ui
 
+from transcription.app_state import resolve_session_factory
 from transcription.db import get_session
 from transcription.services.store import create_upload_job
 from transcription.ui.components.upload import render_upload_widget
@@ -16,8 +17,10 @@ def register_page() -> None:
 
     @ui.page("/upload", title="Upload Document")
     def upload_page(request: Request) -> None:
+        session_factory = resolve_session_factory(request.app.state)
+
         async def submit_upload(filename: str, file_bytes: bytes):
-            async with get_session(session_factory=request.app.state.runtime.session_factory) as session:
+            async with get_session(session_factory=session_factory) as session:
                 return await create_upload_job(
                     filename=filename,
                     file_bytes=file_bytes,