service_api: route all refresh paths and body-level auth through the base

- Split _auto_refresh into cheap-attempt + _full_refresh; _ensure_refreshed(full=True)
  forces the full script (skipping the cheap hook) under the same single-flight task,
  waiting out any in-flight cheap-allowed refresh first.
- Booking send_message's 2FA retry and ServiceAPI.refresh_credentials now go through
  _ensure_refreshed(full=True) instead of hand-rolled run_in_executor calls, gaining
  single-flight and stale-flag bookkeeping. The classmethod fallback remains only for
  services with no live client (failed to load at startup).
- New _auth_shaped(response) hook on the base classifies auth failures (default:
  HTTP 400/401/403). Airbnb overrides it to recognize GQL error extensions on
  HTTP 200, so body-signaled session expiry gets the same refresh -> retry-once ->
  stale sequence instead of failing straight to stale (replaces _check_auth).

room_cal/service_api/_airbnb.py

@@ -14,11 +14,10 @@ import logging
 
 from ._scripts import run_refresh_script
 from room_cal._utils import get_config
-from ._base import ServiceClientBase, _AUTH_HINT
+from ._base import ServiceClientBase
 from ._month_cache import MonthCache, TTLCache
 from ._registry import register_service
 from ._utils import month_bounds, months_in_range
-from .credentials import mark_stale
 from .types import CalendarEvent, DayInfo, ServiceAuthError, Reservation
 
 _log = logging.getLogger(__name__)
@@ -200,13 +199,21 @@ class _AirbnbClient(ServiceClientBase):
       h.update(extra)
     return h
 
-  def _check_auth(self, body: dict) -> None:
-    if any(
+  def _auth_shaped(self, r: httpx.Response) -> bool:
+    # Airbnb also signals session expiry as GQL errors on HTTP 200; route those
+    # through the base's refresh → retry-once → stale sequence too.
+    if super()._auth_shaped(r):
+      return True
+    if r.status_code != 200:
+      return False
+    try:
+      body = r.json()
+    except ValueError:
+      return False
+    return any(
       e.get('extensions', {}).get('response', {}).get('status') in (400, 401, 403)
       for e in body.get('errors', [])
-    ):
-      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:
     url = f'{_BASE}/{op}/{h}'
@@ -220,9 +227,7 @@ class _AirbnbClient(ServiceClientBase):
     async def _attempt() -> httpx.Response:
       return await self._http.get(url, params=params, headers=self._headers(referer))
     r = await self._request(_attempt, op=op)
-    body = r.json()
-    self._check_auth(body)
-    return body
+    return r.json()
 
   async def _mutate(self, op: str, h: str, variables: dict, referer: str) -> dict:
     url = f'{_BASE}/{op}/{h}?operationName={op}&locale=en&currency=USD'
@@ -234,9 +239,7 @@ class _AirbnbClient(ServiceClientBase):
         headers=self._headers(referer, {'Origin': 'https://www.airbnb.com'}),
       )
     r = await self._request(_attempt, op=op)
-    body = r.json()
-    self._check_auth(body)
-    return body
+    return r.json()
 
   async def _fetch_month_days(self, month: int, year: int) -> list[dict]:
     start, end = month_bounds(month, year)

room_cal/service_api/_base.py

@@ -36,6 +36,7 @@ class ServiceClientBase:
   def __init__(self) -> None:
     self._refresh_lock = asyncio.Lock()
     self._refresh_task: asyncio.Task | None = None
+    self._refresh_is_full = False
 
   # ── credentials and staleness ──────────────────────────────────────────────
 
@@ -71,15 +72,24 @@ class ServiceClientBase:
 
   # ── request skeleton ───────────────────────────────────────────────────────
 
+  def _auth_shaped(self, r: httpx.Response) -> bool:
+    """True when a response signals an auth failure. Default: HTTP status.
+
+    Override to also recognize body-level signals (e.g. Airbnb's GQL error
+    extensions on HTTP 200) so they get the same refresh → retry-once → stale
+    sequence as status-shaped failures.
+    """
+    return r.status_code in _AUTH_STATUSES
+
   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."""
+    """Run attempt; on an auth-shaped response, 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',
+    if self._auth_shaped(r):
+      _log.warning('%s %s auth-shaped response (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:
+      if self._auth_shaped(r):
         mark_stale(self.service_key)
         raise self.auth_error_cls(
           f'{self.service_name} session expired (HTTP {r.status_code}) — run: {_AUTH_HINT}')
@@ -100,13 +110,29 @@ class ServiceClientBase:
 
   # ── 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 _ensure_refreshed(self, *, full: bool = False) -> None:
+    """Single-flight refresh: concurrent callers await the same in-progress attempt.
+
+    full=True skips the cheap-refresh hook and forces the class's
+    refresh_credentials — for failures a cheap refresh can't fix even though
+    the session looks alive (e.g. Booking's expired messaging-2FA grant). If a
+    cheap-allowed refresh is already in flight, it is allowed to finish and a
+    full refresh runs after it.
+    """
+    while True:
+      async with self._refresh_lock:
+        if self._refresh_task is None or self._refresh_task.done():
+          coro = self._full_refresh() if full else self._auto_refresh()
+          self._refresh_task = asyncio.create_task(coro)
+          self._refresh_is_full = full
+        task = self._refresh_task
+        satisfied = self._refresh_is_full or not full
+      if satisfied:
+        await task
+        return
+      # need a full refresh but a cheap-allowed one is in flight — let it
+      # finish (ignoring its outcome), then loop to start or join a full one
+      await asyncio.gather(task, return_exceptions=True)
 
   async def _try_cheap_refresh(self) -> bool:
     """Attempt a lightweight in-process session refresh; True on success.
@@ -122,12 +148,21 @@ class ServiceClientBase:
       _log.info('%s session refreshed in place', self.service_name)
       clear_stale(self.service_key)
       return
+    _log.warning('%s: falling back to full credential refresh', self.service_name)
+    await self._full_refresh()
+
+  async def _full_refresh(self) -> None:
+    """Run the class's refresh_credentials in an executor, then reload from env.
+
+    Owns the stale-flag bookkeeping for the full-refresh path; never attempts
+    the cheap refresh. Reach it via _ensure_refreshed(full=True) so concurrent
+    callers share one attempt.
+    """
     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()

room_cal/service_api/_booking.py

@@ -11,7 +11,7 @@ import httpx
 
 from ._scripts import _booking_lock, run_refresh_script
 from room_cal._utils import get_config
-from ._base import ServiceClientBase, _AUTH_HINT
+from ._base import ServiceClientBase
 from ._month_cache import MonthCache
 from ._registry import register_service
 from ._utils import month_bounds, months_in_range
@@ -584,10 +584,10 @@ class _BookingClient(ServiceClientBase):
       if 'Internal Server Error' not in str(e):
         raise
       # GraphQL ISE on send typically means the messaging 2FA session has expired.
-      # Re-trigger the SMS challenge and retry once.
+      # The session itself is alive, so a cheap refresh would succeed without
+      # re-priming 2FA — force the full refresh (SMS challenge included).
       _log.warning('booking send_message ISE on attempt 1 res=%s idempotency=%s — re-triggering credential refresh', reservation_id, idempotency)
-      await asyncio.get_running_loop().run_in_executor(None, _BookingClient.refresh_credentials)
-      self._reload_from_env()
+      await self._ensure_refreshed(full=True)
       raw = await self._msg_gql('postFreeTextMessage', _POST_FREE_TEXT_MUTATION, payload, attempt=2)
       msg = ((raw.get('data') or {}).get('chatMessage') or {}).get('postFreeTextMessage') or {}
       result = msg.get('message') or {}

room_cal/service_api/client.py

@@ -3,6 +3,7 @@ import logging
 from datetime import date
 
 from . import _airbnb, _booking  # noqa: F401 — import registers each client via @register_service
+from ._base import ServiceClientBase
 from ._protocols import ChatClient, ServiceClient
 from ._registry import client_classes
 from .types import DayInfo, ServiceAuthError
@@ -60,10 +61,20 @@ class ServiceAPI:
 
 
   async def refresh_credentials(self, key: str) -> None:
-    """Trigger credential refresh for the named service. Raises ServiceAuthError on failure."""
+    """Trigger credential refresh for the named service. Raises ServiceAuthError on failure.
+
+    A live client refreshes through the base's single-flight orchestration
+    (full=True: a user-driven refresh must run the real script, not the cheap
+    session bump), so it can't race a concurrently triggered auto-refresh. A
+    service with no live client (failed to load at startup) falls back to the
+    classmethod so newly supplied credentials can enable it.
+    """
+    client = self._clients.get(key)
+    if isinstance(client, ServiceClientBase):
+      await client._ensure_refreshed(full=True)
+      return
     cls = client_classes().get(key)
     if cls is None or not hasattr(cls, 'refresh_credentials'):
-      from .types import ServiceAuthError
       raise ServiceAuthError(f'No automatic refresh handler for service: {key}')
     loop = asyncio.get_running_loop()
     await loop.run_in_executor(None, cls.refresh_credentials)

tests/test_service_base.py

@@ -122,6 +122,36 @@ class TestRequest:
     await client._request(_attempt, op='x')
     assert seen == ['initial', 'refreshed']
 
+  async def test_body_shaped_auth_refreshes_and_retries_once(self):
+    """A client overriding _auth_shaped gets the same retry as status-shaped auth."""
+    class _BodyAuthClient(_FakeClient):
+      def _auth_shaped(self, r):
+        return super()._auth_shaped(r) or 'expired' in r.text
+
+    client = _BodyAuthClient()
+    refreshes = []
+    async def _refresh():
+      refreshes.append(True)
+    client._auto_refresh = _refresh
+    responses = [_resp(200, 'expired'), _resp(200, 'ok')]
+    r = await client._request(lambda: asyncio.sleep(0, responses.pop(0)), op='x')
+    assert r.text == 'ok'
+    assert refreshes == [True]
+
+  async def test_body_shaped_auth_twice_marks_stale_and_raises(self):
+    class _BodyAuthClient(_FakeClient):
+      def _auth_shaped(self, r):
+        return super()._auth_shaped(r) or 'expired' in r.text
+
+    client = _BodyAuthClient()
+    async def _refresh():
+      pass
+    client._auto_refresh = _refresh
+    responses = [_resp(200, 'expired'), _resp(200, 'expired')]
+    with pytest.raises(FakeAuthError, match='session expired'):
+      await client._request(lambda: asyncio.sleep(0, responses.pop(0)), op='x')
+    assert is_stale('fake')
+
   async def test_http_error_status_raises_plain(self):
     client = _FakeClient()
     with pytest.raises(httpx.HTTPStatusError):
@@ -180,6 +210,69 @@ class TestAutoRefresh:
     await asyncio.gather(client._ensure_refreshed(), client._ensure_refreshed(), client._ensure_refreshed())
     assert calls == [True]
 
+  async def test_full_skips_cheap_refresh(self):
+    class _Refreshable(_FakeClient):
+      @staticmethod
+      def refresh_credentials():
+        _FakeClient._test_env = {'FAKE_TOKEN': 'refreshed'}
+
+    client = _Refreshable()
+    cheap_calls = []
+    async def _cheap():
+      cheap_calls.append(True)
+      return True
+    client._try_cheap_refresh = _cheap
+    await client._ensure_refreshed(full=True)
+    assert cheap_calls == []
+    assert client.token == 'refreshed'
+
+  async def test_full_marks_stale_when_no_refresh_handler(self):
+    client = _FakeClient()
+    with pytest.raises(FakeAuthError, match='no automatic refresh'):
+      await client._ensure_refreshed(full=True)
+    assert is_stale('fake')
+
+  async def test_full_runs_after_inflight_cheap_refresh(self):
+    class _Refreshable(_FakeClient):
+      @staticmethod
+      def refresh_credentials():
+        pass
+
+    client = _Refreshable()
+    order = []
+    started = asyncio.Event()
+    async def _cheap():
+      order.append('cheap')
+      started.set()
+      await asyncio.sleep(0.01)
+      return True
+    client._try_cheap_refresh = _cheap
+    async def _full():
+      order.append('full')
+    client._full_refresh = _full
+
+    async def _late_full():
+      await started.wait()
+      await client._ensure_refreshed(full=True)
+
+    await asyncio.gather(client._ensure_refreshed(), _late_full())
+    assert order == ['cheap', 'full']
+
+  async def test_full_callers_share_one_attempt(self):
+    class _Refreshable(_FakeClient):
+      @staticmethod
+      def refresh_credentials():
+        pass
+
+    client = _Refreshable()
+    calls = []
+    async def _full():
+      calls.append(True)
+      await asyncio.sleep(0.01)
+    client._full_refresh = _full
+    await asyncio.gather(client._ensure_refreshed(full=True), client._ensure_refreshed(full=True))
+    assert calls == [True]
+
 
 class TestDefaults:
 

tests/test_service_client_conformance.py

@@ -390,3 +390,61 @@ class TestNoOpWrites:
     d, _price, available = adapter.seed_no_op(client)
     await client.set_availability(d, available)
     assert captured == []
+
+
+# ── Airbnb body-shaped auth (GQL errors on HTTP 200) ──────────────────────────
+
+_GQL_AUTH_BODY = {'errors': [{'extensions': {'response': {'status': 401}}}]}
+
+
+class TestAirbnbBodyAuth:
+  """Airbnb signals session expiry as GQL error extensions on HTTP 200; the
+  _auth_shaped override must route these through the same refresh → retry-once
+  → stale sequence as status-shaped failures."""
+
+  async def test_body_auth_refreshes_and_retries_once(self):
+    adapter = _airbnb_adapter()
+    clear_stale(adapter.name)
+    client, captured = adapter.make_client([
+      httpx.Response(200, json=_GQL_AUTH_BODY),
+      httpx.Response(200, json={'data': {}}),
+    ])
+    refreshed = []
+
+    async def fake_refresh():
+      adapter.mutate_state(client)
+      refreshed.append(True)
+
+    client._auto_refresh = fake_refresh
+    body = await adapter.raw_request(client)
+    assert body == {'data': {}}
+    assert refreshed == [True]
+    assert len(captured) == 2
+    assert not is_stale(adapter.name)
+
+  async def test_body_auth_twice_marks_stale_and_raises(self):
+    adapter = _airbnb_adapter()
+    clear_stale(adapter.name)
+    client, _ = adapter.make_client([
+      httpx.Response(200, json=_GQL_AUTH_BODY),
+      httpx.Response(200, json=_GQL_AUTH_BODY),
+    ])
+
+    async def fake_refresh():
+      pass
+
+    client._auto_refresh = fake_refresh
+    with pytest.raises(adapter.auth_error_cls, match='session expired'):
+      await adapter.raw_request(client)
+    assert is_stale(adapter.name)
+    clear_stale(adapter.name)
+
+  async def test_non_auth_gql_errors_pass_through(self):
+    adapter = _airbnb_adapter()
+    clear_stale(adapter.name)
+    body = {'errors': [{'message': 'validation failed'}], 'data': {}}
+    client, captured = adapter.make_client([httpx.Response(200, json=body)])
+    result = await adapter.raw_request(client)
+    assert result == body
+    assert len(captured) == 1
+    assert not is_stale(adapter.name)