extract ServiceClientBase from service clients

The base (service_api/_base.py) owns the cross-client invariants both
clients hand-rolled: the stale-credential fast-fail (_check_fresh), error
classification, the attempt → refresh → retry-once request skeleton
(_request), single-flight refresh orchestration with a _try_cheap_refresh
hook (Booking's HTTP session bump), env-file loading helpers, and the
default probe. Platform specifics (headers, payloads, parsing, endpoints)
stay flat in each client.

Deliberate behavior fixes now enforced structurally:
- Timeouts are never wrapped as auth errors. Airbnb _mutate and all four
  Booking request paths caught TransportError bare, so TimeoutException
  (its subclass) became an auth error and poisoned the service in
  _gather() with a misleading message.
- clear_stale runs on every successful refresh path for all clients
  (previously only Booking did this, guarded by a comment).
- Transport errors on the post-refresh retry are now classified the same
  as first-attempt errors instead of propagating raw.
- A client with no refresh_credentials (token-based platforms) gets a
  clear 'no automatic refresh available' auth error instead of an
  AttributeError.

room_cal/service_api/_airbnb.py

@@ -1,7 +1,6 @@
 import asyncio
 import base64
 import json
-import os
 import random
 import string
 import uuid
@@ -14,10 +13,11 @@ from dataclasses import dataclass
 import logging
 
 from ._scripts import _assert_not_async, _npm_env, _ts_dir, _ensure_installed
-from room_cal._utils import data_home, get_config, read_env
+from room_cal._utils import get_config
+from ._base import ServiceClientBase, _AUTH_HINT
 from ._month_cache import MonthCache, TTLCache
 from ._utils import month_bounds, months_in_range
-from .credentials import is_stale, mark_stale
+from .credentials import mark_stale
 from .types import CalendarEvent, DayInfo, ServiceAuthError, Reservation
 
 _log = logging.getLogger(__name__)
@@ -58,8 +58,6 @@ _HASHES = {
   'CreateBulkMessagesMutation':       '94ac2c4bd07edace539dbf2b9665d9030b6dee479db345ba8a8bbc234b3bbfa3',
 }
 
-_AUTH_HINT = 'refresh-creds'
-
 
 def _parse_price(s: str) -> float | None:
   """Parse a formatted price string like '$143.56' → 143.56, or None on failure."""
@@ -121,10 +119,13 @@ def _trace_id() -> str:
   return ''.join(random.choices(string.ascii_lowercase + string.digits, k=28))
 
 
-class _AirbnbClient:
+class _AirbnbClient(ServiceClientBase):
+  service_key = 'airbnb'
   service_name = 'AirBnB'
+  auth_error_cls = AirbnbAuthError
 
   def __init__(self, cookies: str, listing_id: str, tz: str = 'America/Los_Angeles'):
+    super().__init__()
     self._cookies = cookies
     self._listing_id = listing_id
     self._global_id = _encode_listing_id(listing_id)
@@ -144,22 +145,17 @@ class _AirbnbClient:
       log=_log,
     )
     self._details: TTLCache[dict] = TTLCache()
-    self._refresh_lock = asyncio.Lock()
-    self._refresh_task: asyncio.Task | None = None
     self._http = httpx.AsyncClient(timeout=10.0)
     self._booked_nets: dict[str, int] = {}
 
   @classmethod
   def from_env(cls) -> '_AirbnbClient':
-    env_file = data_home() / 'airbnb.env'
-    env = {**read_env(env_file), **os.environ}
-    cookies = env.get('AIRBNB_COOKIES', '')
-    if not cookies:
-      raise ValueError(f'AIRBNB_COOKIES must be set in {env_file}')
+    env = cls._read_env()
+    cls._require_env(env, 'AIRBNB_COOKIES')
     listing_id = get_config().services['airbnb'].room_id
     if not listing_id:
       raise ValueError('airbnb.room_id must be set in config.json')
-    return cls(cookies=cookies, listing_id=listing_id)
+    return cls(cookies=env['AIRBNB_COOKIES'], listing_id=listing_id)
 
   def clear_cache(self) -> None:
     self._cache.clear()
@@ -168,11 +164,8 @@ class _AirbnbClient:
     self._details.clear()
     self._booked_nets.clear()
 
-  def _reload_from_env(self) -> None:
-    env_file = data_home() / 'airbnb.env'
-    env = {**read_env(env_file), **os.environ}
+  def _apply_env(self, env: dict[str, str]) -> None:
     self._cookies = env.get('AIRBNB_COOKIES', self._cookies)
-    self.clear_cache()
 
   @staticmethod
   def refresh_credentials() -> None:
@@ -190,22 +183,6 @@ class _AirbnbClient:
       raise ServiceAuthError(f'Airbnb session expired — re-run without HEADLESS to re-authenticate (exit {e.returncode})') from None
     _log.info('airbnb: credential refresh complete')
 
-  async def _auto_refresh(self) -> None:
-    _log.warning('Airbnb session may be expired — launching headless Playwright')
-    try:
-      await asyncio.get_running_loop().run_in_executor(None, _AirbnbClient.refresh_credentials)
-      self._reload_from_env()
-    except ServiceAuthError:
-      mark_stale('airbnb')
-      raise
-
-  async def _ensure_refreshed(self) -> None:
-    async with self._refresh_lock:
-      if self._refresh_task is None or self._refresh_task.done():
-        self._refresh_task = asyncio.create_task(self._auto_refresh())
-      task = self._refresh_task
-    await task
-
   def _headers(self, referer: str, extra: dict | None = None) -> dict[str, str]:
     h = {
       'Cookie': self._cookies,
@@ -238,7 +215,7 @@ class _AirbnbClient:
       e.get('extensions', {}).get('response', {}).get('status') in (400, 401, 403)
       for e in body.get('errors', [])
     ):
-      mark_stale('airbnb')
+      mark_stale(self.service_key)
       raise AirbnbAuthError(f'Airbnb session expired — run: {_AUTH_HINT}')
 
   async def _gql(self, op: str, h: str, variables: dict, referer: str) -> dict:
@@ -252,19 +229,7 @@ class _AirbnbClient:
     }
     async def _attempt() -> httpx.Response:
       return await self._http.get(url, params=params, headers=self._headers(referer))
-    try:
-      r = await _attempt()
-    except httpx.TimeoutException as e:
-      raise httpx.ReadTimeout(f'Airbnb request timed out ({op})') from e
-    except (httpx.TransportError, OSError, ExceptionGroup) as e:
-      raise AirbnbAuthError(f'Airbnb connection failed — session may be expired, run: {_AUTH_HINT}') from e
-    if r.status_code in (400, 401, 403):
-      await self._ensure_refreshed()
-      r = await _attempt()
-    if r.status_code in (400, 401, 403):
-      mark_stale('airbnb')
-      raise AirbnbAuthError(f'Airbnb session expired (HTTP {r.status_code}) — run: {_AUTH_HINT}')
-    r.raise_for_status()
+    r = await self._request(_attempt, op=op)
     body = r.json()
     self._check_auth(body)
     return body
@@ -278,17 +243,7 @@ class _AirbnbClient:
         url, json=payload,
         headers=self._headers(referer, {'Origin': 'https://www.airbnb.com'}),
       )
-    try:
-      r = await _attempt()
-    except (httpx.TransportError, OSError, ExceptionGroup) as e:
-      raise AirbnbAuthError(f'Airbnb connection failed — session may be expired, run: {_AUTH_HINT}') from e
-    if r.status_code in (400, 401, 403):
-      await self._ensure_refreshed()
-      r = await _attempt()
-    if r.status_code in (400, 401, 403):
-      mark_stale('airbnb')
-      raise AirbnbAuthError(f'Airbnb session expired (HTTP {r.status_code}) — run: {_AUTH_HINT}')
-    r.raise_for_status()
+    r = await self._request(_attempt, op=op)
     body = r.json()
     self._check_auth(body)
     return body
@@ -345,8 +300,7 @@ class _AirbnbClient:
       del self._cache[d]
 
   async def get_calendar(self, date_from: str, date_to: str) -> dict[str, DayInfo]:
-    if is_stale('airbnb'):
-      raise AirbnbAuthError(f'Airbnb credentials are stale — run: {_AUTH_HINT}')
+    self._check_fresh()
     month_list = months_in_range(date_from, date_to)
     all_days = await asyncio.gather(*[self._months.get(m, y) for m, y in month_list])
     for (month, year), days in zip(month_list, all_days):
@@ -376,8 +330,7 @@ class _AirbnbClient:
     }
 
   async def get_reservations(self, date_from: str, date_to: str) -> list[Reservation]:
-    if is_stale('airbnb'):
-      raise AirbnbAuthError(f'Airbnb credentials are stale — run: {_AUTH_HINT}')
+    self._check_fresh()
     month_list = months_in_range(date_from, date_to)
     all_days = await asyncio.gather(*[self._months.get(m, y) for m, y in month_list])
     booked: dict[str, str] = {}
@@ -465,8 +418,7 @@ class _AirbnbClient:
     Results are cached in _booked_nets; repeated calls (e.g. after get_reservations)
     hit the detail cache and make no additional network calls.
     """
-    if is_stale('airbnb'):
-      raise AirbnbAuthError(f'Airbnb credentials are stale — run: {_AUTH_HINT}')
+    self._check_fresh()
     month_list = months_in_range(date_from, date_to)
     all_days = await asyncio.gather(*[self._months.get(m, y) for m, y in month_list])
 
@@ -503,8 +455,7 @@ class _AirbnbClient:
 
   async def set_price(self, d: str, price: int) -> None:
     price = int(round(price))
-    if is_stale('airbnb'):
-      raise AirbnbAuthError(f'Airbnb credentials are stale — run: {_AUTH_HINT}')
+    self._check_fresh()
     if (cached := self._cache.get(d)) and cached.price == price:
       return
     await self._mutate(
@@ -522,8 +473,7 @@ class _AirbnbClient:
     _log.info('airbnb set_price date=%s price=$%d', d, price)
 
   async def set_availability(self, d: str, available: bool) -> None:
-    if is_stale('airbnb'):
-      raise AirbnbAuthError(f'Airbnb credentials are stale — run: {_AUTH_HINT}')
+    self._check_fresh()
     if (cached := self._cache.get(d)) and cached.available == available:
       return
     body = await self._mutate(
@@ -571,8 +521,7 @@ class _AirbnbClient:
 
   async def get_thread(self, thread_id: str) -> AirbnbThread:
     """Fetch a conversation thread and its messages by numeric thread ID."""
-    if is_stale('airbnb'):
-      raise AirbnbAuthError(f'Airbnb credentials are stale — run: {_AUTH_HINT}')
+    self._check_fresh()
     resp = await self._gql(
       'ViaductGetThreadAndDataQuery', _HASHES['ViaductGetThreadAndDataQuery'],
       {
@@ -619,8 +568,7 @@ class _AirbnbClient:
 
   async def send_message(self, thread_id: str, body: str) -> None:
     """Send a text message to an existing thread."""
-    if is_stale('airbnb'):
-      raise AirbnbAuthError(f'Airbnb credentials are stale — run: {_AUTH_HINT}')
+    self._check_fresh()
     await self._mutate(
       'CreateBulkMessagesMutation', _HASHES['CreateBulkMessagesMutation'],
       {
@@ -642,8 +590,7 @@ class _AirbnbClient:
 
   async def get_suggested_prices(self, date_from: str, date_to: str) -> dict[str, int]:
     """Airbnb's per-day smart-pricing suggestions (USD). Only includes days where shouldShowTip=True."""
-    if is_stale('airbnb'):
-      raise AirbnbAuthError(f'Airbnb credentials are stale — run: {_AUTH_HINT}')
+    self._check_fresh()
     resp = await self._gql(
       'getNightlyPriceTips', _HASHES['getNightlyPriceTips'],
       {
@@ -670,15 +617,9 @@ class _AirbnbClient:
     )
     return AirbnbData(suggested_prices=suggested_prices, calendar_events=calendar_events)
 
-  async def probe(self) -> None:
-    from datetime import date
-    today = date.today().isoformat()
-    await self.get_calendar(today, today)
-
   async def get_calendar_events(self) -> list[CalendarEvent]:
     """Local events and holidays Airbnb shows on the host calendar (holidays, major events, etc.)."""
-    if is_stale('airbnb'):
-      raise AirbnbAuthError(f'Airbnb credentials are stale — run: {_AUTH_HINT}')
+    self._check_fresh()
     resp = await self._gql(
       'HostCalendarListingEventsQuery', _HASHES['HostCalendarListingEventsQuery'],
       {'listingId': self._global_id},

room_cal/service_api/_base.py

@@ -0,0 +1,146 @@
+"""Shared base for service clients: staleness, request retry, credential refresh.
+
+ServiceClientBase owns the cross-client invariants — the stale-credential
+fast-fail, error classification (a timeout is never an auth error), the
+attempt → refresh → retry-once request skeleton, and single-flight refresh
+orchestration. Platform specifics (headers, payloads, parsing, endpoints)
+stay flat in each client module.
+
+Request state must be built inside the attempt closure, not captured outside
+it, so a retry after _ensure_refreshed() uses the reloaded session (e.g.
+Booking.com's ses param).
+"""
+import asyncio
+import logging
+import os
+from collections.abc import Awaitable, Callable
+from datetime import date
+
+import httpx
+
+from room_cal._utils import data_home, read_env
+from .credentials import clear_stale, is_stale, mark_stale
+from .types import ServiceAuthError
+
+_log = logging.getLogger(__name__)
+
+_AUTH_HINT = 'refresh-creds'
+_AUTH_STATUSES = (400, 401, 403)
+
+
+class ServiceClientBase:
+  service_key: str      # registry key; also the env-file stem (<key>.env) and stale-flag name
+  service_name: str     # human-readable display name
+  auth_error_cls: type[ServiceAuthError] = ServiceAuthError
+
+  def __init__(self) -> None:
+    self._refresh_lock = asyncio.Lock()
+    self._refresh_task: asyncio.Task | None = None
+
+  # ── credentials and staleness ──────────────────────────────────────────────
+
+  def _check_fresh(self) -> None:
+    """Fast-fail when credentials are known-stale; call at the top of every public method."""
+    if is_stale(self.service_key):
+      raise self.auth_error_cls(f'{self.service_name} credentials are stale — run: {_AUTH_HINT}')
+
+  @classmethod
+  def _env_file(cls):
+    return data_home() / f'{cls.service_key}.env'
+
+  @classmethod
+  def _read_env(cls) -> dict[str, str]:
+    return {**read_env(cls._env_file()), **os.environ}
+
+  @classmethod
+  def _require_env(cls, env: dict[str, str], *keys: str) -> None:
+    missing = [k for k in keys if not env.get(k)]
+    if missing:
+      raise ValueError(f'Missing in {cls._env_file()}: {", ".join(missing)}')
+
+  def _apply_env(self, env: dict[str, str]) -> None:
+    """Pick up refreshed credential values; implemented by each client."""
+    raise NotImplementedError
+
+  def _reload_from_env(self) -> None:
+    self._apply_env(self._read_env())
+    self.clear_cache()
+
+  def clear_cache(self) -> None:
+    """Clear data caches; overridden by clients that cache."""
+
+  # ── request skeleton ───────────────────────────────────────────────────────
+
+  async def _request(self, attempt: Callable[[], Awaitable[httpx.Response]], *, op: str = '') -> httpx.Response:
+    """Run attempt; on an auth-shaped status, refresh credentials and retry once."""
+    r = await self._classified(attempt, op)
+    if r.status_code in _AUTH_STATUSES:
+      _log.warning('%s %s http=%d — refreshing credentials and retrying; body=%.500s',
+                   self.service_key, op, r.status_code, r.text)
+      await self._ensure_refreshed()
+      r = await self._classified(attempt, op)
+      if r.status_code in _AUTH_STATUSES:
+        mark_stale(self.service_key)
+        raise self.auth_error_cls(
+          f'{self.service_name} session expired (HTTP {r.status_code}) — run: {_AUTH_HINT}')
+      _log.info('%s %s recovered after refresh (http=%d)', self.service_key, op, r.status_code)
+    r.raise_for_status()
+    return r
+
+  async def _classified(self, attempt: Callable[[], Awaitable[httpx.Response]], op: str) -> httpx.Response:
+    try:
+      return await attempt()
+    except httpx.TimeoutException as e:
+      # A slow response is not a credential failure — never wrap it as an auth error,
+      # or the whole service gets skipped in _gather() with a misleading message.
+      raise httpx.ReadTimeout(f'{self.service_name} request timed out ({op})') from e
+    except (httpx.TransportError, OSError, ExceptionGroup) as e:
+      raise self.auth_error_cls(
+        f'{self.service_name} connection failed — session may be expired, run: {_AUTH_HINT}') from e
+
+  # ── refresh orchestration ──────────────────────────────────────────────────
+
+  async def _ensure_refreshed(self) -> None:
+    """Single-flight refresh: concurrent callers await the same in-progress attempt."""
+    async with self._refresh_lock:
+      if self._refresh_task is None or self._refresh_task.done():
+        self._refresh_task = asyncio.create_task(self._auto_refresh())
+      task = self._refresh_task
+    await task
+
+  async def _try_cheap_refresh(self) -> bool:
+    """Attempt a lightweight in-process session refresh; True on success.
+
+    Override where the platform allows it (e.g. Booking.com's HTTP session
+    bump); the default goes straight to the full credential refresh.
+    """
+    return False
+
+  async def _auto_refresh(self) -> None:
+    _log.warning('%s session may be expired — attempting credential refresh', self.service_name)
+    if await self._try_cheap_refresh():
+      _log.info('%s session refreshed in place', self.service_name)
+      clear_stale(self.service_key)
+      return
+    refresh = getattr(type(self), 'refresh_credentials', None)
+    if refresh is None:
+      mark_stale(self.service_key)
+      raise self.auth_error_cls(
+        f'{self.service_name} session expired and no automatic refresh is available — run: {_AUTH_HINT}')
+    _log.warning('%s: falling back to full credential refresh', self.service_name)
+    try:
+      await asyncio.get_running_loop().run_in_executor(None, refresh)
+      self._reload_from_env()
+      # a file-mode refresh writes the env file without set_credentials, so the
+      # stale flag must be cleared here or the service stays poisoned
+      clear_stale(self.service_key)
+    except ServiceAuthError:
+      mark_stale(self.service_key)
+      raise
+
+  # ── defaults ───────────────────────────────────────────────────────────────
+
+  async def probe(self) -> None:
+    """Liveness check for the health monitor; override if a calendar fetch is not a reliable signal."""
+    today = date.today().isoformat()
+    await self.get_calendar(today, today)  # type: ignore[attr-defined]

room_cal/service_api/_booking.py

@@ -1,7 +1,6 @@
 import asyncio
 import json
 import logging
-import os
 import random
 import time
 from dataclasses import dataclass
@@ -11,10 +10,11 @@ from urllib.parse import parse_qs, urlparse
 import httpx
 
 from ._scripts import _assert_not_async, _booking_lock, _npm_env, _ts_dir, _ensure_installed
-from room_cal._utils import data_home, get_config, read_env
+from room_cal._utils import get_config
+from ._base import ServiceClientBase, _AUTH_HINT
 from ._month_cache import MonthCache
 from ._utils import month_bounds, months_in_range
-from .credentials import clear_stale, is_stale, mark_stale, set_credentials
+from .credentials import set_credentials
 from .types import DayInfo, ServiceAuthError, Reservation
 
 _log = logging.getLogger(__name__)
@@ -25,7 +25,6 @@ def _is_past(d: str) -> bool:
 
 
 _BASE = 'https://admin.booking.com'
-_AUTH_HINT = 'refresh-creds'
 _SENDER_LABELS: dict[str, str] = {'GUEST': 'Guest', 'PARTNER': 'Host'}
 
 _UPDATE_MUTATION = """mutation updateRoomDateMutation($input: PartnerAvailabilityUpdateInput) {
@@ -145,8 +144,10 @@ def _page_view_id() -> str:
   return ''.join(random.choices('0123456789abcdef', k=16))
 
 
-class _BookingClient:
+class _BookingClient(ServiceClientBase):
+  service_key = 'booking'
   service_name = 'Booking.com'
+  auth_error_cls = BookingAuthError
 
   def __init__(
     self,
@@ -159,6 +160,7 @@ class _BookingClient:
     csrf_token: str | None = None,
     lang: str = 'xu',
   ):
+    super().__init__()
     self._cookies = cookies
     self._session = session
     self._hotel_id = hotel_id
@@ -175,18 +177,12 @@ class _BookingClient:
       label='Booking.com',
       log=_log,
     )
-    self._refresh_lock = asyncio.Lock()
-    self._refresh_task: asyncio.Task | None = None
     self._http = httpx.AsyncClient(timeout=10.0)
 
   @classmethod
   def from_env(cls) -> '_BookingClient':
-    env_file = data_home() / 'booking.env'
-    env = {**read_env(env_file), **os.environ}
-    required_env = ['BOOKING_COOKIES', 'BOOKING_SESSION']
-    missing_env = [k for k in required_env if not env.get(k)]
-    if missing_env:
-      raise ValueError(f'Missing in {env_file}: {", ".join(missing_env)}')
+    env = cls._read_env()
+    cls._require_env(env, 'BOOKING_COOKIES', 'BOOKING_SESSION')
     svc = get_config().services['booking']
     missing_ids = [k for k in ('hotel_id', 'account_id', 'room_id', 'rate_id') if not getattr(svc, k)]
     if missing_ids:
@@ -205,13 +201,10 @@ class _BookingClient:
     self._cache.clear()
     self._months.clear()
 
-  def _reload_from_env(self) -> None:
-    env_file = data_home() / 'booking.env'
-    env = {**read_env(env_file), **os.environ}
+  def _apply_env(self, env: dict[str, str]) -> None:
     self._cookies = env.get('BOOKING_COOKIES', self._cookies)
     self._session = env.get('BOOKING_SESSION', self._session)
     self._csrf_token = env.get('BOOKING_CSRF_TOKEN', self._csrf_token)
-    self.clear_cache()
 
   def _update_csrf(self, headers: httpx.Headers) -> None:
     """Capture CSRF from a response header if present and not yet known."""
@@ -220,7 +213,7 @@ class _BookingClient:
       self._csrf_token = csrf
       set_credentials('booking', {'BOOKING_CSRF_TOKEN': csrf})
 
-  async def _try_http_session_refresh(self) -> bool:
+  async def _try_cheap_refresh(self) -> bool:
     """GET the monthly calendar page; extract a fresh ses= from the final URL.
 
     Also captures CSRF from response headers if present.
@@ -272,30 +265,6 @@ class _BookingClient:
         raise ServiceAuthError(f'Booking.com credential refresh failed (exit {e.returncode})') from None
     _log.info('booking: credential refresh complete')
 
-  async def _auto_refresh(self, *, trigger: str = 'unknown') -> None:
-    _log.warning('Booking.com session may be expired — trying HTTP session refresh')
-    if await self._try_http_session_refresh():
-      _log.info('Booking.com session refreshed via HTTP')
-      clear_stale('booking')
-      return
-    _log.warning('HTTP refresh failed — launching headless Playwright')
-    try:
-      await asyncio.get_running_loop().run_in_executor(None, _BookingClient.refresh_credentials)
-      self._reload_from_env()
-      # the file-mode refresh script writes booking.env directly (no set_credentials),
-      # so the stale flag must be cleared here or the service stays poisoned
-      clear_stale('booking')
-    except ServiceAuthError:
-      mark_stale('booking')
-      raise
-
-  async def _ensure_refreshed(self, *, trigger: str) -> None:
-    async with self._refresh_lock:
-      if self._refresh_task is None or self._refresh_task.done():
-        self._refresh_task = asyncio.create_task(self._auto_refresh(trigger=trigger))
-      task = self._refresh_task
-    await task
-
   def _actor(self) -> dict:
     return {'actorId': str(self._hotel_id), 'actorType': 'HOTEL'}
 
@@ -401,20 +370,7 @@ class _BookingClient:
         json={'operationName': operation_name, 'variables': variables, 'query': query},
         headers=self._msg_headers(),
         )
-    try:
-      r = await _do_request()
-    except (httpx.TransportError, OSError, ExceptionGroup) as e:
-      _log.warning('booking _msg_gql %s attempt=%d transport error: %s', operation_name, attempt, e)
-      raise BookingAuthError(f'Booking.com connection failed — session may be expired, run: {_AUTH_HINT}') from e
-    _log.debug('booking _msg_gql %s attempt=%d http=%d', operation_name, attempt, r.status_code)
-    if r.status_code in (400, 401, 403):
-      _log.warning('booking _msg_gql %s attempt=%d http=%d body=%.500s', operation_name, attempt, r.status_code, r.text)
-      await self._ensure_refreshed(trigger='messaging')
-      r = await _do_request()
-      level = logging.WARNING if r.status_code in (400, 401, 403) else logging.INFO
-      _log.log(level, 'booking _msg_gql %s attempt=%d (after refresh) http=%d', operation_name, attempt, r.status_code)
-    await self._check_auth(r)
-    r.raise_for_status()
+    r = await self._request(_do_request, op=operation_name)
     raw = r.json()
     if raw.get('errors'):
       _log.warning('booking _msg_gql %s attempt=%d gql errors: %s', operation_name, attempt, json.dumps(raw['errors']))
@@ -422,25 +378,12 @@ class _BookingClient:
       raise RuntimeError(f'GraphQL error: {msgs}')
     return raw
 
-  async def _check_auth(self, r: httpx.Response) -> None:
-    if r.status_code in (400, 401, 403):
-      mark_stale('booking')
-      raise BookingAuthError(f'Booking.com session expired (HTTP {r.status_code}) — run: {_AUTH_HINT}')
-
   async def _multipart_post(self, url: str, referer: str, body: dict, extra_headers: dict | None = None) -> dict:
     async def _attempt() -> httpx.Response:
       headers = self._rest_headers(referer, extra_headers)
       return await self._http.post(url, params=self._common_params(),
                                    files={'request': (None, json.dumps(body))}, headers=headers)
-    try:
-      r = await _attempt()
-    except (httpx.TransportError, OSError, ExceptionGroup) as e:
-      raise BookingAuthError(f'Booking.com connection failed — session may be expired, run: {_AUTH_HINT}') from e
-    if r.status_code in (400, 401, 403):
-      await self._ensure_refreshed(trigger='inventory/fetch')
-      r = await _attempt()
-    await self._check_auth(r)
-    r.raise_for_status()
+    r = await self._request(_attempt, op=url.removeprefix(_BASE))
     self._update_csrf(r.headers)
     return r.json()
 
@@ -456,15 +399,7 @@ class _BookingClient:
               'extensions': {}, 'query': _UPDATE_MUTATION},
         headers=self._gql_headers(referer),
       )
-    try:
-      r = await _attempt()
-    except (httpx.TransportError, OSError, ExceptionGroup) as e:
-      raise BookingAuthError(f'Booking.com connection failed — session may be expired, run: {_AUTH_HINT}') from e
-    if r.status_code in (400, 401, 403):
-      await self._ensure_refreshed(trigger='graphql')
-      r = await _attempt()
-    await self._check_auth(r)
-    r.raise_for_status()
+    r = await self._request(_attempt, op='updateRoomDateMutation')
     raw = r.json()
     if raw.get('errors'):
       msgs = ', '.join(e.get('message', '') for e in raw['errors'])
@@ -544,22 +479,15 @@ class _BookingClient:
       del self._cache[d]
 
   async def get_calendar(self, date_from: str, date_to: str) -> dict[str, DayInfo]:
-    if is_stale('booking'):
-      raise BookingAuthError(f'Booking.com credentials are stale — run: {_AUTH_HINT}')
+    self._check_fresh()
     month_list = months_in_range(date_from, date_to)
     all_results = await asyncio.gather(*[self._months.get(m, y) for m, y in month_list])
     for month_result in all_results:
       self._cache.update(month_result)
     return {d: info for d, info in self._cache.items() if date_from <= d <= date_to}
 
-  async def probe(self) -> None:
-    from datetime import date
-    today = date.today().isoformat()
-    await self.get_calendar(today, today)
-
   async def get_reservations(self, date_from: str, date_to: str) -> list[Reservation]:
-    if is_stale('booking'):
-      raise BookingAuthError(f'Booking.com credentials are stale — run: {_AUTH_HINT}')
+    self._check_fresh()
     async def _attempt() -> httpx.Response:
       referer = self._monthly_referer()
       params = self._common_params({
@@ -573,15 +501,7 @@ class _BookingClient:
         params=params,
         headers=self._rest_headers(referer),
       )
-    try:
-      r = await _attempt()
-    except (httpx.TransportError, OSError, ExceptionGroup) as e:
-      raise BookingAuthError(f'Booking.com connection failed — session may be expired, run: {_AUTH_HINT}') from e
-    if r.status_code in (400, 401, 403):
-      await self._ensure_refreshed(trigger='get_reservations')
-      r = await _attempt()
-    await self._check_auth(r)
-    r.raise_for_status()
+    r = await self._request(_attempt, op='get_reservations')
     raw = r.json()
 
     reservations = raw.get('data', {}).get('reservations') or {}
@@ -602,8 +522,7 @@ class _BookingClient:
 
   async def get_thread(self, reservation_id: str) -> BookingThread:
     """Fetch a conversation thread and its messages by reservation ID."""
-    if is_stale('booking'):
-      raise BookingAuthError(f'Booking.com credentials are stale — run: {_AUTH_HINT}')
+    self._check_fresh()
     raw = await self._msg_gql(
       'conversationMessages', _CONVERSATION_MESSAGES_QUERY,
       {'input': {
@@ -641,8 +560,7 @@ class _BookingClient:
 
   async def send_message(self, reservation_id: str, body: str) -> None:
     """Send a text message to the guest conversation for a reservation."""
-    if is_stale('booking'):
-      raise BookingAuthError(f'Booking.com credentials are stale — run: {_AUTH_HINT}')
+    self._check_fresh()
     idempotency = f'{reservation_id}-{int(time.time() * 1000)}'
     payload = {'input': {
       'actor': self._actor(),
@@ -676,8 +594,7 @@ class _BookingClient:
 
   async def set_price(self, d: str, price: int, occupancy: int = 2) -> None:
     price = int(round(price))
-    if is_stale('booking'):
-      raise BookingAuthError(f'Booking.com credentials are stale — run: {_AUTH_HINT}')
+    self._check_fresh()
     if _is_past(d):
       return
     if (cached := self._cache.get(d)) and cached.price == price:
@@ -712,8 +629,7 @@ class _BookingClient:
     _log.info('booking set_price date=%s price=$%d', d, price)
 
   async def set_availability(self, d: str, available: bool) -> None:
-    if is_stale('booking'):
-      raise BookingAuthError(f'Booking.com credentials are stale — run: {_AUTH_HINT}')
+    self._check_fresh()
     if _is_past(d):
       return
     expected_rts = 1 if available else 0
@@ -743,8 +659,7 @@ class _BookingClient:
     _log.info('booking set_availability date=%s available=%s', d, available)
 
   async def set_rooms_to_sell(self, d: str, count: int) -> None:
-    if is_stale('booking'):
-      raise BookingAuthError(f'Booking.com credentials are stale — run: {_AUTH_HINT}')
+    self._check_fresh()
     if _is_past(d):
       return
     entry: dict = {

tests/test_service_base.py

@@ -0,0 +1,206 @@
+"""Unit tests for service_api._base (ServiceClientBase).
+
+Covers the cross-client invariants: stale fast-fail, error classification
+(timeouts are never auth errors), the attempt → refresh → retry-once request
+skeleton, and refresh orchestration bookkeeping.
+"""
+import asyncio
+
+import httpx
+import pytest
+
+from room_cal.service_api._base import ServiceClientBase
+from room_cal.service_api.credentials import clear_stale, is_stale, mark_stale
+from room_cal.service_api.types import ServiceAuthError
+
+
+class FakeAuthError(ServiceAuthError):
+  pass
+
+
+class _FakeClient(ServiceClientBase):
+  service_key = 'fake'
+  service_name = 'Fake'
+  auth_error_cls = FakeAuthError
+
+  def __init__(self):
+    super().__init__()
+    self.env = {'FAKE_TOKEN': 'initial'}
+    self.token = 'initial'
+    self.cleared_cache = 0
+
+  @classmethod
+  def _read_env(cls):  # avoid the filesystem
+    return _FakeClient._test_env
+
+  def _apply_env(self, env):
+    self.token = env.get('FAKE_TOKEN', self.token)
+
+  def clear_cache(self):
+    self.cleared_cache += 1
+
+
+_FakeClient._test_env = {'FAKE_TOKEN': 'initial'}
+
+
+def _resp(status, body=''):
+  return httpx.Response(status, text=body, request=httpx.Request('GET', 'https://fake.test/x'))
+
+
+@pytest.fixture(autouse=True)
+def _fresh():
+  clear_stale('fake')
+  _FakeClient._test_env = {'FAKE_TOKEN': 'initial'}
+  yield
+  clear_stale('fake')
+
+
+class TestCheckFresh:
+
+  def test_passes_when_fresh(self):
+    _FakeClient()._check_fresh()
+
+  def test_raises_when_stale(self):
+    mark_stale('fake')
+    with pytest.raises(FakeAuthError, match='credentials are stale'):
+      _FakeClient()._check_fresh()
+
+
+class TestRequest:
+
+  async def test_success_passthrough(self):
+    client = _FakeClient()
+    r = await client._request(lambda: asyncio.sleep(0, _resp(200, 'ok')), op='x')
+    assert r.status_code == 200
+
+  async def test_timeout_is_not_auth_error(self):
+    client = _FakeClient()
+    async def _attempt():
+      raise httpx.ReadTimeout('slow')
+    with pytest.raises(httpx.TimeoutException):
+      await client._request(_attempt, op='x')
+    assert not is_stale('fake')
+
+  async def test_transport_error_becomes_auth_error(self):
+    client = _FakeClient()
+    async def _attempt():
+      raise httpx.ConnectError('refused')
+    with pytest.raises(FakeAuthError, match='connection failed'):
+      await client._request(_attempt, op='x')
+
+  async def test_auth_status_refreshes_and_retries_once(self):
+    client = _FakeClient()
+    refreshes = []
+    async def _refresh():
+      refreshes.append(True)
+    client._auto_refresh = _refresh
+    responses = [_resp(401), _resp(200)]
+    r = await client._request(lambda: asyncio.sleep(0, responses.pop(0)), op='x')
+    assert r.status_code == 200
+    assert refreshes == [True]
+
+  async def test_second_auth_status_marks_stale_and_raises(self):
+    client = _FakeClient()
+    async def _refresh():
+      pass
+    client._auto_refresh = _refresh
+    responses = [_resp(401), _resp(403)]
+    with pytest.raises(FakeAuthError, match='session expired \\(HTTP 403\\)'):
+      await client._request(lambda: asyncio.sleep(0, responses.pop(0)), op='x')
+    assert is_stale('fake')
+
+  async def test_retry_sees_state_reloaded_by_refresh(self):
+    """The ses lesson: attempt closures must read client state per call."""
+    client = _FakeClient()
+    seen = []
+    async def _attempt():
+      seen.append(client.token)
+      return _resp(401 if len(seen) == 1 else 200)
+    async def _refresh():
+      client.token = 'refreshed'
+    client._auto_refresh = _refresh
+    await client._request(_attempt, op='x')
+    assert seen == ['initial', 'refreshed']
+
+  async def test_http_error_status_raises_plain(self):
+    client = _FakeClient()
+    with pytest.raises(httpx.HTTPStatusError):
+      await client._request(lambda: asyncio.sleep(0, _resp(500)), op='x')
+    assert not is_stale('fake')
+
+
+class TestAutoRefresh:
+
+  async def test_cheap_refresh_success_clears_stale(self):
+    client = _FakeClient()
+    mark_stale('fake')
+    async def _cheap():
+      return True
+    client._try_cheap_refresh = _cheap
+    await client._auto_refresh()
+    assert not is_stale('fake')
+
+  async def test_no_refresh_handler_marks_stale_and_raises(self):
+    client = _FakeClient()
+    with pytest.raises(FakeAuthError, match='no automatic refresh'):
+      await client._auto_refresh()
+    assert is_stale('fake')
+
+  async def test_full_refresh_reloads_env_and_clears_stale(self):
+    class _Refreshable(_FakeClient):
+      @staticmethod
+      def refresh_credentials():
+        _FakeClient._test_env = {'FAKE_TOKEN': 'refreshed'}
+
+    client = _Refreshable()
+    mark_stale('fake')
+    await client._auto_refresh()
+    assert client.token == 'refreshed'
+    assert client.cleared_cache == 1
+    assert not is_stale('fake')
+
+  async def test_failed_refresh_marks_stale(self):
+    class _Failing(_FakeClient):
+      @staticmethod
+      def refresh_credentials():
+        raise ServiceAuthError('login failed')
+
+    client = _Failing()
+    with pytest.raises(ServiceAuthError, match='login failed'):
+      await client._auto_refresh()
+    assert is_stale('fake')
+
+  async def test_ensure_refreshed_single_flight(self):
+    client = _FakeClient()
+    calls = []
+    async def _refresh():
+      calls.append(True)
+      await asyncio.sleep(0.01)
+    client._auto_refresh = _refresh
+    await asyncio.gather(client._ensure_refreshed(), client._ensure_refreshed(), client._ensure_refreshed())
+    assert calls == [True]
+
+
+class TestDefaults:
+
+  async def test_probe_fetches_today(self):
+    client = _FakeClient()
+    fetched = []
+    async def _get_calendar(a, b):
+      fetched.append((a, b))
+      return {}
+    client.get_calendar = _get_calendar
+    await client.probe()
+    assert len(fetched) == 1
+    assert fetched[0][0] == fetched[0][1]
+
+  def test_reload_from_env_applies_and_clears_cache(self):
+    client = _FakeClient()
+    _FakeClient._test_env = {'FAKE_TOKEN': 'new'}
+    client._reload_from_env()
+    assert client.token == 'new'
+    assert client.cleared_cache == 1
+
+  def test_require_env_raises_listing_missing_keys(self):
+    with pytest.raises(ValueError, match='FAKE_MISSING'):
+      _FakeClient._require_env({'FAKE_TOKEN': 'x'}, 'FAKE_TOKEN', 'FAKE_MISSING')