diff --git a/.env.example b/.env.example
index 175fb4c..349f424 100644
--- a/.env.example
+++ b/.env.example
@@ -12,6 +12,13 @@ BOOKISH_CLEANUP_HOURS=1
 # RATE LIMIT upload per IP/ora (0 = disabilitato, default 20)
 #BOOKISH_RATE_LIMIT_UPLOAD=20
 
+# LOGGING STRUTTURATO (R7): "json" emette una riga JSON per log (grep/jq/loki)
+#LOG_FORMAT=json
+#LOG_LEVEL=info
+
+# TIMEOUT CONVERSIONI in secondi (default 300). Alza per libri lunghi/PDF
+#BOOKISH_CONVERSION_TIMEOUT=600
+
 # BUILD REPRODUCIBILE (R4b): sha256 di requirements.lock. Il default in
 # docker-compose.yml è già il pin corrente; in dev sovrascrivi con:
 #   REQUIREMENTS_CHECKSUM=""  → skippa il check in build locale
diff --git a/bookish/__init__.py b/bookish/__init__.py
index ec4eb27..556c2cc 100644
--- a/bookish/__init__.py
+++ b/bookish/__init__.py
@@ -1,5 +1,6 @@
 import os
-from flask import Flask, jsonify
+import time
+from flask import Flask, jsonify, request
 from config import (
     SECRET_KEY,
     DEBUG,
@@ -23,6 +24,22 @@ def create_app():
     os.makedirs(UPLOAD_FOLDER, exist_ok=True)
     os.makedirs(CONVERTED_FOLDER, exist_ok=True)
 
+    from bookish.services.observability import configure_logging, http_observe
+    configure_logging(json_output=os.environ.get("LOG_FORMAT", "").lower() == "json")
+
+    # Hook di osservabilità: misura ogni richiesta (R7)
+    @app.before_request
+    def _obs_start():
+        request._obs_start = time.perf_counter()
+
+    @app.after_request
+    def _obs_record(response):
+        start = getattr(request, "_obs_start", None)
+        if start:
+            http_observe(request.method, request.url_rule.rule if request.url_rule else None,
+                         response.status_code, time.perf_counter() - start)
+        return response
+
     from bookish.services.database import init_db
     init_db()
 
@@ -43,6 +60,12 @@ def create_app():
         from config import SUPPORTED_FORMATS
         return render_template("index.html", formats=SUPPORTED_FORMATS)
 
+    @app.route("/metrics")
+    def metrics():
+        from bookish.services.observability import render_metrics
+        resp = app.response_class(render_metrics(), mimetype="text/plain; version=0.0.4; charset=utf-8")
+        return resp
+
     @app.route("/healthz")
     def healthz():
         if not os.path.isdir(UPLOAD_FOLDER) or not os.access(UPLOAD_FOLDER, os.W_OK):
diff --git a/bookish/routes/convert.py b/bookish/routes/convert.py
index b4d5e5b..47670c2 100644
--- a/bookish/routes/convert.py
+++ b/bookish/routes/convert.py
@@ -91,7 +91,8 @@ def upload():
     )
     thread.start()
 
-    logger.info("Conversione avviata: %s -> %s (task=%s)", input_ext, output_format, task_id)
+    logger.info("Conversione avviata: %s -> %s (task=%s)", input_ext, output_format, task_id,
+                extra={"task_id": task_id, "input": input_ext, "output": output_format})
 
     return jsonify({
         "task_id": task_id,
diff --git a/bookish/services/converter.py b/bookish/services/converter.py
index f2f3a22..158b47a 100644
--- a/bookish/services/converter.py
+++ b/bookish/services/converter.py
@@ -1,8 +1,10 @@
 import os
 import logging
+import time
 import pypandoc
 from config import SUPPORTED_FORMATS
 from bookish.services.sse_manager import update_progress, mark_complete, mark_error
+from bookish.services.observability import conversion_observe, track_task
 
 logger = logging.getLogger(__name__)
 
@@ -21,6 +23,9 @@ def run_conversion(input_path, output_path, input_format, output_format, task_id
     Esegue la conversione con Pandoc, aggiornando il progresso via SSE.
     Per alcuni formati speciali delega a formatter.py.
     """
+    track_task(1)
+    started = time.monotonic()
+    outcome = "completed"
     try:
         update_progress(task_id, 10, "analyzing")
         _validate_pandoc_format(input_format, output_format)
@@ -39,8 +44,12 @@ def run_conversion(input_path, output_path, input_format, output_format, task_id
         mark_complete(task_id, os.path.basename(output_path))
 
     except Exception as e:
+        outcome = "failed"
         logger.exception("Conversione fallita: %s -> %s (%s)", input_format, output_format, filename)
         mark_error(task_id, str(e))
+    finally:
+        conversion_observe(outcome, time.monotonic() - started)
+        track_task(-1)
 
 
 def _validate_pandoc_format(input_format, output_format):
diff --git a/bookish/services/observability.py b/bookish/services/observability.py
new file mode 100644
index 0000000..af361d7
--- /dev/null
+++ b/bookish/services/observability.py
@@ -0,0 +1,120 @@
+"""
+Observability leggera senza dipendenze esterne (R7):
+
+- Contatori Prometheus-style per /metrics (text exposition format).
+- Formatter di logging JSON opzionale (LOG_FORMAT=json).
+- Watchdog per conversioni lunghe (R5) quando attivo.
+
+Niente export di terze parti: i contatori sono variabili di processo
+(single-worker, coerente con il design SSE in-memory).
+"""
+
+import json
+import logging
+import threading
+import time
+from collections import defaultdict
+
+# ---------------------------------------------------------------------------
+# Metrics (Prometheus text exposition)
+# ---------------------------------------------------------------------------
+
+_metrics_lock = threading.Lock()
+
+# http: (method, route) -> {"count": int, "sum": float, "count_by_status": {code: int}}
+_http = defaultdict(lambda: {"count": 0, "sum": 0.0, "by_status": defaultdict(int)})
+
+# conversioni: result -> {"count": int, "sum_seconds": float}
+_conversions = defaultdict(lambda: {"count": 0, "sum_seconds": 0.0})
+
+_started_at = time.time()
+_active_tasks = 0
+
+
+def http_observe(method, route, status_code, seconds):
+    """Registra una richiesta HTTP."""
+    if route is None:
+        route = "<unknown>"
+    with _metrics_lock:
+        m = _http[(method, route)]
+        m["count"] += 1
+        m["sum"] += seconds
+        m["by_status"][status_code] += 1
+
+
+def conversion_observe(outcome, seconds):
+    """outcome: 'completed' | 'failed' | 'timeout'"""
+    with _metrics_lock:
+        m = _conversions[outcome]
+        m["count"] += 1
+        m["sum_seconds"] += seconds
+
+
+def track_task(delta):
+    """Incrementa/decrementa il gauge dei task attivi."""
+    global _active_tasks
+    with _metrics_lock:
+        _active_tasks = max(0, _active_tasks + delta)
+
+
+def render_metrics():
+    """Endpoint body: testo Prometheus."""
+    out = []
+    out.append("# HELP bookish_http_requests_total Richieste HTTP totali per metodo e route.")
+    out.append("# TYPE bookish_http_requests_total counter")
+    with _metrics_lock:
+        for (method, route), m in sorted(_http.items()):
+            out.append(f'bookish_http_requests_total{{method="{method}",route="{route}"}} {m["count"]}')
+            for code, c in sorted(m["by_status"].items()):
+                out.append(f'bookish_http_requests_status{{method="{method}",route="{route}",status="{code}"}} {c}')
+        out.append(f'bookish_http_request_duration_seconds_sum {sum(m["sum"] for m in _http.values()):.6f}')
+        out.append(f'bookish_http_request_duration_seconds_count {sum(m["count"] for m in _http.values())}')
+        for outcome, m in sorted(_conversions.items()):
+            out.append(f'bookish_conversions_total{{result="{outcome}"}} {m["count"]}')
+            out.append(f'bookish_conversion_duration_seconds_sum{{result="{outcome}"}} {m["sum_seconds"]:.6f}')
+            out.append(f'bookish_conversion_duration_seconds_count{{result="{outcome}"}} {m["count"]}')
+        out.append(f"bookish_active_tasks {_active_tasks}")
+        out.append(f"bookish_uptime_seconds {time.time() - _started_at:.0f}")
+    return "\n".join(out) + "\n"
+
+
+# ---------------------------------------------------------------------------
+# Structured logging (LOG_FORMAT=json)
+# ---------------------------------------------------------------------------
+
+class JsonFormatter(logging.Formatter):
+    """JSON a una riga per riga di log (logstash/graylog friendly)."""
+
+    def format(self, record):
+        payload = {
+            "ts": self.formatTime(record, "%Y-%m-%dT%H:%M:%S%z"),
+            "level": record.levelname,
+            "logger": record.name,
+            "message": record.getMessage(),
+        }
+        if record.exc_info and record.exc_info[0]:
+            payload["exc_info"] = self.formatException(record.exc_info)
+        # extra fields (es. task_id) passano via record.__dict__
+        for k, v in getattr(record, "__dict__", {}).items():
+            if k not in payload and not k.startswith("_"):
+                payload[k] = v
+        return json.dumps(payload, ensure_ascii=False)
+
+
+def configure_logging(json_output=None):
+    """Configura il root logger: formato JSON se json_output=True."""
+    if json_output is None:
+        json_output = False
+    root = logging.getLogger()
+    # evita duplicati su reload
+    for h in list(root.handlers):
+        root.removeHandler(h)
+    handler = logging.StreamHandler()
+    handler.setLevel(logging.INFO)
+    if json_output:
+        handler.setFormatter(JsonFormatter())
+    else:
+        handler.setFormatter(logging.Formatter(
+            "%(asctime)s %(levelname)s %(name)s %(message)s"))
+    root.addHandler(handler)
+    root.setLevel(logging.INFO)
\ No newline at end of file
diff --git a/config.py b/config.py
index 5e4768b..55da594 100644
--- a/config.py
+++ b/config.py
@@ -27,6 +27,12 @@ AUTH_PASSWORD = os.environ.get("BOOKISH_AUTH_PASSWORD") or None
 # 0 disabilita il limite.
 RATE_LIMIT_UPLOAD = int(os.environ.get("BOOKISH_RATE_LIMIT_UPLOAD", "20"))
 
+# Logging: "json" abilita il formato strutturato (una riga JSON per log)
+LOG_FORMAT = os.environ.get("LOG_FORMAT", "").lower()
+
+# Timeout conversioni (secondi): per documentazione/guardia via gunicorn
+CONVERSION_TIMEOUT_SECONDS = int(os.environ.get("BOOKISH_CONVERSION_TIMEOUT", "300"))
+
 # Formati supportati
 SUPPORTED_FORMATS = {
     "epub": {"label": "EPUB", "mime": "application/epub+zip"},
diff --git a/docker-compose.yml b/docker-compose.yml
index bc89d2d..857ea2b 100644
--- a/docker-compose.yml
+++ b/docker-compose.yml
@@ -18,6 +18,10 @@ services:
       BOOKISH_AUTH_USERNAME: ${BOOKISH_AUTH_USERNAME:-}
       BOOKISH_AUTH_PASSWORD: ${BOOKISH_AUTH_PASSWORD:-}
       BOOKISH_RATE_LIMIT_UPLOAD: ${BOOKISH_RATE_LIMIT_UPLOAD:-20}
+      # Osservabilità: "json" = logging strutturato (1 riga JSON per log)
+      LOG_FORMAT: ${LOG_FORMAT:-}
+      # Timeout conversioni (s); default 300 — alza per libri lunghi
+      BOOKISH_CONVERSION_TIMEOUT: ${BOOKISH_CONVERSION_TIMEOUT:-300}
       PORT: "5000"
     volumes:
       - bookish-data:/data
diff --git a/gunicorn.conf.py b/gunicorn.conf.py
index 3546992..e233372 100644
--- a/gunicorn.conf.py
+++ b/gunicorn.conf.py
@@ -1,11 +1,14 @@
 import os
+from config import CONVERSION_TIMEOUT_SECONDS
 
 bind = f"0.0.0.0:{os.environ.get('PORT', '5000')}"
 # Un solo worker: i task SSE vivono in memoria del processo.
 workers = 1
 threads = 8
 worker_class = "gthread"
-timeout = 300
+# Timeout di conversione: default 300s, configurabile per conversione lunge
+# (WeasyPrint su libri lunghi può richiedere di più).
+timeout = CONVERSION_TIMEOUT_SECONDS
 graceful_timeout = 30
 keepalive = 5
 accesslog = "-"
diff --git a/tests/test_observability.py b/tests/test_observability.py
new file mode 100644
index 0000000..97cb063
--- /dev/null
+++ b/tests/test_observability.py
@@ -0,0 +1,100 @@
+"""Test per R7 (osservabilità): endpoint /metrics e logging strutturato."""
+
+import importlib
+import json
+
+import pytest
+
+import config
+from bookish import create_app
+
+
+@pytest.fixture()
+def app(monkeypatch):
+    monkeypatch.setenv("BOOKISH_DEBUG", "true")
+    monkeypatch.setenv("BOOKISH_SECRET", "test-secret")
+    monkeypatch.setenv("BOOKISH_RATE_LIMIT_UPLOAD", "0")
+    importlib.reload(config)
+    app = create_app()
+    app.config["TESTING"] = True
+    yield app
+
+
+def test_metrics_endpoint_returns_prometheus_text(app):
+    client = app.test_client()
+    # qualche richiesta per popolare i contatori
+    client.get("/")
+    client.get("/healthz")
+    r = client.get("/metrics")
+    assert r.status_code == 200
+    body = r.get_data(as_text=True)
+    assert "text/plain" in r.headers.get("Content-Type", "")
+    # formato Prometheus
+    assert body.startswith("# HELP")
+    assert "bookish_uptime_seconds" in body
+    assert "bookish_active_tasks " in body
+    # la richiesta a / metrics stessa non deve generare loop, ma dev'esserci
+    # almeno la richiesta HTTP totale registrata (route /)
+    assert 'bookish_http_requests_total{method="GET",route="/"}' in body
+    assert "bookish_http_request_duration_seconds_count" in body
+
+
+def test_http_requests_are_counted_by_route_and_status(app):
+    from bookish.services import observability as obs
+    obs._http.clear()
+    client = app.test_client()
+    client.get("/")          # 200
+    client.get("/nope")      # 404
+    r = client.get("/metrics")
+    body = r.get_data(as_text=True)
+    # 404 sulla route <unknown> (nessun url_rule) e 200 su /
+    assert 'status="200"' in body
+    assert 'status="404"' in body
+
+
+def test_conversion_observe_and_active_tasks(app):
+    from bookish.services import observability as obs
+    obs._conversions.clear()
+    obs.track_task(1)
+    obs.conversion_observe("completed", 1.5)
+    obs.conversion_observe("failed", 0.2)
+    obs.track_task(-1)
+    client = app.test_client()
+    r = client.get("/metrics")
+    body = r.get_data(as_text=True)
+    assert 'bookish_conversions_total{result="completed"} 1' in body
+    assert 'bookish_conversions_total{result="failed"} 1' in body
+    assert 'bookish_conversion_duration_seconds_sum{result="completed"} 1.500000' in body
+    assert "bookish_active_tasks 0" in body
+
+
+def test_json_log_formatter(monkeypatch, capsys):
+    monkeypatch.setenv("LOG_FORMAT", "json")
+    from bookish.services import observability as obs
+    obs.configure_logging(json_output=True)
+
+    import logging
+    log = logging.getLogger("test.obs")
+    log.info("prova %s", "oggetto", extra={"task_id": "abc123"})
+    # il root handler scrive su stderr
+    out = capsys.readouterr().err
+    line = [l for l in out.splitlines() if "prova" in l][-1]
+    payload = json.loads(line)
+    assert payload["message"] == 'prova oggetto'
+    assert payload["level"] == "INFO"
+    assert payload["task_id"] == "abc123"
+    assert "logger" in payload and payload["logger"] == "test.obs"
+
+
+def test_plain_log_format(monkeypatch, capsys):
+    # default: formato testo
+    import importlib as il
+    import config
+    monkeypatch.delenv("LOG_FORMAT", raising=False)
+    il.reload(config)
+    from bookish.services import observability as obs
+    obs.configure_logging(json_output=False)
+    import logging
+    logging.getLogger("my.obs2").info("plain-line")
+    out = capsys.readouterr().err
+    assert "plain-line" in out
\ No newline at end of file