diff --git a/.env.example b/.env.example
index 1d587aa..9676668 100644
--- a/.env.example
+++ b/.env.example
@@ -24,6 +24,12 @@ JWT_SECRET=change-this-development-secret
 # ============================================================================
 INITIAL_ADMIN_PASSWORD=admin12345
 
+# Backend loguru stderr sink level. One of DEBUG / INFO / WARNING / ERROR
+# / CRITICAL. Anything else (e.g. lowercase) falls back to INFO inside
+# configure_logging(). Change to DEBUG to see request bodies in
+# runtime_client._jupyter_request.
+LOG_LEVEL=INFO
+
 # Object storage. Two modes are supported:
 #   STORAGE_BACKEND=s3    — connects to an S3-compatible service (MinIO,
 #                           RustFS, SeaweedFS, AWS S3, …). Requires the
diff --git a/backend/src/backend/main.py b/backend/src/backend/main.py
index 35bb910..70a32fd 100644
--- a/backend/src/backend/main.py
+++ b/backend/src/backend/main.py
@@ -1,32 +1,40 @@
 from __future__ import annotations
 
+import time
+from collections.abc import AsyncIterator
 from contextlib import asynccontextmanager
-from typing import Any, AsyncIterator
+from typing import Any
 
 import httpx
-
 from common.config import settings
 from common.db import create_database_engine, create_session_factory
+from common.logging import configure_logging
 from common.service_app import create_service_app
 from common.storage import (
-    AsyncStorageBackend,
     PURPOSE_BUCKETS,
+    AsyncStorageBackend,
     actual_bucket_name,
     build_storage_config,
     create_storage,
 )
+from fastapi import Request
+from fastapi.responses import JSONResponse
+from loguru import logger
+
 from backend.admin import router as admin_router
 from backend.auth import router as auth_router
-from backend.platform import router as platform_router
 from backend.jupyter import router as jupyter_router
+from backend.platform import router as platform_router
+from backend.rclone_rc_client import RcloneRCClient
 from backend.resources import router as resources_router
 from backend.runtime_client import RuntimeClient
-from backend.rclone_rc_client import RcloneRCClient
 from backend.schedule_runs import router as schedule_runs_router
 from backend.schedules import router as schedules_router
 from backend.scripts import router as scripts_router
 from backend.storage_api import router as storage_api_router
 
+configure_logging(settings.log_level)
+
 
 @asynccontextmanager
 async def lifespan(app: Any) -> AsyncIterator[None]:
@@ -39,7 +47,7 @@ async def lifespan(app: Any) -> AsyncIterator[None]:
     # bucket name (e.g. "versions"), matching ``UploadSessions.bucket_name``
     # and ``StorageObjects.bucket_name`` so call sites can do
     # ``object_stores[upload.bucket_name].put(...)`` directly.
-    app.state.object_stores: dict[str, AsyncStorageBackend] = {
+    app.state.object_stores: dict[str, StorageBackend | AsyncStorageBackend] = {  # noqa: F821
         actual_bucket_name(purpose): create_storage(build_storage_config(purpose))
         for purpose in PURPOSE_BUCKETS
     }
@@ -78,3 +86,38 @@ app.include_router(platform_router)
 
 # Reuse the proven storage endpoints without running another FastAPI service.
 app.include_router(storage_api_router, prefix="/internal")
+
+
+@app.middleware("http")
+async def access_log(request: Request, call_next):
+    start = time.perf_counter()
+    try:
+        response = await call_next(request)
+    except Exception:
+        elapsed_ms = (time.perf_counter() - start) * 1000
+        logger.exception(
+            "request failed {method} {path} after {ms:.1f}ms",
+            method=request.method, path=request.url.path, ms=elapsed_ms,
+        )
+        raise
+    elapsed_ms = (time.perf_counter() - start) * 1000
+    logger.info(
+        "{method} {path} -> {status} in {ms:.1f}ms",
+        method=request.method, path=request.url.path,
+        status=response.status_code, ms=elapsed_ms,
+    )
+    return response
+
+
+@app.exception_handler(Exception)
+async def unhandled_exception_handler(request: Request, exc: Exception):
+    logger.error(
+        "unhandled exception on {method} {path} from {client}: {exc!r}",
+        method=request.method, path=request.url.path,
+        client=request.client.host if request.client else "-",
+        exc=exc,
+    )
+    return JSONResponse(
+        status_code=500,
+        content={"detail": "Internal server error"},
+    )
diff --git a/backend/tests/__init__.py b/backend/tests/__init__.py
new file mode 100644
index 0000000..e69de29
diff --git a/backend/tests/test_runtime_client_directories.py b/backend/tests/test_runtime_client_directories.py
new file mode 100644
index 0000000..7601277
--- /dev/null
+++ b/backend/tests/test_runtime_client_directories.py
@@ -0,0 +1,275 @@
+"""Unit tests for the three new directory methods on RuntimeClient.
+
+Covers:
+- `create_directory` — PUT with `{"type": "directory"}` body.
+- `delete_directory` — DELETE, surfaces Jupyter's 409 on non-empty dirs.
+- `ensure_directory` — GET-first, falls back to `create_directory` on 404.
+
+Uses `respx` to mock httpx transport so we don't need a live Jupyter.
+The runtime descriptor (`get_workspace`) is patched to a synchronous return
+so `_ensure_workspace` short-circuits without hitting the Runtime service.
+"""
+
+from __future__ import annotations
+
+import httpx
+import pytest
+import respx
+
+from backend.runtime_client import RuntimeClient, RuntimeClientError
+
+WORKSPACE_ID = "01HWS0000000000000000000A"
+BASE_URL = "http://runtime"
+PORT = 34567
+TOKEN = "test-token"
+JUPYTER_URL = f"{BASE_URL}:{PORT}/jupyter/{WORKSPACE_ID}/api/contents"
+
+
+def _running_descriptor() -> dict:
+    return {
+        "status": "running",
+        "workspace_id": WORKSPACE_ID,
+        "base_url": BASE_URL,
+        "port": PORT,
+        "token": TOKEN,
+    }
+
+
+@pytest.fixture
+def client() -> httpx.AsyncClient:
+    return httpx.AsyncClient(timeout=httpx.Timeout(5.0))
+
+
+@pytest.fixture
+def runtime(client: httpx.AsyncClient) -> RuntimeClient:
+    rt = RuntimeClient(client)
+    # Bypass the real `_ensure_workspace` so tests don't have to mock the
+    # Runtime service. The descriptor is otherwise identical to what the
+    # production path returns.
+    rt._ensure_workspace = _ensure_workspace_stub  # type: ignore[assignment]
+    return rt
+
+
+async def _ensure_workspace_stub(workspace_id: str) -> dict:
+    return _running_descriptor()
+
+
+# ---------------------------------------------------------------------------
+# create_directory
+# ---------------------------------------------------------------------------
+
+
+async def test_create_directory_happy_path(runtime: RuntimeClient) -> None:
+    with respx.mock(assert_all_called=True) as router:
+        route = router.put(f"{JUPYTER_URL}/01DIRAAAAAAAAAAAAAAA").mock(
+            return_value=httpx.Response(
+                201,
+                json={
+                    "name": "01DIRAAAAAAAAAAAAAAA",
+                    "type": "directory",
+                    "path": "01DIRAAAAAAAAAAAAAAA",
+                },
+            )
+        )
+
+        async with httpx.AsyncClient() as transport:
+            runtime.client = transport  # type: ignore[assignment]
+            result = await runtime.create_directory(
+                WORKSPACE_ID, name="01DIRAAAAAAAAAAAAAAA"
+            )
+
+    assert route.called
+    assert result["type"] == "directory"
+    # Verify request body shape (PUT contents/ directory).
+    request = route.calls[0].request
+    assert request.headers["Authorization"] == f"token {TOKEN}"
+    assert request.headers["Content-Type"] == "application/json"
+    assert request.content == b'{"type":"directory"}'
+
+
+async def test_create_directory_nested_path(runtime: RuntimeClient) -> None:
+    """Nested path `{parent_ulid}/{dir_ulid}` lands on Jupyter correctly."""
+    nested = "01DIR_PARENT_ULID/01DIR_CHILD_ULID"
+    with respx.mock(assert_all_called=True) as router:
+        route = router.put(f"{JUPYTER_URL}/{nested}").mock(
+            return_value=httpx.Response(
+                201, json={"name": "01DIR_CHILD_ULID", "type": "directory"}
+            )
+        )
+
+        async with httpx.AsyncClient() as transport:
+            runtime.client = transport  # type: ignore[assignment]
+            await runtime.create_directory(WORKSPACE_ID, name=nested)
+
+    assert route.called
+
+
+async def test_create_directory_propagates_jupyter_4xx(
+    runtime: RuntimeClient,
+) -> None:
+    with respx.mock() as router:
+        put_route = router.put(f"{JUPYTER_URL}/01DIRAAAAAAAAAAAAAAA").mock(
+            return_value=httpx.Response(
+                400,
+                json={
+                    "detail": {
+                        "code": "BAD_REQUEST",
+                        "message": "invalid name",
+                    }
+                },
+            )
+        )
+
+        async with httpx.AsyncClient() as transport:
+            runtime.client = transport  # type: ignore[assignment]
+            with pytest.raises(RuntimeClientError) as exc_info:
+                await runtime.create_directory(
+                    WORKSPACE_ID, name="01DIRAAAAAAAAAAAAAAA"
+                )
+
+    assert put_route.called
+    assert exc_info.value.status_code == 400
+    # `_jupyter_request` surfaces the whole JSON body as `detail` — the
+    # inner `detail` envelope is preserved verbatim.
+    assert exc_info.value.detail == {
+        "detail": {
+            "code": "BAD_REQUEST",
+            "message": "invalid name",
+        }
+    }
+
+
+# ---------------------------------------------------------------------------
+# delete_directory
+# ---------------------------------------------------------------------------
+
+
+async def test_delete_directory_happy_path(runtime: RuntimeClient) -> None:
+    with respx.mock(assert_all_called=True) as router:
+        route = router.delete(f"{JUPYTER_URL}/01DIRAAAAAAAAAAAAAAA").mock(
+            return_value=httpx.Response(204)
+        )
+
+        async with httpx.AsyncClient() as transport:
+            runtime.client = transport  # type: ignore[assignment]
+            result = await runtime.delete_directory(
+                WORKSPACE_ID, name="01DIRAAAAAAAAAAAAAAA"
+            )
+
+    assert route.called
+    assert result is None
+
+
+async def test_delete_directory_non_empty_409(runtime: RuntimeClient) -> None:
+    """Jupyter rejects non-empty directory deletes with 409; surface as-is."""
+    with respx.mock() as router:
+        router.delete(f"{JUPYTER_URL}/01DIRAAAAAAAAAAAAAAA").mock(
+            return_value=httpx.Response(
+                409,
+                json={
+                    "detail": {
+                        "code": "DIRECTORY_NOT_EMPTY",
+                        "message": "Directory is not empty",
+                    }
+                },
+            )
+        )
+
+        async with httpx.AsyncClient() as transport:
+            runtime.client = transport  # type: ignore[assignment]
+            with pytest.raises(RuntimeClientError) as exc_info:
+                await runtime.delete_directory(
+                    WORKSPACE_ID, name="01DIRAAAAAAAAAAAAAAA"
+                )
+
+    assert exc_info.value.status_code == 409
+
+
+# ---------------------------------------------------------------------------
+# ensure_directory
+# ---------------------------------------------------------------------------
+
+
+async def test_ensure_directory_already_exists(runtime: RuntimeClient) -> None:
+    """GET succeeds → no PUT. The lazy-backfill is a no-op."""
+    put_called = False
+
+    def _track_put(request: httpx.Request) -> httpx.Response:
+        nonlocal put_called
+        put_called = True
+        return httpx.Response(201, json={})
+
+    with respx.mock(assert_all_called=False) as router:
+        router.get(f"{JUPYTER_URL}/01DIRAAAAAAAAAAAAAAA").mock(
+            return_value=httpx.Response(
+                200,
+                json={"name": "01DIRAAAAAAAAAAAAAAA", "type": "directory"},
+            )
+        )
+        router.put(f"{JUPYTER_URL}/01DIRAAAAAAAAAAAAAAA").mock(
+            side_effect=_track_put
+        )
+
+        async with httpx.AsyncClient() as transport:
+            runtime.client = transport  # type: ignore[assignment]
+            await runtime.ensure_directory(
+                WORKSPACE_ID, "01DIRAAAAAAAAAAAAAAA"
+            )
+
+    assert not put_called, "PUT must not be issued when GET already shows the dir exists"
+
+
+async def test_ensure_directory_missing_creates_it(
+    runtime: RuntimeClient,
+) -> None:
+    """GET 404 → PUT. The lazy-backfill creates the missing directory."""
+    with respx.mock(assert_all_called=True) as router:
+        get_route = router.get(f"{JUPYTER_URL}/01DIRAAAAAAAAAAAAAAA").mock(
+            return_value=httpx.Response(
+                404, json={"detail": {"code": "NOT_FOUND"}}
+            )
+        )
+        put_route = router.put(f"{JUPYTER_URL}/01DIRAAAAAAAAAAAAAAA").mock(
+            return_value=httpx.Response(
+                201, json={"type": "directory"}
+            )
+        )
+
+        async with httpx.AsyncClient() as transport:
+            runtime.client = transport  # type: ignore[assignment]
+            await runtime.ensure_directory(
+                WORKSPACE_ID, "01DIRAAAAAAAAAAAAAAA"
+            )
+
+    assert get_route.called
+    assert put_route.called
+
+
+async def test_ensure_directory_propagates_non_404_error(
+    runtime: RuntimeClient,
+) -> None:
+    """GET 500 → propagate; do NOT fall through to PUT."""
+    put_called = False
+
+    def _track_put(request: httpx.Request) -> httpx.Response:
+        nonlocal put_called
+        put_called = True
+        return httpx.Response(201, json={})
+
+    with respx.mock(assert_all_called=False) as router:
+        router.get(f"{JUPYTER_URL}/01DIRAAAAAAAAAAAAAAA").mock(
+            return_value=httpx.Response(500, text="internal error")
+        )
+        router.put(f"{JUPYTER_URL}/01DIRAAAAAAAAAAAAAAA").mock(
+            side_effect=_track_put
+        )
+
+        async with httpx.AsyncClient() as transport:
+            runtime.client = transport  # type: ignore[assignment]
+            with pytest.raises(RuntimeClientError) as exc_info:
+                await runtime.ensure_directory(
+                    WORKSPACE_ID, "01DIRAAAAAAAAAAAAAAA"
+                )
+
+    assert exc_info.value.status_code == 500
+    assert not put_called, "PUT must not be issued after a non-404 GET error"
\ No newline at end of file
diff --git a/common/pyproject.toml b/common/pyproject.toml
index 22ae627..8c1e319 100644
--- a/common/pyproject.toml
+++ b/common/pyproject.toml
@@ -10,6 +10,7 @@ dependencies = [
     "boto3>=1.34,<2",
     "fastapi==0.116.1",
     "pydantic-settings>=2.14.2",
+    "loguru>=0.7.2",
     "passlib==1.7.4",
     "bcrypt>=4.0,<4.1",
     "aiofiles>=25.1.0",
diff --git a/common/src/common/config.py b/common/src/common/config.py
index 28cb4d1..ed869a1 100644
--- a/common/src/common/config.py
+++ b/common/src/common/config.py
@@ -60,6 +60,15 @@ class Settings(BaseSettings):
         description="Service label surfaced in lifespan / health checks.",
     )
 
+    # ── logging ───────────────────────────────────────────────────
+    log_level: str = Field(
+        default="DEBUG",
+        description=(
+            "loguru stderr sink level. One of DEBUG/INFO/WARNING/ERROR/CRITICAL; "
+            "anything else falls back to INFO inside configure_logging()."
+        ),
+    )
+
     # ── runtime container endpoint ───────────────────────────────
     runtime_api_url: str = Field(
         default="http://runtime:8000",
diff --git a/common/src/common/logging.py b/common/src/common/logging.py
new file mode 100644
index 0000000..2553a66
--- /dev/null
+++ b/common/src/common/logging.py
@@ -0,0 +1,39 @@
+"""Centralised loguru configuration. Call :func:`configure_logging` once per process, as early as possible. Idempotent."""
+
+from __future__ import annotations
+
+import sys
+
+from loguru import logger
+
+_DEFAULT_FORMAT = (
+    "<green>{time:YYYY-MM-DD HH:mm:ss.SSS}</green> "
+    "<level>{level: <8}</level> | "
+    "<cyan>{name}</cyan>:<cyan>{function}</cyan>:<cyan>{line}</cyan> - "
+    "<level>{message}</level>"
+)
+_ALLOWED_LEVELS = {"DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"}
+_CONFIGURED: bool = False
+
+
+def configure_logging(level: str = "INFO") -> str:
+    """Configure loguru's default stderr sink and return the normalised level."""
+    global _CONFIGURED
+    if _CONFIGURED:
+        return level.upper()
+
+    normalised = level.upper() if level.upper() in _ALLOWED_LEVELS else "INFO"
+
+    logger.remove()
+    logger.add(
+        sys.stderr,
+        level=normalised,
+        format=_DEFAULT_FORMAT,
+        backtrace=True,
+        diagnose=False,
+        enqueue=False,
+        catch=True,
+    )
+
+    _CONFIGURED = True
+    return normalised
diff --git a/uv.lock b/uv.lock
index 5a6ae25..282f753 100644
--- a/uv.lock
+++ b/uv.lock
@@ -688,6 +688,7 @@ dependencies = [
     { name = "boto3" },
     { name = "fastapi" },
     { name = "greenlet" },
+    { name = "loguru" },
     { name = "passlib" },
     { name = "pydantic-settings" },
     { name = "sqlalchemy" },
@@ -708,6 +709,7 @@ requires-dist = [
     { name = "boto3", specifier = ">=1.34,<2" },
     { name = "fastapi", specifier = "==0.116.1" },
     { name = "greenlet", specifier = ">=3.0.0" },
+    { name = "loguru", specifier = ">=0.7.2" },
     { name = "passlib", specifier = "==1.7.4" },
     { name = "pydantic-settings", specifier = ">=2.14.2" },
     { name = "sqlalchemy", specifier = "==2.0.51" },