"""Serve the local Mountain Twin visual prototype without cloud services."""

from __future__ import annotations

import argparse
import http.server
import json
import os
import re
import sqlite3
import sys
import threading
import time
from concurrent.futures import ThreadPoolExecutor
from datetime import datetime, timezone
from functools import partial
from pathlib import Path
from urllib.parse import parse_qs, quote, unquote, urlsplit
from zoneinfo import ZoneInfo

from mountain_twin.accounts import http_api as accounts_api
from mountain_twin.accounts.store import AccountStore
from mountain_twin.api_v1.router import API_PREFIX, ApiV1, RateLimiter
from mountain_twin.api_v1.router import Context as ApiContext
from mountain_twin.brand import brand_script
from mountain_twin.budgets import state_document as budgets_state_document
from mountain_twin.explorer_service import ExplorerAnalysisService, parse_explorer_start_datetime
from mountain_twin.exposure import RoutePoint
from mountain_twin.heavy_queue import BUSY_DOCUMENT, HeavyBusy, HeavyQueue
from mountain_twin.journey import (
    CampMarker,
    JourneyCatalogService,
    JourneyPlanRepository,
    JourneyReadService,
    JourneyRouteRepository,
    JourneySqliteDatabase,
    RouteSelectionAnchor,
    TravelLegRepository,
    derive_day_segments,
    load_tmb_stage_summaries,
    route_file_service,
    tmb_day_1_repository,
)
from mountain_twin.journey.activities import (
    activities_document,
    activity_for,
    pace_factor_for_activity,
)
from mountain_twin.journey.calendar_events import (
    CalendarEventRepository,
    calendar_event_fields_from_payload,
)
from mountain_twin.journey.camps_service import (
    DEFAULT_PACE_PROFILE_ID,
    CampsService,
    durable_camps_document,
    night_conditions_for_point,
)
from mountain_twin.journey.day_derivation import (
    NO_PLANNED_START,
    RELATIVE_TIMING_REFERENCE_START,
    DayDerivationUnavailable,
    derive_relative_day_segments,
    derive_untimed_day_segments,
    relative_day_document,
)
from mountain_twin.journey.geometry_bootstrap import geometry_only_bootstrap_document
from mountain_twin.journey.plan_persistence import (
    CreateJourneyPlan,
    camp_marker_inputs_from_payload,
    journey_plan_document,
)
from mountain_twin.journey.route_files import MAX_FILE_BYTES
from mountain_twin.journey.route_imports import PendingImports, RouteImportRepository
from mountain_twin.journey.route_persistence import (
    CreateRouteRevision,
    route_points_from_geometry_geojson,
    route_points_from_payload,
    route_revision_document,
)
from mountain_twin.journey.route_profile import route_profile_document
from mountain_twin.journey.route_provider import fetch_route_segment_detail
from mountain_twin.journey.route_segments import movement_segment_indexes, validate_segments
from mountain_twin.journey.route_surface import route_surface_document
from mountain_twin.journey.route_timezone import resolve_route_timezone
from mountain_twin.journey.telemetry import (
    next_hours_telemetry_document,
    planned_telemetry_document,
    storms_document,
)
from mountain_twin.journey.terrain_light import terrain_light_document
from mountain_twin.journey.travel_legs import travel_leg_fields_from_payload
from mountain_twin.journey.weather_anchor import anchored_plan, next_start
from mountain_twin.journey.weather_prefetch import (
    PrefetchCandidate,
    WeatherPrefetcher,
    prefetch_enabled,
    prefetch_interval_seconds,
)
from mountain_twin.journey.weather_prefetch import (
    fresh_for as prefetch_fresh_for,
)
from mountain_twin.journey.weather_series import walking_times_document, weather_series_document
from mountain_twin.live_tracking import LiveTrackingError, LiveTrackingService
from mountain_twin.live_tracking.http_api import LiveTrackingHttp
from mountain_twin.live_tracking.owner import LinkBook, LocalLiveOwner, RemoteLiveOwner
from mountain_twin.offline import (  # AV-045: offline mode for tests
    offline_cache_directory,
    offline_mode,
    urlopen,
)
from mountain_twin.pace.pauses import generate_relative_pauses
from mountain_twin.pace.profiles import pace_profiles_document
from mountain_twin.places.osm_tags import OsmTagLookup
from mountain_twin.places.service import PlaceWeatherService
from mountain_twin.route_analysis import prepare_route
from mountain_twin.terrain.copernicus import CopernicusGlo90
from mountain_twin.usage import UsageStore
from mountain_twin.usage import configure as configure_usage
from mountain_twin.usage import store as usage_store
from mountain_twin.usage_alarms import UsageAlarms
from mountain_twin.usage_report import usage_summary_document
from mountain_twin.visualization import write_solar_visualization_export
from mountain_twin.weather.clouds import EumetsatCloudProvider, disabled_clouds_document
from mountain_twin.weather.forecast_cache import ForecastCellCache, fresh_for_at_least
from mountain_twin.weather.live import LiveWeatherPointResolver
from mountain_twin.weather.met_norway import MetNorwayClient
from mountain_twin.weather.provider import OpenMeteoProvider
from mountain_twin.weather.radar import (
    RainViewerRadarProvider,
    disabled_radar_document,
    radar_enabled,
)

# The saved-Journey database lives in the repository's git-ignored
# data/local/ (the existing local-cache directory), not in /tmp, which the OS
# may clear. LEGACY_JOURNEY_DB is the old default, migrated once by copy.
DEFAULT_JOURNEY_DB = Path("data/local/aventurro-journeys.sqlite3")
LEGACY_JOURNEY_DB = Path("/tmp/aventurro-journey-dev.sqlite3")


def prepare_journey_db(
    explicit: Path | None, root: Path, legacy: Path = LEGACY_JOURNEY_DB
) -> tuple[Path, str | None]:
    """Return the Journey database path to serve and a startup note (or None).

    An explicit --journey-db is used as given. Otherwise the default under
    <root>/data/local/ is used, its directory created; when it does not exist
    yet but the legacy /tmp file does, the legacy database is COPIED (SQLite
    backup API, consistent under WAL) -- never moved, the original is kept,
    and an existing new file is never overwritten."""
    if explicit is not None:
        return explicit.resolve(), None
    target = (root / DEFAULT_JOURNEY_DB).resolve()
    target.parent.mkdir(parents=True, exist_ok=True)
    if target.exists() or not legacy.exists():
        return target, None
    partial_copy = target.with_name(target.name + ".partial")
    partial_copy.unlink(missing_ok=True)
    source = sqlite3.connect(f"file:{legacy}?mode=ro", uri=True)
    try:
        destination = sqlite3.connect(partial_copy)
        try:
            source.backup(destination)
        finally:
            destination.close()
    finally:
        source.close()
    # Only a complete copy becomes the database; an interrupted one is retried.
    partial_copy.rename(target)
    return target, f"Copied saved Journeys from {legacy} to {target} (the original is kept)."


DEFAULT_SOURCE = Path(
    "data/generated/route_solar_exposure/tmb_day_01_glo30_surface_illumination_adaptive.json"
)
DEFAULT_EXPORT = Path("data/generated/visualization/tmb_day_01_solar_exposure_v0_1.json")


class ExplorerHTTPServer(http.server.ThreadingHTTPServer):
    """Local-only server with a lazily initialized cached Explorer service."""

    allow_reuse_address = True
    # A whole-journey view asks for one profile per day at once (11 for the
    # reference TMB); the socketserver default backlog of 5 resets some.
    request_queue_size = 64

    def __init__(self, address, handler, root: Path, journey_db: Path, journey_owner: str):
        super().__init__(address, handler)
        self.explorer_root = root
        self.explorer_service = None
        self.explorer_service_lock = threading.Lock()
        self.journey_service = None
        self.journey_service_lock = threading.Lock()
        self.journey_db = journey_db
        self.journey_owner = journey_owner
        self.camps_service = None
        self.camps_service_lock = threading.Lock()
        self.live_weather_resolver = None
        self.live_weather_resolver_lock = threading.Lock()
        self.forecast_cache = None
        # AV-048: stale forecasts are fetched again here, one task at a time.
        self.weather_refresher = ThreadPoolExecutor(max_workers=1, thread_name_prefix="weather")
        # The 3D view's terrain (Copernicus GLO-90, cached on disk) and the
        # light documents already computed for a route and moment.
        self.terrain_dem_provider = None
        self.terrain_light_cache: dict[tuple, dict] = {}
        self.terrain_light_lock = threading.Lock()
        # Storms tab radar (docs/profile_storms_v0_1_design.md section 4):
        # one adapter instance (and its ~5 min index cache) per server;
        # AVENTURRO_RADAR=off turns the layer off before a commercial release.
        self.radar_provider = RainViewerRadarProvider()
        # Storms / Precipitation tab cloud imagery (EUMETSAT, same flag).
        self.clouds_provider = EumetsatCloudProvider()
        # Route drawing v0.1 (docs/route_drawing_v0_1_design.md, Krok 2):
        # swappable so tests can inject a fake opener instead of making a
        # real brouter.de call, the same injection point
        # mountain_twin.weather.provider.OpenMeteoProvider already uses.
        self.routing_opener = urlopen
        # AV-054 (E1): accounts -- sign-in, sessions, one owner per request --
        # unless AUTH_MODE=none says otherwise, explicitly (local_dev.env on
        # the development Mac): a server started without the setting is never
        # open to everyone by mistake.
        self.accounts = (os.environ.get("AUTH_MODE") or "").strip().lower() != "none"
        self.login_limiter = accounts_api.LoginLimiter()
        # D1: one heavy analysis at a time (mountain_twin/heavy_queue.py).
        self.heavy = HeavyQueue()
        # D5: the reference TMB Journey's Explorer analysis, computed once per
        # (route, start) and kept -- ~30 s of one core each time otherwise.
        self.reference_documents: dict = {}
        self.reference_documents_lock = threading.Lock()
        # E2: API v1 for the mobile app (mountain_twin/api_v1/router.py).
        self.api_v1 = ApiV1(RateLimiter())
        # AV-053: route files between their preview and the save.
        self.pending_imports = PendingImports()
        # Live tracking (AV-022): built on first use, see get_live().
        self.live = None
        # "Pogoda" for any place (AV-062): built on first use.
        self.place_weather = None
        self.place_weather_lock = threading.Lock()
        self.live_lock = threading.Lock()

    def get_explorer_service(self) -> ExplorerAnalysisService:
        if self.explorer_service is None:
            with self.explorer_service_lock:
                if self.explorer_service is None:
                    self.explorer_service = ExplorerAnalysisService(self.explorer_root)
        return self.explorer_service

    def get_journey_service(self) -> JourneyReadService:
        if self.journey_service is None:
            with self.journey_service_lock:
                if self.journey_service is None:
                    explorer = self.get_explorer_service()
                    self.journey_service = JourneyReadService(
                        self.explorer_root,
                        tmb_day_1_repository(),
                        producer_document=lambda route_id, start: self.reference_document(
                            explorer, route_id, start
                        ),
                    )
        return self.journey_service

    def reference_document(self, explorer, route_id, start):
        key = (route_id, start.isoformat())
        with self.reference_documents_lock:
            if key in self.reference_documents:
                return self.reference_documents[key]
        document = self.heavy.run(lambda: explorer.analyze_explorer_request(route_id, start))
        with self.reference_documents_lock:
            if len(self.reference_documents) >= 8:
                self.reference_documents.pop(next(iter(self.reference_documents)))
            self.reference_documents[key] = document
        return document

    def warm_reference_journey(self) -> None:
        """D5: the reference Journey analysed once, in the background, right
        after start -- the first participant opening it does not wait."""
        try:
            service = self.get_journey_service()
            journey_id = "journey-tmb-day-01-v0_1"
            run = service.get_run(journey_id)
            scenarios = service.repository.scenarios_for_run(run.analysis_run_id)
            scenario = next((item for item in scenarios if item.name == "NOMINAL"), scenarios[0])
            service.get_product_bootstrap(journey_id, scenario.analysis_scenario_id)
            print("Reference Journey analysed and kept (D5)", file=sys.stderr, flush=True)
        except Exception as error:  # pragma: no cover - a warm-up never stops the server
            print(f"Reference Journey warm-up failed: {error}", file=sys.stderr, flush=True)

    def get_live(self):
        """Live tracking (AV-022): (owner backend, local HTTP surface or None).

        LIVE_TRACKING_URL + LIVE_TRACKING_ADMIN_SECRET (local_dev.env) -> the
        public server's owner API; its pages live there. Otherwise the same
        service in this process, next to the Journey database (its own
        SQLite file), with its pages served here under /live/ -- the same
        code and rules, only not reachable from outside this machine."""
        if self.live is None:
            with self.live_lock:
                if self.live is None:
                    folder = self.journey_db.parent
                    book = LinkBook(folder / "live-tracking-links.json")
                    url = (os.environ.get("LIVE_TRACKING_URL") or "").strip().rstrip("/")
                    secret = (os.environ.get("LIVE_TRACKING_ADMIN_SECRET") or "").strip()
                    if url and secret:
                        self.live = (RemoteLiveOwner(url, secret, book), None)
                    else:
                        service = LiveTrackingService(
                            folder / "live-tracking.sqlite3",
                            _local_secret(folder / "live-tracking-grant.key"),
                        )
                        surface = LiveTrackingHttp(
                            service,
                            code_root=self.explorer_root,
                            brand_script=brand_script,
                            map_config=_local_live_map_config,
                        )
                        self.live = (LocalLiveOwner(service, book), surface)
        return self.live

    def get_journey_catalog(self) -> JourneyCatalogService:
        """Return a connection-free catalog; each operation owns its request connection."""

        return JourneyCatalogService(JourneySqliteDatabase(self.journey_db))

    def get_journey_route_repository(self) -> JourneyRouteRepository:
        """Same per-request-connection philosophy as get_journey_catalog():
        route persistence v0.1 (docs/route_persistence_v0_1_design.md) is a
        durable-Journey/SQLite concern, unrelated to tmb_day_1_repository()."""

        return JourneyRouteRepository(JourneySqliteDatabase(self.journey_db))

    def get_journey_plan_repository(self) -> JourneyPlanRepository:
        """Same per-request-connection philosophy as get_journey_catalog();
        Work Package 2 of docs/route_persistence_v0_1_design.md."""

        return JourneyPlanRepository(JourneySqliteDatabase(self.journey_db))

    def get_travel_leg_repository(self) -> TravelLegRepository:
        """Same per-request-connection philosophy as get_journey_catalog();
        the Kalendarz tab's manually-added travel legs (AV-009)."""

        return TravelLegRepository(JourneySqliteDatabase(self.journey_db))

    def get_calendar_event_repository(self) -> CalendarEventRepository:
        """AV-038: the Kalendarz tab's lodging and other events."""

        return CalendarEventRepository(JourneySqliteDatabase(self.journey_db))

    def radar_frames_document(self) -> dict:
        if not radar_enabled():
            return disabled_radar_document()
        return self.radar_provider.frames_document()

    def clouds_frame_document(self, latitude: float, longitude: float) -> dict:
        if not radar_enabled():
            return disabled_clouds_document()
        return self.clouds_provider.frame_document(latitude, longitude)

    def get_live_weather_resolver(self) -> LiveWeatherPointResolver:
        """One resolver (and its in-memory per-point cache) for the whole
        server process, shared by CampsService and the stateless
        day-derivation preview -- a repeated render of the same point must
        not re-hit Open-Meteo."""
        if self.live_weather_resolver is None:
            with self.live_weather_resolver_lock:
                if self.live_weather_resolver is None:
                    # AV-048: the shared grid-cell cache with stale-while-
                    # revalidate for the multi-point lookups (Weather, Storms).
                    # AV-051: MET Norway when Open-Meteo refuses or nears
                    # its limit (WEATHER_PRIMARY=met_norway asks it first).
                    met_directory = (
                        offline_cache_directory()
                        if offline_mode() is not None
                        else self.explorer_root / "data/cache/weather"
                    ) / "met_norway"
                    self.live_weather_resolver = LiveWeatherPointResolver(
                        OpenMeteoProvider(self.explorer_root / "data/cache/weather/open_meteo"),
                        cell_cache=self.get_forecast_cache(),
                        refresher=self.weather_refresher.submit,
                        log=lambda line: print(line, file=sys.stderr, flush=True),
                        fallback=MetNorwayClient(met_directory),
                        primary=(os.environ.get("WEATHER_PRIMARY") or "open_meteo").strip().lower(),
                    )
        return self.live_weather_resolver

    def get_forecast_cache(self) -> ForecastCellCache:
        """AV-048: one SQLite file for every Journey's forecasts (and the
        day's provider call count); offline mode keeps its own, never the
        development server's."""
        if self.forecast_cache is None:
            directory = (
                offline_cache_directory()
                if offline_mode() is not None
                else self.explorer_root / "data/cache/weather"
            )
            # AV-051: the events say which server wrote them (this port),
            # so a test or a measurement never mixes with the real use.
            self.forecast_cache = ForecastCellCache(
                directory / "forecast_cache.sqlite3",
                instance=f"serve_visual_prototype:{self.server_address[1]}",
            )
        return self.forecast_cache

    def get_place_weather(self) -> PlaceWeatherService:
        """AV-062: "Pogoda" for a peak, a town or a point -- the same live
        resolver and forecast cache as the Journeys, the same DEM as the 3D
        view, favourites in the Journey database."""
        if self.place_weather is None:
            with self.place_weather_lock:
                if self.place_weather is None:
                    # AV-063: a peak's own OSM height and importance
                    # (Nominatim), kept for good; offline mode keeps its own.
                    directory = (
                        offline_cache_directory()
                        if offline_mode() is not None
                        else self.explorer_root / "data/cache/places"
                    )
                    osm_tags = OsmTagLookup(directory / "osm_tags.sqlite3")
                    self.place_weather = PlaceWeatherService(
                        resolver=self.get_live_weather_resolver,
                        dem=self.get_terrain_dem_provider,
                        favorites_database=lambda: JourneySqliteDatabase(self.journey_db),
                        osm_tags=lambda: osm_tags,
                    )
        return self.place_weather

    def get_terrain_dem_provider(self) -> CopernicusGlo90:
        if self.terrain_dem_provider is None:
            with self.terrain_light_lock:
                if self.terrain_dem_provider is None:
                    self.terrain_dem_provider = CopernicusGlo90(
                        self.explorer_root / "data/cache/dem/copernicus_glo90"
                    )
        return self.terrain_dem_provider

    def get_camps_service(self) -> CampsService:
        if self.camps_service is None:
            with self.camps_service_lock:
                if self.camps_service is None:
                    # Shares the SAME repository as get_journey_service() (not a
                    # fresh tmb_day_1_repository()) so an added camp is visible
                    # to every other Journey endpoint on this server instance.
                    repository = self.get_journey_service().repository
                    self.camps_service = CampsService(
                        self.explorer_root,
                        repository,
                        live_resolver=self.get_live_weather_resolver(),
                    )
        return self.camps_service


class ExplorerRequestHandler(http.server.SimpleHTTPRequestHandler):
    """Static-file handler plus a local Explorer analysis request endpoint."""

    # --- Accounts (AV-054 E1) -----------------------------------------------------
    # Public without a session: the app's static files (the sign-in page
    # among them), /api/auth/..., the live-tracking pages (their own tokens)
    # and these few documents that carry no one's data.
    PUBLIC_API = {
        "/api/runtime/server",
        "/api/runtime/map",
        "/api/activities",
        "/api/pace-profiles",
    }

    @property
    def owner_id(self) -> str:
        """Whose data this request reads and writes: the signed-in user, or
        the server's single owner in local mode -- never from the request."""
        if self.server.accounts:
            return self._session.user.user_id
        return self.server.journey_owner

    @property
    def is_system_owner(self) -> bool:
        return not self.server.accounts or self._session.user.role == "OWNER"

    def _accounts_store(self) -> AccountStore:
        return AccountStore(JourneySqliteDatabase(self.server.journey_db))

    def _resolve_session(self):
        token = accounts_api.cookie_value(self.headers.get("Cookie"))
        store = self._accounts_store()
        try:
            return token, store.session(token)
        finally:
            store.close()

    def _auth_error(self, status, code, message):
        _send_json(self, http.HTTPStatus(status), {"error": {"code": code, "message": message}})

    @property
    def real_ip(self) -> str:
        """The client's address: X-Real-IP only when the connection is nginx's
        on this machine (127.0.0.1), otherwise the socket's own address -- a
        client cannot make up where it comes from."""
        address = self.client_address[0]
        forwarded = (self.headers.get("X-Real-IP") or "").strip()
        if address in ("127.0.0.1", "::1") and forwarded:
            return forwarded
        return address

    def _auth_gate(self) -> bool:
        """True: the request was answered here (refused, or /api/auth/...).
        AUTH_MODE=none: no accounts -- and nothing through a proxy."""
        self._session = None
        if not self.server.accounts and (
            self.headers.get("X-Forwarded-For") or self.headers.get("X-Real-IP")
        ):
            # Without accounts the server is for this machine only: a request
            # that came through a proxy means it is exposed -- refuse it, loudly.
            self.log_error("AUTH_MODE=none refuses a proxied request (%s)", self.path.split("?")[0])
            _send_json(
                self,
                http.HTTPStatus.SERVICE_UNAVAILABLE,
                {
                    "error": {
                        "code": "NO_ACCOUNTS_BEHIND_PROXY",
                        "message": "Serwer bez kont nie przyjmuje ruchu z zewnątrz.",
                    }
                },
            )
            return True
        if not self.server.accounts:
            if urlsplit(self.path).path == "/api/auth/me":
                # The page asks; local mode has no sign-in at all.
                _send_json(self, http.HTTPStatus.OK, {"mode": "local", "user": None})
                return True
            return False
        path = urlsplit(self.path).path
        token, self._session = self._resolve_session()
        authorization = self.headers.get("Authorization") or ""
        if authorization:
            # E2: an API token (mobile app) -- only for /api/v1, never with a cookie.
            if not path.startswith(API_PREFIX + "/") or not authorization.startswith("Bearer "):
                self._auth_error(401, "AUTH_REQUIRED", "Token API działa tylko dla /api/v1.")
                return True
            store = self._accounts_store()
            try:
                self._session = store.token_session(authorization[7:].strip())
            finally:
                store.close()
            if self._session is None:
                self._auth_error(401, "BAD_TOKEN", "Token jest nieważny, odwołany albo wygasł.")
                return True
        if path.startswith("/api/auth/"):
            self._accounts_request(path, token)
            return True
        if path.startswith("/live/admin/"):
            # The public server's own live service has the admin API (E3);
            # never through the app.
            _send_json(self, http.HTTPStatus.NOT_FOUND, {"error": "not found"})
            return True
        if path.startswith("/live/") or path in self.PUBLIC_API:
            return False
        if not path.startswith("/api/"):
            return False  # static files: the page itself asks /api/auth/me
        if self._session is None:
            self._auth_error(401, "AUTH_REQUIRED", "Zaloguj się.")
            return True
        if (
            self.command != "GET"
            and not self._session.via_token
            and not accounts_api.csrf_ok(
                self._session, dict(self.headers.items()), self.headers.get("Host")
            )
        ):
            self._auth_error(403, "CSRF", "Odśwież stronę i spróbuj ponownie.")
            return True
        if path.startswith("/api/usage/") and not self.is_system_owner:
            self._auth_error(404, "NOT_FOUND", "Nie ma takiej strony.")
            return True
        parts = path.strip("/").split("/")
        if len(parts) >= 4 and parts[:3] == ["api", "v1", "journeys"]:
            # API v1: only the user's own saved Journeys (never the reference one).
            journey_id = unquote(parts[3])
            if self.server.get_journey_catalog().is_reference(
                journey_id
            ) or not self._may_use_journey(journey_id):
                _send_json(
                    self,
                    http.HTTPStatus.NOT_FOUND,
                    {"error": {"code": "NOT_FOUND", "message": "Nie ma takiej wyprawy."}},
                )
                return True
        if len(parts) >= 3 and parts[:2] == ["api", "journeys"]:
            if not self._may_use_journey(unquote(parts[2])):
                # Someone else's Journey, or none: never says which (404).
                _send_json(self, http.HTTPStatus.NOT_FOUND, {"error": "journey not found"})
                return True
        return False

    def _may_use_journey(self, journey_id: str) -> bool:
        catalog = self.server.get_journey_catalog()
        if catalog.is_reference(journey_id):
            # The reference TMB Journey: everyone reads it, only the owner changes it.
            return self.command == "GET" or self.is_system_owner
        connection = JourneySqliteDatabase(self.server.journey_db).connect()
        try:
            row = connection.execute(
                "SELECT 1 FROM journeys WHERE journey_id = ? AND owner_id = ?",
                (journey_id, self.owner_id),
            ).fetchone()
        finally:
            connection.close()
        return row is not None

    def _accounts_request(self, path: str, token) -> None:
        body = None
        if self.command in ("POST", "DELETE", "PUT"):
            length = int(self.headers.get("Content-Length") or 0)
            try:
                body = json.loads(self.rfile.read(length) or b"{}") if length else {}
            except ValueError:
                self._auth_error(400, "BAD_JSON", "Nieprawidłowe dane.")
                return
        store = self._accounts_store()
        try:
            status, document, headers = accounts_api.handle(
                self.command,
                path,
                parse_qs(urlsplit(self.path).query),
                body if isinstance(body, dict) else {},
                store=store,
                session=self._session,
                cookie_token=token,
                client=self.real_ip,
                headers=dict(self.headers.items()),
                host=self.headers.get("Host"),
                limiter=self.server.login_limiter,
                delete_user_data=self._delete_user_data,
            )
        finally:
            store.close()
        payload = json.dumps(document, sort_keys=True, allow_nan=False).encode("utf-8")
        self.send_response(status)
        self.send_header("Content-Type", "application/json; charset=utf-8")
        self.send_header("Cache-Control", "no-store")
        for name, value in headers.items():
            self.send_header(name, value)
        self.send_header("Content-Length", str(len(payload)))
        self.end_headers()
        self.wfile.write(payload)

    def _delete_user_data(self, user_id: str) -> dict:
        """ "Usuń konto": every Journey of the user (routes, plans, imported
        files, calendar, live tracking), their favourite places and files
        waiting for a save. The account row goes after (AccountStore)."""
        catalog = self.server.get_journey_catalog()
        connection = JourneySqliteDatabase(self.server.journey_db).connect()
        try:
            journeys = [
                row["journey_id"]
                for row in connection.execute(
                    "SELECT journey_id FROM journeys WHERE owner_id = ?", (user_id,)
                )
            ]
            favorites = connection.execute(
                "DELETE FROM place_favorites WHERE owner_id = ?", (user_id,)
            ).rowcount
        finally:
            connection.close()
        for journey_id in journeys:
            catalog.delete(user_id, journey_id)
            try:
                self.server.get_live()[0].delete_journey(journey_id)
            except LiveTrackingError as error:
                self.log_error(
                    "Live-tracking data of %s not removed: %s", journey_id, error.message
                )
        self.server.pending_imports.forget_owner(user_id)
        return {"journeys_deleted": len(journeys), "favorites_deleted": favorites}

    # --- API v1 (AV-054 E2) ----------------------------------------------------------
    def _api_v1(self) -> None:
        parsed = urlsplit(self.path)
        length = int(self.headers.get("Content-Length") or 0)
        raw = self.rfile.read(length) if length else b""
        session = self._session if self.server.accounts else None
        principal = (
            f"token:{session.token_id}"
            if session is not None and session.via_token
            else f"user:{self.owner_id}"
        )
        catalog = self.server.get_journey_catalog()

        def delete_journey(journey_id):
            catalog.delete(self.owner_id, journey_id)
            try:
                self.server.get_live()[0].delete_journey(journey_id)
            except LiveTrackingError as error:
                self.log_error(
                    "Live-tracking data of %s not removed: %s", journey_id, error.message
                )

        context = ApiContext(
            journey_db=self.server.journey_db,
            owner_id=self.owner_id,
            principal=principal,
            session=session,
            accounts_store=self._accounts_store if self.server.accounts else None,
            days=lambda journey_id: self._durable_camps_document(journey_id, None),
            weather=lambda journey_id, query: weather_series_for(self.server, journey_id, query),
            gpx=self._gpx_file,
            delete_journey=delete_journey,
            create_journey=lambda payload: catalog.create(self.owner_id, payload),
        )
        status, document, headers = self.server.api_v1.handle(
            context,
            self.command,
            parsed.path,
            parse_qs(parsed.query),
            raw,
            dict(self.headers.items()),
        )
        if isinstance(document, tuple):  # a file: (bytes, content type, name)
            body, content_type, name = document
            self.send_response(status)
            self.send_header("Content-Type", content_type)
            self.send_header(
                "Content-Disposition",
                f'attachment; filename="{route_file_service.ascii_file_name(name)}"',
            )
        else:
            body = json.dumps(document, sort_keys=True, allow_nan=False).encode("utf-8")
            self.send_response(status)
            self.send_header("Content-Type", "application/json; charset=utf-8")
        self.send_header("Cache-Control", "no-store")
        for name, value in headers.items():
            self.send_header(name, value)
        self.send_header("Content-Length", str(len(body)))
        self.end_headers()
        self.wfile.write(body)

    def _gpx_file(self, journey_id: str, query) -> tuple[bytes, str]:
        """API v1's GPX: the same file as the web app's "Pobierz GPX"."""
        owner = self.owner_id
        journey = self.server.get_journey_catalog().get_journey(owner, journey_id)
        routes = self.server.get_journey_route_repository()
        plans = self.server.get_journey_plan_repository()
        imports = RouteImportRepository(JourneySqliteDatabase(self.server.journey_db))
        try:
            result = routes.get_current_route(journey_id=journey_id, owner_id=owner)
            plan = plans.get_current_plan(journey_id=journey_id, owner_id=owner)
            record = None if result is None else imports.get(result[0].route_revision_id)
        finally:
            routes.close()
            plans.close()
            imports.close()
        if result is None:
            raise LookupError(journey_id)
        route, geometry = result
        camps = (
            [marker.to_dict() for marker in plan.camp_markers]
            if plan is not None and plan.route_revision_id == route.route_revision_id
            else []
        )
        day = query.get("day", [None])[0]
        return route_file_service.export_gpx(
            title=journey.title,
            geometry=geometry,
            camps=camps,
            record=record,
            day=None if day in (None, "") else int(day),
            layout=query.get("layout", ["segments"])[0],
            link=None,
        )

    def do_GET(self):  # noqa: N802
        if self._auth_gate():
            return
        if urlsplit(self.path).path.startswith(API_PREFIX + "/"):
            self._api_v1()
            return
        parsed = urlsplit(self.path)
        if self._live_request():
            return
        if parsed.path == "/api/runtime/server":
            _send_json(self, http.HTTPStatus.OK, server_code_state())
            return
        if parsed.path == "/api/runtime/map":
            self._map_runtime_get()
            return
        if parsed.path == "/aventurro/brand.js":
            # The brand (AV-013): aventurro/brand.json, plus BRAND_NAME from
            # this server's environment when set -- see mountain_twin/brand.py.
            body = brand_script().encode("utf-8")
            self.send_response(http.HTTPStatus.OK)
            self.send_header("Content-Type", "text/javascript; charset=utf-8")
            self.send_header("Content-Length", str(len(body)))
            self.end_headers()
            self.wfile.write(body)
            return
        if parsed.path == "/api/clouds/frame":
            query = parse_qs(parsed.query)
            try:
                latitude = float(query["lat"][0])
                longitude = float(query["lon"][0])
                if not (-90 <= latitude <= 90 and -180 <= longitude <= 180):
                    raise ValueError("lat/lon out of range")
            except (KeyError, ValueError) as error:
                _send_json(self, http.HTTPStatus.BAD_REQUEST, {"error": f"lat and lon: {error}"})
                return
            _send_json(
                self, http.HTTPStatus.OK, self.server.clouds_frame_document(latitude, longitude)
            )
            return
        if parsed.path == "/api/radar/frames":
            _send_json(self, http.HTTPStatus.OK, self.server.radar_frames_document())
            return
        if parsed.path == "/api/pace-profiles":
            _send_json(self, http.HTTPStatus.OK, {"profiles": pace_profiles_document()})
            return
        if parsed.path == "/api/activities":
            _send_json(self, http.HTTPStatus.OK, {"activities": activities_document()})
            return
        if parsed.path == "/api/journeys":
            self._journey_collection_get()
            return
        if parsed.path.startswith("/api/journeys/"):
            self._journey_get(parsed.path, parse_qs(parsed.query))
            return
        if parsed.path.startswith("/api/places/"):
            self._places(parsed)
            return
        if parsed.path == "/api/usage/state":
            _send_json(self, http.HTTPStatus.OK, budgets_state_document())
            return
        if parsed.path == "/api/usage/summary":
            self._usage_summary()
            return
        if parsed.path == "/pogoda" or parsed.path.startswith("/pogoda/"):
            self._place_page()
            return
        super().do_GET()

    def end_headers(self):  # noqa: N802
        # Aventurro's static files change continuously during local
        # development; any HTTP caching (even the browser's own heuristic
        # caching of an unset-Cache-Control response) can pair a newer
        # app.js with a stale cached locale.js and reintroduce the WP0
        # "copy.<field> is undefined" crash. Never cache them.
        if urlsplit(self.path).path.startswith("/aventurro/"):
            self.send_header("Cache-Control", "no-store")
        super().end_headers()

    # --- "Pogoda" for any place (AV-062) ---------------------------------------
    def _places(self, parsed) -> None:
        """/api/places/... -- mountain_twin.places.service. There are no
        accounts yet (AV-054): with PLACE_WEATHER_ACCESS=signed_in nobody
        is signed in, so the search and the forecast answer 401."""
        body = None
        if self.command == "POST":
            length = int(self.headers.get("Content-Length") or 0)
            try:
                body = json.loads(self.rfile.read(length) or b"{}") if length else {}
            except ValueError:
                _send_json(self, http.HTTPStatus.BAD_REQUEST, {"error": "BAD_JSON"})
                return
        status, document, headers = self.server.get_place_weather().handle(
            self.command,
            parsed.path,
            parse_qs(parsed.query),
            body,
            owner_id=self.owner_id,
            client=self.real_ip,
            signed_in=self.server.accounts and self._session is not None,
        )
        payload = json.dumps(document, sort_keys=True, allow_nan=False).encode("utf-8")
        self.send_response(status)
        self.send_header("Content-Type", "application/json; charset=utf-8")
        self.send_header("Cache-Control", "no-store")
        for name, value in headers.items():
            self.send_header(name, value)
        self.send_header("Content-Length", str(len(payload)))
        self.end_headers()
        self.wfile.write(payload)

    def _usage_summary(self) -> None:
        """AV-052: the "Zużycie usług" panel's data -- for the owner only:
        this machine's own requests (the local server has no accounts; the
        public one needs AV-054's sign-in before it serves this)."""
        if not self.server.accounts and self.client_address[0] not in ("127.0.0.1", "::1"):
            _send_json(self, http.HTTPStatus.FORBIDDEN, {"error": "OWNER_ONLY"})
            return
        _send_json(self, http.HTTPStatus.OK, usage_summary_document(usage_store()))

    def _place_page(self) -> None:
        """/pogoda/<place>: the app's page, its relative addresses resolved
        from /aventurro/ -- a shared link or a reload opens the same place."""
        page = (self.server.explorer_root / "aventurro" / "index.html").read_text(encoding="utf-8")
        body = page.replace("<head>", '<head><base href="/aventurro/" />', 1).encode("utf-8")
        self.send_response(http.HTTPStatus.OK)
        self.send_header("Content-Type", "text/html; charset=utf-8")
        self.send_header("Cache-Control", "no-store")
        self.send_header("Content-Length", str(len(body)))
        self.end_headers()
        self.wfile.write(body)

    # --- Live tracking (AV-022) ---------------------------------------------
    def log_message(self, format, *args):  # noqa: A002
        # A live-tracking link's token is in its path: never write it to the log.
        if args and isinstance(args[0], str) and "/live/" in args[0]:
            args = (LIVE_TOKEN_IN_PATH.sub(r"/live/\1\2/…", args[0]),) + args[1:]
        super().log_message(format, *args)

    def _live_request(self) -> bool:
        """/live/... (the local public pages, local mode only) and
        /api/journeys/{id}/live... (the owner's panel). False: not ours."""
        path = urlsplit(self.path).path
        parts = path.strip("/").split("/")
        if len(parts) >= 4 and parts[:2] == ["api", "journeys"] and parts[3] == "live":
            self._live_owner(unquote(parts[2]), parts[4:])
            return True
        if not path.startswith("/live/"):
            return False
        surface = self.server.get_live()[1]
        if surface is None:
            _send_json(self, http.HTTPStatus.NOT_FOUND, {"error": "LIVE_PAGES_ON_PUBLIC_SERVER"})
            return True
        length = int(self.headers.get("Content-Length") or 0)
        body = self.rfile.read(length) if length else b""
        response = surface.handle(
            self.command, path, dict(self.headers.items()), body, secure=False
        )
        self.send_response(response.status)
        self.send_header("Content-Type", response.content_type)
        self.send_header("Content-Length", str(len(response.body)))
        for name, value in response.headers:
            self.send_header(name, value)
        self.end_headers()
        self.wfile.write(response.body)
        return True

    def _live_owner(self, journey_id: str, rest: list) -> None:
        """GET    /api/journeys/{id}/live                  status + links + last positions
        POST   /api/journeys/{id}/live/links/{kind}     new link (route published with it)
        DELETE /api/journeys/{id}/live/links/{kind}     revoke
        DELETE /api/journeys/{id}/live/positions        delete the position history"""
        owner = self.server.get_live()[0]
        origin = f"http://{self.headers.get('Host') or '127.0.0.1'}"
        try:
            if self.server.get_journey_catalog().is_reference(journey_id):
                raise LiveTrackingError(
                    403,
                    "REFERENCE_JOURNEY",
                    "Śledzenie na żywo jest dostępne dla zapisanych wypraw z trasą.",
                )
            route_document = self._live_route(journey_id)
            if not rest and self.command == "GET":
                revision = route_document["route_revision_id"] if route_document else None
                _send_json(self, http.HTTPStatus.OK, owner.status(journey_id, revision, origin))
                return
            if len(rest) == 2 and rest[0] == "links" and self.command == "POST":
                if route_document is None:
                    raise LiveTrackingError(409, "NO_ROUTE", "Ta wyprawa nie ma jeszcze trasy.")
                length = int(self.headers.get("Content-Length") or 0)
                payload = json.loads(self.rfile.read(length) or b"{}") if length else {}
                password = payload.get("password") if isinstance(payload, dict) else None
                issued = owner.issue(
                    journey_id,
                    rest[1],
                    password or None,
                    title=route_document["title"],
                    route_revision_id=route_document["route_revision_id"],
                    latlngs=route_document["latlngs"],
                    request_origin=origin,
                )
                _send_json(self, http.HTTPStatus.CREATED, issued)
                return
            if len(rest) == 2 and rest[0] == "links" and self.command == "DELETE":
                owner.revoke(journey_id, rest[1])
                _send_json(self, http.HTTPStatus.OK, {"revoked": rest[1]})
                return
            if rest == ["positions"] and self.command == "DELETE":
                _send_json(
                    self, http.HTTPStatus.OK, {"deleted": owner.delete_positions(journey_id)}
                )
                return
            _send_json(self, http.HTTPStatus.NOT_FOUND, {"error": "NOT_FOUND"})
        except LiveTrackingError as error:
            _send_json(self, http.HTTPStatus(error.status), error.document())
        except json.JSONDecodeError:
            _send_json(
                self,
                http.HTTPStatus.BAD_REQUEST,
                {"error": "JSON", "message": "Nieprawidłowe zapytanie."},
            )
        except KeyError:
            _send_json(
                self, http.HTTPStatus.NOT_FOUND, {"error": f"journey not found: {journey_id}"}
            )

    def _live_route(self, journey_id: str):
        """The saved Journey's title and current route as [lat, lon] pairs, or
        None when it has no route (KeyError for an unknown Journey)."""
        journey = self.server.get_journey_catalog().get_journey(self.owner_id, journey_id)
        route_repository = self.server.get_journey_route_repository()
        try:
            current = route_repository.get_current_route(
                journey_id=journey_id, owner_id=self.owner_id
            )
        finally:
            route_repository.close()
        if current is None:
            return None
        route, geometry = current
        latlngs = [[point[1], point[0]] for point in geometry["coordinates"]]
        return {
            "title": journey.title,
            "route_revision_id": route.route_revision_id,
            "latlngs": latlngs,
        }

    def _map_runtime_get(self) -> None:
        """Where the page's maps come from (AV-042), see map_runtime_config."""
        document = map_runtime_config(os.environ)
        if document is None:
            _send_json(
                self,
                http.HTTPStatus.SERVICE_UNAVAILABLE,
                {"error": "MAP_CONFIGURATION_UNAVAILABLE"},
            )
            return
        _send_json(self, http.HTTPStatus.OK, document)

    def do_PUT(self):  # noqa: N802
        """PUT /api/journeys/{id}/travel-legs/{leg_id} (AV-009): edits one
        travel leg in place -- a plain mutable row, never versioned."""
        if self._auth_gate():
            return
        if urlsplit(self.path).path.startswith(API_PREFIX + "/"):
            self._api_v1()
            return
        if self._live_request():
            return
        parts = urlsplit(self.path).path.strip("/").split("/")
        if (
            len(parts) != 5
            or parts[:2] != ["api", "journeys"]
            or parts[3] not in ("travel-legs", "calendar-events")
        ):
            self.send_error(http.HTTPStatus.NOT_FOUND)
            return
        journey_id, travel_leg_id = unquote(parts[2]), unquote(parts[4])
        try:
            length = int(self.headers.get("Content-Length", "0"))
            payload = json.loads(self.rfile.read(length))
            if parts[3] == "calendar-events":
                self._calendar_event_write(journey_id, travel_leg_id, payload)
                return
            self._journey_travel_leg_put(journey_id, travel_leg_id, payload)
        except (json.JSONDecodeError, KeyError, TypeError, ValueError) as error:
            _send_json(self, http.HTTPStatus.BAD_REQUEST, {"error": str(error)})
        except Exception as error:  # pragma: no cover - defensive local-server boundary
            self.log_error("Travel leg update failed: %s", error)
            _send_json(self, http.HTTPStatus.INTERNAL_SERVER_ERROR, {"error": str(error)})

    def do_DELETE(self):  # noqa: N802
        """DELETE /api/journeys/{id} (docs/journey_edit_delete_v0_1_design.md
        section 1): the Journey with its routes and plans; 403 for the
        immutable reference Journey, 404 for an unknown one. Also DELETE
        /api/journeys/{id}/travel-legs/{leg_id} (AV-009), and DELETE
        /api/places/favorites/{key} (AV-062)."""
        if self._auth_gate():
            return
        if urlsplit(self.path).path.startswith(API_PREFIX + "/"):
            self._api_v1()
            return
        if self._live_request():
            return
        if urlsplit(self.path).path.startswith("/api/places/"):
            self._places(urlsplit(self.path))
            return
        parts = urlsplit(self.path).path.strip("/").split("/")
        if (
            len(parts) == 5
            and parts[:2] == ["api", "journeys"]
            and parts[3] in ("travel-legs", "calendar-events")
        ):
            journey_id, travel_leg_id = unquote(parts[2]), unquote(parts[4])
            try:
                if parts[3] == "calendar-events":
                    self._calendar_event_delete(journey_id, travel_leg_id)
                    return
                self._journey_travel_leg_delete(journey_id, travel_leg_id)
            except Exception as error:  # pragma: no cover - defensive local-server boundary
                self.log_error("Travel leg deletion failed: %s", error)
                _send_json(self, http.HTTPStatus.INTERNAL_SERVER_ERROR, {"error": str(error)})
            return
        if len(parts) != 3 or parts[:2] != ["api", "journeys"]:
            self.send_error(http.HTTPStatus.NOT_FOUND)
            return
        journey_id = unquote(parts[2])
        try:
            self.server.get_journey_catalog().delete(self.owner_id, journey_id)
        except PermissionError as error:
            _send_json(self, http.HTTPStatus.FORBIDDEN, {"error": str(error)})
            return
        except KeyError:
            _send_json(
                self, http.HTTPStatus.NOT_FOUND, {"error": f"journey not found: {journey_id}"}
            )
            return
        except ValueError as error:
            _send_json(self, http.HTTPStatus.BAD_REQUEST, {"error": str(error)})
            return
        except Exception as error:  # pragma: no cover - defensive local-server boundary
            self.log_error("Journey deletion failed: %s", error)
            _send_json(
                self, http.HTTPStatus.INTERNAL_SERVER_ERROR, {"error": "JOURNEY_DELETE_FAILED"}
            )
            return
        # Live tracking keeps positions as long as the Journey exists (AV-022):
        # with it go its links, positions and published route.
        live_removed = True
        try:
            self.server.get_live()[0].delete_journey(journey_id)
        except LiveTrackingError as error:
            live_removed = False
            self.log_error("Live-tracking data of %s not removed: %s", journey_id, error.message)
        _send_json(
            self, http.HTTPStatus.OK, {"deleted": journey_id, "live_tracking_removed": live_removed}
        )

    def do_POST(self):  # noqa: N802
        if self._auth_gate():
            return
        if urlsplit(self.path).path.startswith(API_PREFIX + "/"):
            self._api_v1()
            return
        parsed = urlsplit(self.path)
        if self._live_request():
            return
        if parsed.path == "/api/journeys":
            self._journey_collection_post()
            return
        if parsed.path.startswith("/api/journeys/"):
            self._journey_post(parsed.path)
            return
        if parsed.path == "/api/day-derivation/preview":
            self._day_derivation_preview_post()
            return
        if parsed.path == "/api/routing/segment":
            self._routing_segment_post()
            return
        if parsed.path == "/api/gpx/preview":
            self._gpx_preview_post(parse_qs(parsed.query))
            return
        if parsed.path == "/api/gpx/match":
            self._gpx_match_post()
            return
        if parsed.path.startswith("/api/places/"):
            self._places(parsed)
            return
        if self.path != "/api/explorer/analyze":
            self.send_error(http.HTTPStatus.NOT_FOUND)
            return
        request_started = time.perf_counter()
        try:
            length = int(self.headers.get("Content-Length", "0"))
            payload = json.loads(self.rfile.read(length))
            start_datetime = parse_explorer_start_datetime(payload)
            preset = payload.get("preset", "NOMINAL")
            service_started = time.perf_counter()
            document = self.server.heavy.run(
                lambda: self.server.get_explorer_service().analyze_explorer_request(
                    payload.get("route_id", "tmb_day_01"), start_datetime, preset
                )
            )
            document["performance"]["request_parse_seconds"] = service_started - request_started
            document["performance"]["service_request_seconds"] = (
                time.perf_counter() - service_started
            )
            _send_json(self, http.HTTPStatus.OK, document)
        except HeavyBusy:
            _send_json(self, http.HTTPStatus.SERVICE_UNAVAILABLE, BUSY_DOCUMENT)
        except (json.JSONDecodeError, KeyError, TypeError, ValueError) as error:
            _send_json(self, http.HTTPStatus.BAD_REQUEST, {"error": str(error)})
        except Exception as error:  # pragma: no cover - defensive local-server boundary
            self.log_error("Explorer analysis failed: %s", error)
            _send_json(self, http.HTTPStatus.INTERNAL_SERVER_ERROR, {"error": str(error)})

    def _journey_get(self, path, query):
        parts = path.strip("/").split("/")
        # api / journeys / {journey_id} / bootstrap|why / [domain]
        if len(parts) < 4 or parts[:2] != ["api", "journeys"]:
            self.send_error(http.HTTPStatus.NOT_FOUND)
            return
        journey_id, action = unquote(parts[2]), parts[3]
        try:
            catalog = self.server.get_journey_catalog()
            if action == "bootstrap" and len(parts) == 4 and not catalog.is_reference(journey_id):
                _send_json(
                    self,
                    http.HTTPStatus.OK,
                    self._durable_journey_bootstrap(catalog, journey_id),
                )
                return
            if action == "camps" and len(parts) == 4:
                requested_pace = query.get("pace_profile_id", [None])[0]
                if catalog.is_reference(journey_id):
                    document = self.server.get_camps_service().get_camps_document(
                        journey_id, pace_profile_id=requested_pace or DEFAULT_PACE_PROFILE_ID
                    )
                else:
                    document = self._durable_camps_document(journey_id, requested_pace)
                _send_json(self, http.HTTPStatus.OK, document)
                return
            if action == "travel-legs" and len(parts) == 4:
                self._journey_travel_legs_get(journey_id)
                return
            if action == "calendar-events" and len(parts) == 4:
                month = parse_qs(urlsplit(self.path).query).get("month", [None])[0]
                self._calendar_events_get(journey_id, month)
                return
            if action == "telemetry" and len(parts) == 4:
                if catalog.is_reference(journey_id):
                    # The reference Journey keeps its analysed Day 1 profile;
                    # the live cockpit is for saved routes (cockpit_v0_1 §2).
                    _send_json(
                        self,
                        http.HTTPStatus.NOT_FOUND,
                        {"error": "TELEMETRY_NOT_AVAILABLE_FOR_REFERENCE_JOURNEY"},
                    )
                    return
                _send_json(
                    self, http.HTTPStatus.OK, self._durable_telemetry_document(journey_id, query)
                )
                return
            if action == "storms" and len(parts) == 4:
                if catalog.is_reference(journey_id):
                    _send_json(
                        self,
                        http.HTTPStatus.NOT_FOUND,
                        {"error": "STORMS_NOT_AVAILABLE_FOR_REFERENCE_JOURNEY"},
                    )
                    return
                _send_json(
                    self,
                    http.HTTPStatus.OK,
                    self._durable_telemetry_document(journey_id, query, storms=True),
                )
                return
            if action == "walking-times" and len(parts) == 4:
                if catalog.is_reference(journey_id):
                    _send_json(
                        self,
                        http.HTTPStatus.NOT_FOUND,
                        {"error": "WEATHER_SERIES_NOT_AVAILABLE_FOR_REFERENCE_JOURNEY"},
                    )
                    return
                _send_json(
                    self,
                    http.HTTPStatus.OK,
                    self._weather_series_document(journey_id, query, walking_only=True),
                )
                return
            if action == "weather-series" and len(parts) == 4:
                if catalog.is_reference(journey_id):
                    # Like telemetry: the reference Journey keeps its analysed
                    # Day 1 and has no live weather series.
                    _send_json(
                        self,
                        http.HTTPStatus.NOT_FOUND,
                        {"error": "WEATHER_SERIES_NOT_AVAILABLE_FOR_REFERENCE_JOURNEY"},
                    )
                    return
                _send_json(
                    self, http.HTTPStatus.OK, self._weather_series_document(journey_id, query)
                )
                # AV-048: the prefetch keeps recently opened Journeys warm.
                self.server.get_forecast_cache().note_opened(journey_id)
                return
            if action == "terrain-light" and len(parts) == 4:
                _send_json(
                    self,
                    http.HTTPStatus.OK,
                    self._terrain_light_document(catalog, journey_id, query),
                )
                return
            if action == "profile" and len(parts) == 4:
                if catalog.is_reference(journey_id):
                    document = self._reference_stage_profile(query.get("route_id", [None])[0])
                else:
                    document = self._durable_profile_document(journey_id)
                _send_json(self, http.HTTPStatus.OK, document)
                return
            if action in ("gpx", "gpx-options") and len(parts) == 4:
                self._journey_gpx_get(catalog, journey_id, action, query)
                return
            if action == "route" and len(parts) == 4:
                route_repository = self.server.get_journey_route_repository()
                try:
                    result = route_repository.get_current_route(
                        journey_id=journey_id, owner_id=self.owner_id
                    )
                    anchors = route_repository.get_current_anchor_point_indexes(
                        journey_id=journey_id, owner_id=self.owner_id
                    )
                    spans = route_repository.get_current_surface_spans(
                        journey_id=journey_id, owner_id=self.owner_id
                    )
                finally:
                    route_repository.close()
                if result is None:
                    _send_json(self, http.HTTPStatus.OK, {"route": None})
                else:
                    route, geometry = result
                    imports = RouteImportRepository(JourneySqliteDatabase(self.server.journey_db))
                    try:
                        record = imports.get(route.route_revision_id)
                    finally:
                        imports.close()
                    _send_json(
                        self,
                        http.HTTPStatus.OK,
                        {
                            "route": route_revision_document(route),
                            "geometry_geojson": geometry,
                            "anchor_point_indexes": None if anchors is None else list(anchors),
                            "surface": route_surface_document(
                                route_points_from_geometry_geojson(
                                    route.route_revision_id, geometry
                                ),
                                spans,
                            ),
                            # AV-053: where an imported route came from.
                            "import": record,
                        },
                    )
                return
            if action == "plan" and len(parts) == 4:
                plan_repository = self.server.get_journey_plan_repository()
                try:
                    plan = plan_repository.get_current_plan(
                        journey_id=journey_id, owner_id=self.owner_id
                    )
                finally:
                    plan_repository.close()
                _send_json(
                    self,
                    http.HTTPStatus.OK,
                    {"plan": None if plan is None else journey_plan_document(plan)},
                )
                return
            service = self.server.get_journey_service()
            scenario_id = query.get("scenario_id", [None])[0]
            if scenario_id is None:
                run = service.get_run(journey_id)
                scenarios = service.repository.scenarios_for_run(run.analysis_run_id)
                scenario_id = next(
                    (item.analysis_scenario_id for item in scenarios if item.name == "NOMINAL"),
                    scenarios[0].analysis_scenario_id,
                )
            if action == "bootstrap" and len(parts) == 4:
                _send_json(
                    self, http.HTTPStatus.OK, service.get_product_bootstrap(journey_id, scenario_id)
                )
                return
            if action == "why" and len(parts) == 5:
                _send_json(
                    self,
                    http.HTTPStatus.OK,
                    service.get_why_detail(journey_id, scenario_id, parts[4]),
                )
                return
            self.send_error(http.HTTPStatus.NOT_FOUND)
        except HeavyBusy:
            _send_json(self, http.HTTPStatus.SERVICE_UNAVAILABLE, BUSY_DOCUMENT)
        except DayDerivationUnavailable as unavailable:
            # Telemetry/storms for planned hours of a route that cannot be
            # timed: an explicit, readable state -- not a malformed request.
            _send_json(
                self, http.HTTPStatus.UNPROCESSABLE_ENTITY, _plan_timing_unavailable(unavailable)
            )
        except (KeyError, TypeError, ValueError) as error:
            self.log_error("Journey request rejected: %r", error)
            _send_json(self, http.HTTPStatus.BAD_REQUEST, {"error": str(error)})
        except Exception as error:  # pragma: no cover - defensive local-server boundary
            self.log_error("Journey request failed: %s", error)
            _send_json(self, http.HTTPStatus.INTERNAL_SERVER_ERROR, {"error": str(error)})

    def _durable_journey_bootstrap(self, catalog, journey_id):
        """Route persistence v0.1 Work Package 3: a durable (non-reference)
        Journey bootstraps as EMPTY_JOURNEY until route_persistence.py (WP1)
        has given it a route, then as GEOMETRY_ONLY_JOURNEY (real geometry
        and distance/elevation facts, every intelligence domain explicitly
        unavailable -- journey_analysis_runs stays out of scope). Uses the
        same catalog.is_reference() split already established for
        EMPTY_JOURNEY; this only refines what happens on its non-reference
        side, per docs/route_persistence_v0_1_design.md."""
        route_repository = self.server.get_journey_route_repository()
        try:
            route_result = route_repository.get_current_route(
                journey_id=journey_id, owner_id=self.owner_id
            )
        finally:
            route_repository.close()
        if route_result is None:
            return catalog.empty_bootstrap(self.owner_id, journey_id)
        route, geometry = route_result

        plan_repository = self.server.get_journey_plan_repository()
        try:
            plan = plan_repository.get_current_plan(journey_id=journey_id, owner_id=self.owner_id)
        finally:
            plan_repository.close()

        journey = catalog.get_journey(self.owner_id, journey_id)
        return geometry_only_bootstrap_document(
            journey=journey, route=route, geometry_geojson=geometry, plan=plan
        )

    def _durable_camps_document(self, journey_id: str, pace_profile_id: str | None) -> dict:
        """Route persistence v0.1 Work Package 4: the durable-Journey
        counterpart to CampsService.get_camps_document(), for a Journey
        with a route (WP1) and a plan with camp markers (WP2). A Journey
        with no route or no plan yet has no camps to derive -- returns the
        same empty-but-valid shape rather than a 400, matching the
        explicit missing-data philosophy elsewhere in this codebase (no
        camps yet is not an error)."""
        route_repository = self.server.get_journey_route_repository()
        try:
            route_result = route_repository.get_current_route(
                journey_id=journey_id, owner_id=self.owner_id
            )
        finally:
            route_repository.close()
        plan = None
        if route_result is not None:
            plan_repository = self.server.get_journey_plan_repository()
            try:
                plan = plan_repository.get_current_plan(
                    journey_id=journey_id, owner_id=self.owner_id
                )
            finally:
                plan_repository.close()
        if route_result is None or plan is None:
            return {
                "journey_id": journey_id,
                "pace_profile_id": pace_profile_id or DEFAULT_PACE_PROFILE_ID,
                "day_segments": [],
                "camps": [],
            }
        route, geometry = route_result
        # Without an explicit ?pace_profile_id, the plan's own saved pace --
        # otherwise Mission Control (which never passes one) would silently
        # re-time a durable Journey at the default pace, not the one it was
        # planned and saved with.
        return durable_camps_document(
            journey_id=journey_id,
            route=route,
            geometry_geojson=geometry,
            plan=plan,
            pace_profile_id=pace_profile_id or plan.pace_policy_id,
            live_resolver=self.server.get_live_weather_resolver(),
        )

    def _durable_profile_document(self, journey_id: str) -> dict:
        """docs/profile_storms_v0_1_design.md section 2: the saved route's
        elevation profile with its server-side grade, its planned days and
        camps (no plan: the route alone)."""
        route_repository = self.server.get_journey_route_repository()
        try:
            route_result = route_repository.get_current_route(
                journey_id=journey_id, owner_id=self.owner_id
            )
        finally:
            route_repository.close()
        if route_result is None:
            raise ValueError("a profile requires a saved route")
        route, geometry = route_result
        route_points = route_points_from_geometry_geojson(route.route_revision_id, geometry)
        plan_repository = self.server.get_journey_plan_repository()
        try:
            plan = plan_repository.get_current_plan(journey_id=journey_id, owner_id=self.owner_id)
        finally:
            plan_repository.close()
        days, camps, plan_timing = (), (), None
        if plan is not None and plan.planned_start_local is None:
            # A plan without a start (migration 004): its camps and its days
            # by place (step T3) -- the profile's day split needs no clock
            # time; plan_timing says the clock times are what is missing.
            camps = plan.camp_markers
            plan_timing = {
                "state": "UNAVAILABLE",
                "reason_codes": [NO_PLANNED_START],
                "unavailable_elevation_points": 0,
            }
            prepared = prepare_route(route_points).points
            try:
                days = derive_relative_day_segments(
                    route_points,
                    pace_factor=pace_factor_for_activity(route.activity_id, plan.pace_policy_id),
                    pauses=generate_relative_pauses(prepared[-1].cumulative_distance_m),
                    camp_markers=plan.camp_markers,
                )
            except DayDerivationUnavailable as unavailable:
                plan_timing = _plan_timing_unavailable(unavailable)
        elif plan is not None:
            prepared = prepare_route(route_points).points
            camps = plan.camp_markers
            try:
                days = derive_day_segments(
                    route_points,
                    pace_factor=pace_factor_for_activity(route.activity_id, plan.pace_policy_id),
                    pauses=generate_relative_pauses(prepared[-1].cumulative_distance_m),
                    start_datetime=datetime.fromisoformat(plan.planned_start_local),
                    camp_markers=plan.camp_markers,
                    timezone_name=plan.journey_timezone,
                    rest_days_before_start=plan.rest_days_before_start,
                )
            except DayDerivationUnavailable as unavailable:
                # The route alone (no day split): its plan cannot be timed.
                plan_timing = _plan_timing_unavailable(unavailable)
        if (
            plan is not None
            and not days
            and not activity_for(route.activity_id).pace_model_available
        ):
            # AV-032: a bike's days are its camps' places, without a pace.
            days = derive_untimed_day_segments(route_points, camp_markers=plan.camp_markers)
        document = route_profile_document(
            route_points, days=days, camps=camps, elevation_source="BROUTER"
        )
        document["journey_id"] = journey_id
        if plan_timing is not None:
            document["plan_timing"] = plan_timing
        return document

    def _reference_stage_profile(self, route_id: str | None) -> dict:
        """The reference Journey's preserved stage geometry (one TMB day at
        a time, ?route_id=tmb_day_NN) with the same server-side grade."""
        document = route_profile_document(self._reference_stage_points(route_id))
        document["route_id"] = route_id
        return document

    def _reference_stage_points(self, route_id: str | None) -> list[RoutePoint]:
        stages = {
            stage.route_id: stage for stage in load_tmb_stage_summaries(self.server.explorer_root)
        }
        if route_id not in stages:
            raise KeyError(f"unknown stage: {route_id}")
        path = self.server.explorer_root / stages[route_id].geometry_path
        coordinates = json.loads(path.read_text(encoding="utf-8"))["features"][0]["geometry"][
            "coordinates"
        ]
        return [
            RoutePoint(
                route_id=route_id,
                point_index=index,
                track_index=0,
                segment_index=0,
                latitude=coordinate[1],
                longitude=coordinate[0],
                elevation_m=coordinate[2] if len(coordinate) > 2 else None,
            )
            for index, coordinate in enumerate(coordinates)
        ]

    def _terrain_light_document(self, catalog, journey_id: str, query) -> dict:
        """docs/design_reference/terrain3d_sun_spike_v0_1.md: the sun and the
        terrain's cast shadow around the route at ?time=YYYY-MM-DDTHH:MM (the
        route's own local time) -- a saved Journey's current route, or the
        reference Journey's stage ?route_id=tmb_day_NN, all its stages
        without one (read only). Computed once per route and moment, then
        served from memory."""
        time_text = query.get("time", [None])[0]
        if not time_text:
            raise ValueError("terrain light needs ?time=YYYY-MM-DDTHH:MM")
        if catalog.is_reference(journey_id):
            route_key = query.get("route_id", [None])[0]
            if route_key is None:
                route_points = [
                    point
                    for stage in load_tmb_stage_summaries(self.server.explorer_root)
                    for point in self._reference_stage_points(stage.route_id)
                ]
            else:
                route_points = self._reference_stage_points(route_key)
        else:
            route_repository = self.server.get_journey_route_repository()
            try:
                result = route_repository.get_current_route(
                    journey_id=journey_id, owner_id=self.owner_id
                )
            finally:
                route_repository.close()
            if result is None:
                raise ValueError("terrain light requires a saved route")
            route, geometry = result
            route_key = route.route_revision_id
            route_points = route_points_from_geometry_geojson(route.route_revision_id, geometry)
        timezone_name = resolve_route_timezone(route_points[0].latitude, route_points[0].longitude)
        moment = datetime.fromisoformat(time_text)
        if moment.tzinfo is not None:
            raise ValueError("terrain light time is the route's local time, without an offset")
        moment = moment.replace(second=0, microsecond=0, tzinfo=ZoneInfo(timezone_name))
        key = (journey_id, route_key, moment.isoformat())
        cache = self.server.terrain_light_cache
        if key in cache:
            return cache[key]
        document = self.server.heavy.run(
            lambda: terrain_light_document(
                journey_id=journey_id,
                route_points=route_points,
                moment=moment,
                timezone_name=timezone_name,
                dem_provider=self.server.get_terrain_dem_provider(),
            )
        )
        # A failed terrain download is not remembered: the next request asks again.
        if document["shadow"]["state"] != "UNAVAILABLE":
            with self.server.terrain_light_lock:
                if len(cache) >= 96:
                    cache.pop(next(iter(cache)))
                cache[key] = document
        return document

    def _durable_telemetry_document(self, journey_id: str, query, storms: bool = False) -> dict:
        """docs/cockpit_v0_1_design.md section 3: the planned day window when
        the Journey has a plan (?first_day=&last_day=, 0-based, default all
        days), otherwise the next 72 hours along the whole route."""
        route_repository = self.server.get_journey_route_repository()
        try:
            route_result = route_repository.get_current_route(
                journey_id=journey_id, owner_id=self.owner_id
            )
        finally:
            route_repository.close()
        if route_result is None:
            raise ValueError("telemetry requires a saved route")
        route, geometry = route_result
        route_points = route_points_from_geometry_geojson(route.route_revision_id, geometry)
        plan_repository = self.server.get_journey_plan_repository()
        try:
            plan = plan_repository.get_current_plan(journey_id=journey_id, owner_id=self.owner_id)
        finally:
            plan_repository.close()

        def optional_int(name):
            value = query.get(name, [None])[0]
            return None if value in (None, "") else int(value)

        resolver = self.server.get_live_weather_resolver()
        # A plan without a start (migration 004) has no planned hours: the
        # tabs show the next 72 hours, as for a Journey without a plan --
        # along the tree's days when a range is asked for (step T3: its
        # days are places on the route, not dates).
        # AV-032: a route without a pace model (a bike) has no planned hours
        # either: like a plan without a start, the next 72 hours, its days
        # as places on the route.
        timed = activity_for(route.activity_id).pace_model_available
        # AV-035: planned hours follow the start time, on the real dates only
        # when they reach into the forecast (weather_anchor).
        if timed:
            plan, _ = anchored_plan(plan, route_points, route.activity_id)
        timed_plan = (
            plan if plan is not None and plan.planned_start_local is not None and timed else None
        )
        ranged = {}
        first_day, last_day = optional_int("first_day"), optional_int("last_day")
        if plan is not None and timed_plan is None and (first_day, last_day) != (None, None):
            prepared = prepare_route(route_points).points
            relative = (
                derive_relative_day_segments(
                    route_points,
                    pace_factor=pace_factor_for_activity(route.activity_id, plan.pace_policy_id),
                    pauses=generate_relative_pauses(prepared[-1].cumulative_distance_m),
                    camp_markers=plan.camp_markers,
                )
                if timed
                else derive_untimed_day_segments(route_points, camp_markers=plan.camp_markers)
            )
            first = 0 if first_day is None else first_day
            last = len(relative) - 1 if last_day is None else last_day
            if not (0 <= first <= last < len(relative)):
                raise ValueError("telemetry day range is outside the plan's days")
            ranged = {
                "point_range": (relative[first].start_point_index, relative[last].end_point_index),
                "day_range": (first, last),
            }
        if storms:
            # docs/profile_storms_v0_1_design.md section 4: the same windows
            # and samples for the convective variables and weather code.
            if timed_plan is not None:
                return storms_document(
                    planned=True,
                    journey_id=journey_id,
                    route_points=route_points,
                    plan=timed_plan,
                    pace_profile_id=timed_plan.pace_policy_id,
                    live_resolver=resolver,
                    first_day=optional_int("first_day"),
                    last_day=optional_int("last_day"),
                    activity_id=route.activity_id,
                )
            return storms_document(
                planned=False,
                journey_id=journey_id,
                route_points=route_points,
                timezone_name=resolve_route_timezone(
                    route_points[0].latitude, route_points[0].longitude
                ),
                live_resolver=resolver,
                **ranged,
            )
        if timed_plan is not None:
            return planned_telemetry_document(
                journey_id=journey_id,
                route_points=route_points,
                plan=timed_plan,
                pace_profile_id=timed_plan.pace_policy_id,
                live_resolver=resolver,
                first_day=optional_int("first_day"),
                last_day=optional_int("last_day"),
                activity_id=route.activity_id,
            )
        return next_hours_telemetry_document(
            journey_id=journey_id,
            route_points=route_points,
            timezone_name=resolve_route_timezone(
                route_points[0].latitude, route_points[0].longitude
            ),
            live_resolver=resolver,
            **ranged,
        )

    def _weather_series_document(self, journey_id: str, query, walking_only: bool = False) -> dict:
        return weather_series_for(self.server, journey_id, query, walking_only)

    def _journey_collection_get(self):
        try:
            _send_json(
                self,
                http.HTTPStatus.OK,
                {"journeys": self.server.get_journey_catalog().list(self.owner_id)},
            )
        except Exception as error:  # pragma: no cover - defensive local-server boundary
            self.log_error("Journey list failed: %s", error)
            _send_json(
                self, http.HTTPStatus.INTERNAL_SERVER_ERROR, {"error": "JOURNEY_LIST_FAILED"}
            )

    def _journey_collection_post(self):
        try:
            length = int(self.headers.get("Content-Length", "0"))
            payload = json.loads(self.rfile.read(length))
            if not isinstance(payload, dict):
                raise ValueError("Journey payload must be an object")
            _send_json(
                self,
                http.HTTPStatus.CREATED,
                self.server.get_journey_catalog().create(self.owner_id, payload),
            )
        except (json.JSONDecodeError, TypeError, ValueError) as error:
            _send_json(self, http.HTTPStatus.BAD_REQUEST, {"error": str(error)})
        except Exception as error:  # pragma: no cover - defensive local-server boundary
            self.log_error("Journey creation failed: %s", error)
            _send_json(
                self, http.HTTPStatus.INTERNAL_SERVER_ERROR, {"error": "JOURNEY_CREATE_FAILED"}
            )

    def _journey_post(self, path):
        parts = path.strip("/").split("/")
        if (
            len(parts) != 4
            or parts[:2] != ["api", "journeys"]
            or parts[3]
            not in (
                "selection",
                "camps",
                "route",
                "plan",
                "travel-legs",
                "rest-days",
                "completion",
                "calendar-events",
                "day-one",
            )
        ):
            self.send_error(http.HTTPStatus.NOT_FOUND)
            return
        journey_id = unquote(parts[2])
        try:
            length = int(self.headers.get("Content-Length", "0"))
            payload = json.loads(self.rfile.read(length))
            if parts[3] == "travel-legs":
                self._journey_travel_leg_post(journey_id, payload)
                return
            if parts[3] == "calendar-events":
                self._calendar_event_write(journey_id, None, payload)
                return
            if parts[3] == "day-one":
                self._journey_day_one_post(journey_id, payload)
                return
            if parts[3] == "selection":
                anchor = RouteSelectionAnchor(
                    payload["route_revision_id"],
                    payload["analysis_scenario_id"],
                    route_point_index=payload.get("route_point_index"),
                    route_distance_m=payload.get("route_distance_m"),
                    planned_arrival=payload.get("planned_arrival"),
                )
                _send_json(
                    self,
                    http.HTTPStatus.OK,
                    self.server.get_journey_service().get_selection_detail(journey_id, anchor),
                )
                return
            if parts[3] == "route":
                self._journey_route_post(journey_id, payload)
                return
            if parts[3] == "plan":
                self._journey_plan_post(journey_id, payload)
                return
            if parts[3] == "rest-days":
                self._journey_rest_days_post(journey_id, payload)
                return
            if parts[3] == "completion":
                self._journey_completion_post(journey_id, payload)
                return
            document = self.server.get_camps_service().add_camp(
                journey_id,
                route_point_index=int(payload["route_point_index"]),
                label=payload["label"],
                pace_profile_id=payload.get("pace_profile_id", DEFAULT_PACE_PROFILE_ID),
            )
            _send_json(self, http.HTTPStatus.OK, document)
        except (json.JSONDecodeError, KeyError, TypeError, ValueError) as error:
            _send_json(self, http.HTTPStatus.BAD_REQUEST, {"error": str(error)})
        except Exception as error:  # pragma: no cover - defensive local-server boundary
            self.log_error("Journey selection/camp/route/plan request failed: %s", error)
            _send_json(self, http.HTTPStatus.INTERNAL_SERVER_ERROR, {"error": str(error)})

    def _journey_route_post(self, journey_id, payload):
        raw_anchors = payload.get("anchor_point_indexes")
        # AV-053: a route from a GPX/KML/TCX file names its preview.
        route_import = route_file_service.import_request(payload.get("import"))
        command = CreateRouteRevision(
            owner_id=self.owner_id,
            journey_id=journey_id,
            route_name=payload["route_name"],
            points=route_points_from_payload(payload["points"]),
            anchor_point_indexes=None if raw_anchors is None else tuple(raw_anchors),
            # AV-031: absent for a client that names no activity -> trekking.
            activity_id=payload.get("activity_id"),
            point_surfaces=payload.get("point_surfaces"),
            # AV-064: the transfer spans (dojazd), absent = all movement.
            segments=tuple(payload.get("segments") or ()),
        )
        route_repository = self.server.get_journey_route_repository()
        try:
            try:
                route = route_repository.create_route_revision(command)
            except KeyError as error:
                _send_json(
                    self, http.HTTPStatus.NOT_FOUND, {"error": f"journey not found: {error}"}
                )
                return
        finally:
            route_repository.close()
        if route_import is not None:
            imports = RouteImportRepository(JourneySqliteDatabase(self.server.journey_db))
            try:
                imports.record(
                    route_revision_id=route.route_revision_id,
                    pending=self.server.pending_imports.get(
                        route_import["import_id"], owner=self.owner_id
                    ),
                    mode=route_import["mode"],
                    matched=route_import["matched"],
                    points_of_interest=route_import["points_of_interest"],
                    geometry_fingerprint=route.geometry_fingerprint,
                    file_name=route_import["file_name"],
                )
            finally:
                imports.close()
        _send_json(self, http.HTTPStatus.CREATED, {"route": route_revision_document(route)})

    def _gpx_preview_post(self, query):
        """AV-053: a route file's preview (route_file_service.preview)."""
        length = int(self.headers.get("Content-Length", "0") or 0)
        if length > MAX_FILE_BYTES:
            self.close_connection = True  # the body stays unread
            _send_json(
                self,
                http.HTTPStatus.REQUEST_ENTITY_TOO_LARGE,
                {
                    "error": f"Plik jest za duży ({length / 1048576:.0f} MB); największy, jaki "
                    f"wczytamy, ma {MAX_FILE_BYTES // 1048576} MB.",
                    "code": "TOO_LARGE",
                },
            )
            return
        try:
            data = self.rfile.read(length)
            status, document = route_file_service.preview(
                data,
                route_file_service.safe_file_name(query.get("file_name", [""])[0]),
                self.server.pending_imports,
                self.server.get_terrain_dem_provider(),
                owner=self.owner_id,
            )
            _send_json(self, http.HTTPStatus(status), document)
        except Exception as error:  # pragma: no cover - defensive local-server boundary
            self.log_error("GPX preview failed: %s", error)
            _send_json(
                self,
                http.HTTPStatus.INTERNAL_SERVER_ERROR,
                {"error": "Nie udało się wczytać pliku.", "code": "IMPORT_FAILED"},
            )

    def _gpx_match_post(self):
        """AV-053: „Dopasuj do szlaków” (route_file_service.match)."""
        try:
            length = int(self.headers.get("Content-Length", "0"))
            payload = json.loads(self.rfile.read(length))
            status, document = route_file_service.match(
                payload,
                self.server.pending_imports,
                self.server.routing_opener,
                self.server.get_terrain_dem_provider(),
                owner=self.owner_id,
            )
            _send_json(self, http.HTTPStatus(status), document)
        except (json.JSONDecodeError, KeyError, TypeError, ValueError) as error:
            _send_json(self, http.HTTPStatus.BAD_REQUEST, {"error": str(error)})
        except Exception as error:  # pragma: no cover - defensive local-server boundary
            self.log_error("GPX match failed: %s", error)
            _send_json(self, http.HTTPStatus.INTERNAL_SERVER_ERROR, {"error": str(error)})

    def _journey_gpx_get(self, catalog, journey_id, action, query):
        """AV-053: GET /api/journeys/{id}/gpx-options and .../gpx."""
        if catalog.is_reference(journey_id):
            _send_json(self, http.HTTPStatus.NOT_FOUND, {"error": "GPX_FOR_SAVED_JOURNEYS_ONLY"})
            return
        owner = self.owner_id
        try:
            journey = catalog.get_journey(owner, journey_id)
        except KeyError:
            _send_json(self, http.HTTPStatus.NOT_FOUND, {"error": "journey not found"})
            return
        routes = self.server.get_journey_route_repository()
        plans = self.server.get_journey_plan_repository()
        imports = RouteImportRepository(JourneySqliteDatabase(self.server.journey_db))
        try:
            result = routes.get_current_route(journey_id=journey_id, owner_id=owner)
            plan = plans.get_current_plan(journey_id=journey_id, owner_id=owner)
            record = (
                None
                if result is None
                else imports.get(
                    result[0].route_revision_id, with_bytes=query.get("original") == ["1"]
                )
            )
        finally:
            routes.close()
            plans.close()
            imports.close()
        if result is None:
            _send_json(
                self, http.HTTPStatus.NOT_FOUND, {"error": "Ta wyprawa nie ma jeszcze trasy."}
            )
            return
        route, geometry = result
        camps = (
            [marker.to_dict() for marker in plan.camp_markers]
            if plan is not None and plan.route_revision_id == route.route_revision_id
            else []
        )
        if action == "gpx-options":
            _send_json(
                self,
                http.HTTPStatus.OK,
                route_file_service.export_options(geometry=geometry, camps=camps, record=record),
            )
            return
        if query.get("original") == ["1"]:
            if not record or record.get("source_bytes") is None:
                _send_json(self, http.HTTPStatus.NOT_FOUND, {"error": "ORIGINAL_NOT_KEPT"})
                return
            body, file_name = record["source_bytes"], record["file_name"]
        else:
            try:
                day = query.get("day", [None])[0]
                body, file_name = route_file_service.export_gpx(
                    title=journey.title,
                    geometry=geometry,
                    camps=camps,
                    record=record,
                    day=None if day in (None, "") else int(day),
                    layout=query.get("layout", ["segments"])[0],
                    link=f"http://{self.headers.get('Host', '127.0.0.1')}/aventurro/"
                    f"?journey_id={quote(journey_id)}",
                )
            except ValueError as error:
                _send_json(self, http.HTTPStatus.BAD_REQUEST, {"error": str(error)})
                return
        self.send_response(http.HTTPStatus.OK)
        self.send_header("Content-Type", "application/gpx+xml")
        self.send_header(
            "Content-Disposition",
            f'attachment; filename="{route_file_service.ascii_file_name(file_name)}"; '
            f"filename*=UTF-8''{quote(file_name)}",
        )
        self.send_header("Cache-Control", "no-store")
        self.send_header("Content-Length", str(len(body)))
        self.end_headers()
        self.wfile.write(body)

    def _journey_plan_post(self, journey_id, payload):
        command = CreateJourneyPlan(
            owner_id=self.owner_id,
            journey_id=journey_id,
            pace_profile_id=payload["pace_profile_id"],
            journey_timezone=payload.get("journey_timezone"),
            # Absent or null: a plan without a start (migration 004).
            planned_start_local=payload.get("planned_start_local"),
            # AV-035: a start time without a date ("HH:MM").
            planned_start_time=payload.get("planned_start_time"),
            activity_type=payload.get("activity_type"),
            camp_markers=camp_marker_inputs_from_payload(payload.get("camp_markers", [])),
            # Absent: the rest days at the trip's two ends stay as the current
            # plan has them (CreateJourneyPlan) -- a plan edit about something
            # else does not have to resend them to keep them.
            rest_days_before_start=payload.get("rest_days_before_start"),
            rest_days_after_finish=payload.get("rest_days_after_finish"),
        )
        plan_repository = self.server.get_journey_plan_repository()
        try:
            try:
                plan = plan_repository.create_plan(command)
            except KeyError as error:
                _send_json(
                    self, http.HTTPStatus.NOT_FOUND, {"error": f"journey not found: {error}"}
                )
                return
        finally:
            plan_repository.close()
        _send_json(self, http.HTTPStatus.CREATED, {"plan": journey_plan_document(plan)})

    def _journey_rest_days_post(self, journey_id, payload):
        """POST /api/journeys/{id}/rest-days {after_day_number, days}: the
        number of rest days after that walking day (0: before day 1) becomes
        ``days`` -- a new plan version copied from the current one on the
        server (docs/design_reference/rest_days_spike_v0_1.md section 7.1,
        WP2), so the Kalendarz tab never re-sends a whole plan to change one
        number. ``days`` is the resulting number, not an increment."""
        plan_repository = self.server.get_journey_plan_repository()
        try:
            try:
                plan = plan_repository.set_rest_days(
                    journey_id=journey_id,
                    owner_id=self.owner_id,
                    after_day_number=payload["after_day_number"],
                    days=payload["days"],
                )
            except KeyError as error:
                if error.args and error.args[0] == journey_id:
                    _send_json(
                        self, http.HTTPStatus.NOT_FOUND, {"error": f"journey not found: {error}"}
                    )
                    return
                raise
        finally:
            plan_repository.close()
        _send_json(self, http.HTTPStatus.CREATED, {"plan": journey_plan_document(plan)})

    def _journey_completion_post(self, journey_id, payload):
        """POST /api/journeys/{id}/completion {completed: true|false} (AV-027):
        the owner marks the Journey as done ("Oznacz jako zrealizowaną") or
        takes it back ("Cofnij do nadchodzących"). Never set automatically.
        403 for the immutable reference Journey, 404 for an unknown one."""
        try:
            journey = self.server.get_journey_catalog().set_completed(
                self.owner_id, journey_id, payload["completed"]
            )
        except PermissionError as error:
            _send_json(self, http.HTTPStatus.FORBIDDEN, {"error": str(error)})
            return
        except KeyError as error:
            if error.args and error.args[0] == journey_id:
                _send_json(
                    self, http.HTTPStatus.NOT_FOUND, {"error": f"journey not found: {journey_id}"}
                )
                return
            raise
        _send_json(self, http.HTTPStatus.OK, {"journey": journey})

    def _journey_travel_legs_get(self, journey_id):
        """GET /api/journeys/{id}/travel-legs (AV-009): the Kalendarz tab's
        manually-added travel legs, oldest departure first. A Journey with
        none yet gets an empty list, never a 404 -- no legs is not an error."""
        repository = self.server.get_travel_leg_repository()
        try:
            try:
                legs = repository.list_for_journey(journey_id=journey_id, owner_id=self.owner_id)
            except KeyError as error:
                _send_json(
                    self, http.HTTPStatus.NOT_FOUND, {"error": f"journey not found: {error}"}
                )
                return
        finally:
            repository.close()
        _send_json(self, http.HTTPStatus.OK, {"travel_legs": [leg.to_dict() for leg in legs]})

    def _journey_travel_leg_post(self, journey_id, payload):
        fields = travel_leg_fields_from_payload(payload)
        repository = self.server.get_travel_leg_repository()
        try:
            try:
                leg = repository.create(
                    journey_id=journey_id, owner_id=self.owner_id, fields=fields
                )
            except KeyError as error:
                _send_json(
                    self, http.HTTPStatus.NOT_FOUND, {"error": f"journey not found: {error}"}
                )
                return
        finally:
            repository.close()
        _send_json(self, http.HTTPStatus.CREATED, {"travel_leg": leg.to_dict()})

    def _journey_travel_leg_put(self, journey_id, travel_leg_id, payload):
        fields = travel_leg_fields_from_payload(payload)
        repository = self.server.get_travel_leg_repository()
        try:
            try:
                leg = repository.update(
                    travel_leg_id=travel_leg_id,
                    journey_id=journey_id,
                    owner_id=self.owner_id,
                    fields=fields,
                )
            except KeyError as error:
                _send_json(
                    self, http.HTTPStatus.NOT_FOUND, {"error": f"travel leg not found: {error}"}
                )
                return
        finally:
            repository.close()
        _send_json(self, http.HTTPStatus.OK, {"travel_leg": leg.to_dict()})

    # --- AV-038: the calendar's lodging/other events and Day 1 ------------
    def _calendar_events_get(self, journey_id, month):
        """GET /api/journeys/{id}/calendar-events[?month=YYYY-MM]: lodging and
        other events, earliest first; with ``month`` only those touching it
        (an event across a month boundary belongs to both months)."""
        repository = self.server.get_calendar_event_repository()
        try:
            try:
                events = repository.list_for_journey(
                    journey_id=journey_id, owner_id=self.owner_id, month=month
                )
            except KeyError as error:
                _send_json(
                    self, http.HTTPStatus.NOT_FOUND, {"error": f"journey not found: {error}"}
                )
                return
            except ValueError as error:
                _send_json(self, http.HTTPStatus.BAD_REQUEST, {"error": str(error)})
                return
        finally:
            repository.close()
        _send_json(self, http.HTTPStatus.OK, {"calendar_events": [e.to_dict() for e in events]})

    def _calendar_event_write(self, journey_id, event_id, payload):
        """POST (event_id None) creates, PUT edits one event in place."""
        fields = calendar_event_fields_from_payload(payload)
        repository = self.server.get_calendar_event_repository()
        try:
            try:
                if event_id is None:
                    event = repository.create(
                        journey_id=journey_id, owner_id=self.owner_id, fields=fields
                    )
                else:
                    event = repository.update(
                        event_id=event_id,
                        journey_id=journey_id,
                        owner_id=self.owner_id,
                        fields=fields,
                    )
            except KeyError as error:
                _send_json(self, http.HTTPStatus.NOT_FOUND, {"error": f"not found: {error}"})
                return
        finally:
            repository.close()
        status = http.HTTPStatus.CREATED if event_id is None else http.HTTPStatus.OK
        _send_json(self, status, {"calendar_event": event.to_dict()})

    def _calendar_event_delete(self, journey_id, event_id):
        repository = self.server.get_calendar_event_repository()
        try:
            try:
                repository.delete(event_id=event_id, journey_id=journey_id, owner_id=self.owner_id)
            except KeyError as error:
                _send_json(self, http.HTTPStatus.NOT_FOUND, {"error": f"not found: {error}"})
                return
        finally:
            repository.close()
        _send_json(self, http.HTTPStatus.OK, {"deleted": event_id})

    def _journey_day_one_post(self, journey_id, payload):
        """POST /api/journeys/{id}/day-one {date}: Day 1 falls on ``date`` --
        a new plan version with only its start moved (plan_persistence
        set_day_one); every later day follows."""
        if self.server.get_journey_catalog().is_reference(journey_id):
            _send_json(self, http.HTTPStatus.FORBIDDEN, {"error": "reference journey is immutable"})
            return
        plan_repository = self.server.get_journey_plan_repository()
        try:
            try:
                plan = plan_repository.set_day_one(
                    journey_id=journey_id, owner_id=self.owner_id, date=payload["date"]
                )
            except KeyError as error:
                if error.args and error.args[0] == journey_id:
                    _send_json(
                        self, http.HTTPStatus.NOT_FOUND, {"error": f"journey not found: {error}"}
                    )
                    return
                raise
        finally:
            plan_repository.close()
        _send_json(self, http.HTTPStatus.CREATED, {"plan": journey_plan_document(plan)})

    def _journey_travel_leg_delete(self, journey_id, travel_leg_id):
        repository = self.server.get_travel_leg_repository()
        try:
            try:
                repository.delete(
                    travel_leg_id=travel_leg_id,
                    journey_id=journey_id,
                    owner_id=self.owner_id,
                )
            except KeyError as error:
                _send_json(
                    self, http.HTTPStatus.NOT_FOUND, {"error": f"travel leg not found: {error}"}
                )
                return
        finally:
            repository.close()
        _send_json(self, http.HTTPStatus.OK, {"deleted": travel_leg_id})

    def _routing_segment_post(self):
        """Route drawing v0.1 (docs/route_drawing_v0_1_design.md, Krok 2):
        proxies a routing request to the public brouter.de server -- the
        browser never calls it directly, the same pattern
        /api/runtime/map already uses for MapTiler. No journey_id, no
        persistence; the browser composes returned segments into a
        route_geometry and posts the finished route via the existing
        POST /api/journeys/{id}/route (WP1) once the user saves."""
        try:
            length = int(self.headers.get("Content-Length", "0"))
            payload = json.loads(self.rfile.read(length))
            points = route_points_from_payload(payload["points"])
            # AV-031: the activity being drawn picks the BRouter profile from
            # the registry (absent: trekking / hiking-mountain, as before);
            # an unknown activity is a 400, never a fallback profile.
            profile = activity_for(payload.get("activity_id")).routing_profile
            geometry, point_surfaces = fetch_route_segment_detail(
                points, profile=profile, opener=self.server.routing_opener
            )
            # AV-031 (WP2): each point's OSM surface from the same answer.
            _send_json(
                self,
                http.HTTPStatus.OK,
                {"geometry_geojson": geometry, "point_surfaces": point_surfaces},
            )
        except (json.JSONDecodeError, KeyError, TypeError, ValueError) as error:
            _send_json(self, http.HTTPStatus.BAD_REQUEST, {"error": str(error)})
        except RuntimeError as error:
            _send_json(self, http.HTTPStatus.BAD_GATEWAY, {"error": str(error)})
        except Exception as error:  # pragma: no cover - defensive local-server boundary
            self.log_error("Routing segment request failed: %s", error)
            _send_json(self, http.HTTPStatus.INTERNAL_SERVER_ERROR, {"error": str(error)})

    def _day_derivation_preview_post(self):
        """Stateless day-derivation preview for an unsaved, drawn/imported
        Planning Workspace route -- no journey_id, no persistence. Uses the
        exact same derive_day_segments/night_conditions_for_point as the
        saved-Journey CampsService path (see mountain_twin/journey/
        camps_service.py) so a preview and a saved plan never diverge."""
        try:
            length = int(self.headers.get("Content-Length", "0"))
            payload = json.loads(self.rfile.read(length))
            route_points_in = payload["route_points"]
            if not isinstance(route_points_in, list) or not route_points_in:
                raise ValueError("route_points must be a non-empty list")
            # AV-064: transfers (dojazd) are no movement -- the same movement
            # view a saved route gets (route_segments.py).
            segment_of = movement_segment_indexes(
                validate_segments(payload.get("segments"), len(route_points_in)),
                len(route_points_in),
            )
            route_points = [
                RoutePoint(
                    route_id="preview",
                    point_index=index,
                    track_index=0,
                    segment_index=segment_of[index],
                    latitude=float(item["latitude"]),
                    longitude=float(item["longitude"]),
                    elevation_m=(
                        float(item["elevation_m"]) if item.get("elevation_m") is not None else None
                    ),
                )
                for index, item in enumerate(route_points_in)
            ]
            pace_profile_id = payload.get("pace_profile_id", DEFAULT_PACE_PROFILE_ID)
            # AV-031: the activity being drawn (absent: trekking); a bike has
            # no pace model -> PLAN_TIMING_UNAVAILABLE, never a walking time.
            pace_factor = pace_factor_for_activity(payload.get("activity_id"), pace_profile_id)
            # Without a start (Planning Workspace since T2, docs/
            # design_reference/weather_start_models_spike_v0_1.md) the days
            # are timed relative to their own departure: only how long each
            # day takes, no clock times and no camp nights.
            relative = payload.get("start_datetime") is None
            start_datetime = (
                RELATIVE_TIMING_REFERENCE_START
                if relative
                else datetime.fromisoformat(payload["start_datetime"])
            )
            prepared = prepare_route(route_points).points
            markers = []
            for entry in payload.get("camp_markers", []):
                index = int(entry["route_point_index"])
                if not (0 <= index < len(prepared)):
                    raise ValueError("camp route point index is outside the route")
                markers.append(
                    CampMarker(index, prepared[index].cumulative_distance_m, entry["label"])
                )
            sorted_markers = tuple(sorted(markers, key=lambda marker: marker.route_point_index))
            pauses = generate_relative_pauses(prepared[-1].cumulative_distance_m)
            segments = derive_day_segments(
                route_points,
                pace_factor=pace_factor,
                pauses=pauses,
                start_datetime=start_datetime,
                camp_markers=sorted_markers,
            )
            day_segments = [
                {
                    **segment.to_dict(),
                    "duration_s": (segment.arrival_time - segment.departure_time).total_seconds(),
                }
                for segment in segments
            ]
            if relative:
                day_segments = [relative_day_document(segment) for segment in segments]
                _send_json(
                    self,
                    http.HTTPStatus.OK,
                    {
                        "pace_profile_id": pace_profile_id,
                        "timing": "RELATIVE",
                        "day_segments": day_segments,
                        "camps": [
                            {
                                "route_point_index": marker.route_point_index,
                                "label": marker.label,
                                "arrival_time": None,
                                "night_conditions": {
                                    "state": "NOT_EVALUATED",
                                    "reason": NO_PLANNED_START,
                                },
                            }
                            for marker in sorted_markers
                        ],
                    },
                )
                return
            timezone_name = payload.get("timezone_name", "UTC")
            live_resolver = self.server.get_live_weather_resolver()
            camps_out = []
            for marker, segment in zip(sorted_markers, segments):
                point = route_points[marker.route_point_index]
                night_conditions = night_conditions_for_point(
                    live_resolver,
                    latitude=point.latitude,
                    longitude=point.longitude,
                    elevation_m=point.elevation_m,
                    arrival_time=segment.arrival_time,
                    timezone_name=timezone_name,
                )
                camps_out.append(
                    {
                        "route_point_index": marker.route_point_index,
                        "label": marker.label,
                        "arrival_time": segment.arrival_time.isoformat(),
                        "night_conditions": night_conditions,
                    }
                )
            _send_json(
                self,
                http.HTTPStatus.OK,
                {
                    "pace_profile_id": pace_profile_id,
                    "timing": "CLOCK",
                    "day_segments": day_segments,
                    "camps": camps_out,
                },
            )
        except DayDerivationUnavailable as unavailable:
            _send_json(
                self, http.HTTPStatus.UNPROCESSABLE_ENTITY, _plan_timing_unavailable(unavailable)
            )
        except (json.JSONDecodeError, KeyError, TypeError, ValueError) as error:
            _send_json(self, http.HTTPStatus.BAD_REQUEST, {"error": str(error)})
        except Exception as error:  # pragma: no cover - defensive local-server boundary
            self.log_error("Day-derivation preview failed: %s", error)
            _send_json(self, http.HTTPStatus.INTERNAL_SERVER_ERROR, {"error": str(error)})


def journey_owner_of(server, journey_id: str) -> str:
    """Who owns this saved Journey (AV-054): the request was already checked
    at the gate; background work (the prefetch) reads every owner's."""
    connection = JourneySqliteDatabase(server.journey_db).connect()
    try:
        row = connection.execute(
            "SELECT owner_id FROM journeys WHERE journey_id = ?", (journey_id,)
        ).fetchone()
    finally:
        connection.close()
    return row["owner_id"] if row is not None else server.journey_owner


def journey_owners(server) -> list[str]:
    connection = JourneySqliteDatabase(server.journey_db).connect()
    try:
        owners = [row[0] for row in connection.execute("SELECT DISTINCT owner_id FROM journeys")]
    finally:
        connection.close()
    return owners or [server.journey_owner]


def prefetch_candidates(server) -> list[PrefetchCandidate]:
    """AV-048: the saved Journeys as the weather prefetch sees them: route
    or not, completed or not, and -- from the plan, by the same pace model
    as everywhere -- when the trip starts and ends."""
    candidates = []
    journeys = [
        (owner, journey)
        for owner in journey_owners(server)
        for journey in server.get_journey_catalog().list(owner)
    ]
    for owner, journey in journeys:
        if journey.get("source") != "DURABLE_CONTAINER":
            continue
        journey_id = journey["journey_id"]
        has_route = bool(journey.get("capabilities", {}).get("has_route"))
        start = end = None
        if has_route and journey.get("capabilities", {}).get("has_plan"):
            try:
                routes = server.get_journey_route_repository()
                try:
                    route, geometry = routes.get_current_route(
                        journey_id=journey_id, owner_id=owner
                    )
                finally:
                    routes.close()
                plans = server.get_journey_plan_repository()
                try:
                    plan = plans.get_current_plan(journey_id=journey_id, owner_id=owner)
                finally:
                    plans.close()
                if plan is not None and plan.planned_start_local is not None:
                    start = datetime.fromisoformat(plan.planned_start_local)
                    points = route_points_from_geometry_geojson(route.route_revision_id, geometry)
                    prepared = prepare_route(points).points
                    days = derive_day_segments(
                        points,
                        pace_factor=pace_factor_for_activity(
                            route.activity_id, plan.pace_policy_id
                        ),
                        pauses=generate_relative_pauses(prepared[-1].cumulative_distance_m),
                        start_datetime=start,
                        camp_markers=plan.camp_markers,
                        timezone_name=plan.journey_timezone,
                        rest_days_before_start=plan.rest_days_before_start,
                    )
                    end = days[-1].arrival_time if days else start
            except (DayDerivationUnavailable, ValueError):
                pass  # no timed plan: only "recently opened" can make it active
        candidates.append(
            PrefetchCandidate(
                journey_id=journey_id,
                completed=journey.get("completed_at") is not None,
                has_route=has_route,
                trip_start=start,
                trip_end=end,
            )
        )
    return candidates


def start_usage(root: Path, port: int) -> UsageStore:
    """AV-052: the process's usage store (offline mode: its own scratch
    file) with the alarms listening."""
    directory = (
        offline_cache_directory() if offline_mode() is not None else root / "data/cache/usage"
    )
    store = UsageStore(
        directory / "service_usage.sqlite3",
        instance=os.environ.get("ECHTRO_INSTANCE") or f"serve_visual_prototype:{port}",
    )
    store.listeners.append(UsageAlarms(store))
    configure_usage(store)
    return store


def start_weather_prefetch(server) -> WeatherPrefetcher:
    """AV-048: the background prefetch of the active Journeys' default
    Weather document (the default model; the comparison stays on request)."""
    prefetcher = WeatherPrefetcher(
        cache=server.get_forecast_cache(),
        candidates=lambda: prefetch_candidates(server),
        warm=lambda journey_id: weather_series_for(server, journey_id, {}),
        interval_seconds=prefetch_interval_seconds(),
        log=lambda line: print(line, file=sys.stderr, flush=True),
    )
    prefetcher.start()
    return prefetcher


def weather_series_for(server, journey_id: str, query, walking_only: bool = False) -> dict:
    """docs/pogoda_v0_1_design.md: the Weather tab's series along the saved
    route for the tree's range (?first_day=&last_day=, 0-based; a range
    needs a plan that can be timed), both slider modes in one document.
    walking_only: section 7.3's walking-time preview for the same samples
    (?start_local=YYYY-MM-DDTHH:MM&pace_profile_id=...) -- read-only, the
    saved plan is never written. AV-048: a function of the server (not of a
    request), so the background prefetch builds exactly what a page gets."""
    owner = journey_owner_of(server, journey_id)
    route_repository = server.get_journey_route_repository()
    try:
        route_result = route_repository.get_current_route(journey_id=journey_id, owner_id=owner)
    finally:
        route_repository.close()
    if route_result is None:
        raise ValueError("a weather series requires a saved route")
    route, geometry = route_result
    route_points = route_points_from_geometry_geojson(route.route_revision_id, geometry)
    plan_repository = server.get_journey_plan_repository()
    try:
        plan = plan_repository.get_current_plan(journey_id=journey_id, owner_id=owner)
    finally:
        plan_repository.close()

    def optional_int(name):
        value = query.get(name, [None])[0]
        return None if value in (None, "") else int(value)

    timezone_name = resolve_route_timezone(route_points[0].latitude, route_points[0].longitude)
    # AV-048: "Pokaż według zapisanych dat" -- the saved dates whatever they
    # are; by default the next start (weather_anchor).
    dates = query.get("dates", [""])[0]
    if dates not in ("", "saved"):
        raise ValueError("dates must be saved or absent")
    if walking_only:
        return walking_times_document(
            journey_id=journey_id,
            route_points=route_points,
            plan=plan,
            timezone_name=timezone_name,
            first_day=optional_int("first_day"),
            last_day=optional_int("last_day"),
            start_local=_walk_start(query, timezone_name),
            pace_profile_id=query.get("pace_profile_id", [None])[0] or None,
            activity_id=route.activity_id,
            prefer_saved_dates=dates == "saved",
        )
    # Step M2: named models side by side only when asked for -- each
    # model weighs one provider call per sampled location.
    compare = query.get("compare_models", ["0"])[0]
    if compare not in ("0", "1"):
        raise ValueError("compare_models must be 0 or 1")
    compare_models = compare == "1"
    # AV-051: the ground temperature and snow depth charts, asked for.
    extras = query.get("extras", ["0"])[0]
    if extras not in ("0", "1"):
        raise ValueError("extras must be 0 or 1")
    # AV-052: a trip starting far ahead is not refreshed on opening more often
    # than its prefetch would (mountain_twin.journey.weather_prefetch.fresh_for).
    trip_start = None
    if plan is not None and plan.planned_start_local is not None:
        try:
            trip_start = datetime.fromisoformat(plan.planned_start_local).replace(
                tzinfo=ZoneInfo(plan.journey_timezone)
            )
        except (ValueError, KeyError):
            trip_start = None
    with fresh_for_at_least(
        prefetch_fresh_for(trip_start, datetime.now(timezone.utc)) if trip_start else 0.0
    ):
        return weather_series_document(
            journey_id=journey_id,
            route_points=route_points,
            plan=plan,
            timezone_name=timezone_name,
            live_resolver=server.get_live_weather_resolver(),
            first_day=optional_int("first_day"),
            last_day=optional_int("last_day"),
            compare_models=compare_models,
            activity_id=route.activity_id,
            # AV-036/AV-043: 24 h, 72 h or the whole trip; none given: 72 h
            # on the fixed-hour axis, the whole plan on the plan's dates.
            horizon=query.get("horizon", [""])[0] or None,
            prefer_saved_dates=dates == "saved",
            extras=extras == "1",
        )


def _walk_start(query, timezone_name: str) -> str | None:
    """A walking preview's start: ?start_local= (a chosen day), or -- AV-035 --
    ?start_time=HH:MM, its next occurrence in the route's timezone."""
    start_time = query.get("start_time", [None])[0] or None
    if start_time:
        if not re.fullmatch(r"[0-2]\d:[0-5]\d", start_time) or int(start_time[:2]) > 23:
            raise ValueError("start_time must be HH:MM")
        return next_start(start_time, timezone_name, datetime.now(timezone.utc)).strftime(
            "%Y-%m-%dT%H:%M"
        )
    return query.get("start_local", [None])[0] or None


def _plan_timing_unavailable(unavailable: DayDerivationUnavailable) -> dict:
    return {
        "error": "PLAN_TIMING_UNAVAILABLE",
        "state": "UNAVAILABLE",
        "reason_codes": list(unavailable.reason_codes),
        "unavailable_elevation_points": unavailable.unavailable_elevation_points,
        "elevation_gaps": unavailable.gaps_document(),
    }


LIVE_TOKEN_IN_PATH = re.compile(r"/live/(api/)?([vp])/[A-Za-z0-9_-]+")


# AV-026: this process serves the page's files fresh from disk on every
# request, but its own Python code is what it started with. A server left
# running across a code change answers the new page with old routes (live
# tracking: 400/404). The page asks /api/runtime/server and says so.
SERVER_CODE_ROOTS = (
    Path(__file__).resolve(),
    Path(__file__).resolve().parents[1] / "mountain_twin",
)
SERVER_STARTED_AT = time.time()


def _server_code_files():
    for root in SERVER_CODE_ROOTS:
        yield from ([root] if root.is_file() else root.rglob("*.py"))


def server_code_state(started_at: float | None = None) -> dict:
    started_at = SERVER_STARTED_AT if started_at is None else started_at
    changed = sorted(
        str(path.relative_to(Path(__file__).resolve().parents[1]))
        for path in _server_code_files()
        if path.stat().st_mtime > started_at
    )
    return {
        "started_at": datetime.fromtimestamp(started_at).astimezone().isoformat(timespec="seconds"),
        "stale": bool(changed),
        "changed_files": changed[:5],
        "changed_count": len(changed),
    }


def _local_secret(path: Path) -> bytes:
    """The local live-tracking grant key: random, 0600, made once."""
    if path.exists():
        return bytes.fromhex(path.read_text(encoding="ascii").strip())
    import secrets

    path.parent.mkdir(parents=True, exist_ok=True)
    key = secrets.token_bytes(32)
    path.write_text(key.hex(), encoding="ascii")
    os.chmod(path, 0o600)
    return key


def map_runtime_config(environ) -> dict | None:
    """The document of GET /api/runtime/map (AV-042). SELF_HOSTED_TILES_URL
    (the own tile server) gives the outdoor vector basemap, its glyphs and
    sprite, and the Mapterhorn terrain -- no key. MapTiler is an option, not
    a requirement: MAP_BASEMAP=maptiler with MAPTILER_API_KEY, or the key
    alone when there is no own tile server (USE_OWN_SERVER=0). The key is
    handed to the page only when MapTiler is the basemap. None: nothing is
    configured, the pages fall back to the analytical geometry."""
    tiles_url = (environ.get("SELF_HOSTED_TILES_URL") or "").strip().rstrip("/")
    key = (environ.get("MAPTILER_API_KEY") or "").strip()
    choice = (environ.get("MAP_BASEMAP") or "").strip().lower() or (
        "self_hosted" if tiles_url else "maptiler"
    )
    if choice == "maptiler" and key:
        document = {"basemap": "maptiler", "maptilerApiKey": key}
        if tiles_url:
            document["tilesUrl"] = tiles_url
        return document
    if tiles_url:
        return {"basemap": "self_hosted", "tilesUrl": tiles_url}
    return None


def _local_live_map_config() -> dict:
    """The local view page's basemap: the same choice as Przegląd
    (map_runtime_config), as the live page's own document; else none (the
    route on the graphite canvas)."""
    document = map_runtime_config(os.environ)
    if document is None:
        return {"basemap": None}
    if document["basemap"] == "maptiler":
        return {"basemap": {"kind": "maptiler", "maptilerApiKey": document["maptilerApiKey"]}}
    return {"basemap": {"kind": "self_hosted", "tilesUrl": document["tilesUrl"]}}


def _send_json(handler, status: http.HTTPStatus, document: dict) -> None:
    body = json.dumps(document, sort_keys=True, allow_nan=False).encode("utf-8")
    handler.send_response(status)
    handler.send_header("Content-Type", "application/json; charset=utf-8")
    # Product state is local, mutable, and selection-sensitive. A stale browser
    # response must not conceal durable Journeys or replace a newer bootstrap.
    handler.send_header("Cache-Control", "no-store")
    handler.send_header("Content-Length", str(len(body)))
    handler.end_headers()
    handler.wfile.write(body)


LOCAL_DEV_ENV = Path("local_dev.env")
OWN_SERVER_KEYS = (
    "BROUTER_ENDPOINT",
    "SELF_HOSTED_TILES_URL",
    "LIVE_TRACKING_URL",
    "LIVE_TRACKING_ADMIN_SECRET",
)


def load_local_dev_env(path: Path, environ=None) -> str:
    """The owner's local development defaults (AV-021): a git-ignored
    KEY=VALUE file (see local_dev.env.example) whose values apply only where
    the shell has not set the same variable. USE_OWN_SERVER=0 -- in the file
    or, for one run, in the shell -- leaves out the own-server variables
    (routing, maps), i.e. back to public brouter.de and (with a key)
    MapTiler with one setting. Returns one line saying which sources are in use."""
    environ = os.environ if environ is None else environ
    values: dict[str, str] = {}
    if path.exists():
        for raw in path.read_text(encoding="utf-8").splitlines():
            line = raw.strip()
            if not line or line.startswith("#") or "=" not in line:
                continue
            key, value = line.split("=", 1)
            values[key.strip()] = value.strip()
    use_own = environ.get("USE_OWN_SERVER", values.get("USE_OWN_SERVER", "0")).strip().lower()
    own_server = use_own in ("1", "true", "yes", "on")
    for key, value in values.items():
        if key in OWN_SERVER_KEYS and not own_server:
            continue
        environ.setdefault(key, value)
    from mountain_twin.journey.route_provider import brouter_endpoint

    maps = map_runtime_config(environ)
    maps_line = (
        "none (analytical geometry)"
        if maps is None
        else f"own server ({maps['tilesUrl']})"
        if maps["basemap"] == "self_hosted"
        else "MapTiler"
        + (f", 3D terrain own server ({maps['tilesUrl']})" if "tilesUrl" in maps else "")
    )
    source = path if path.exists() else "no local_dev.env"
    live = (
        f"public server ({environ.get('LIVE_TRACKING_URL')})"
        if environ.get("LIVE_TRACKING_URL") and environ.get("LIVE_TRACKING_ADMIN_SECRET")
        else "local (this machine only)"
    )
    return (
        f"Sources ({source}): routing {brouter_endpoint(environ)}; maps {maps_line}; "
        f"live tracking {live}"
    )


def main(argv: list[str] | None = None) -> int:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--root", type=Path, default=Path("."))
    parser.add_argument("--port", type=int, default=8000)
    parser.add_argument(
        "--journey-db",
        type=Path,
        default=None,
        help=f"Journey-container SQLite database (default: <root>/{DEFAULT_JOURNEY_DB})",
    )
    parser.add_argument("--journey-owner", default="aventurro-local-user")
    parser.add_argument(
        "--no-export",
        action="store_true",
        help="serve an existing visualization export without refreshing it",
    )
    args = parser.parse_args(argv)
    root = args.root.resolve()
    print(load_local_dev_env(root / LOCAL_DEV_ENV), flush=True)
    if not args.no_export:
        source = root / DEFAULT_SOURCE
        if not source.exists():
            raise FileNotFoundError(f"authoritative adaptive analysis is absent: {source}")
        write_solar_visualization_export(source, root / DEFAULT_EXPORT)
    journey_db, migration_note = prepare_journey_db(args.journey_db, root, LEGACY_JOURNEY_DB)
    if migration_note:
        print(migration_note, flush=True)
    print(f"Saved Journeys: {journey_db}", flush=True)
    handler = partial(ExplorerRequestHandler, directory=str(root))
    server = ExplorerHTTPServer(
        ("127.0.0.1", args.port),
        handler,
        root,
        journey_db,
        args.journey_owner,
    )
    # AV-052: every outside request counted (per hour, service, feature) and
    # guarded by the budgets; alarms to the owner. This server's own count.
    usage_store = start_usage(root, args.port)
    print(
        f"Service usage: {usage_store.path} (instance {usage_store.instance}); "
        f"alarms: {usage_store.listeners[0].channel}",
        flush=True,
    )
    if getattr(server, "accounts", False):
        threading.Thread(target=server.warm_reference_journey, daemon=True).start()
    if prefetch_enabled():
        prefetcher = start_weather_prefetch(server)
        print(
            f"Weather prefetch: active Journeys every "
            f"{prefetcher.interval_seconds / 3600:g} h, first in "
            f"{prefetcher.initial_delay_seconds:.0f} s (WEATHER_PREFETCH=0 turns it off)",
            flush=True,
        )
    print(f"Open http://127.0.0.1:{args.port}/aventurro/?journey_id=journey-tmb-day-01-v0_1")
    print(f"Explorer remains available at http://127.0.0.1:{args.port}/visual_prototype/")
    try:
        server.serve_forever()
    except KeyboardInterrupt:
        return 0
    finally:
        server.server_close()
    return 0


if __name__ == "__main__":
    raise SystemExit(main())