worldmonitor patterns: circuit breaker + refresh scheduler + progressive hydration

backend/jobs/appointment_followup.py

@@ -14,6 +14,7 @@
 
 from db import get_db
 from services.claude import generate_followup_message
+from utils.circuit_breaker import CircuitBreaker
 from services import cliniko as cliniko_adapter
 from services import square_appointments as square_adapter
 from services import nookal as nookal_adapter
@@ -41,6 +42,20 @@
 
 # All seven booking adapters. Stub adapters (hotdoc/jane/practicepal) register
 # with SUPPORTS_REBOOK=False so the job fails safe instead of 404ing.
+
+# Module-level breaker for Claude follow-up message generation
+_claude_breaker: CircuitBreaker | None = None
+
+
+def _get_followup_breaker(redis) -> CircuitBreaker:
+    global _claude_breaker
+    if _claude_breaker is None:
+        _claude_breaker = CircuitBreaker(
+            "claude-followup", redis, max_failures=2, cooldown_seconds=300, cache_ttl_seconds=0,
+        )
+    return _claude_breaker
+
+
 _BOOKING_ADAPTERS = {
     "cliniko": cliniko_adapter,
     "square": square_adapter,
@@ -257,7 +272,7 @@ async def run_appointment_followup_all_clients(arq_pool=None) -> None:  # type:
 
         for patient in lapsed:
             try:
-                await _process_lapsed_patient(db, client, patient, arq_pool)
+                await _process_lapsed_patient(db, client, patient, arq_pool, redis=arq_pool)
             except Exception as e:
                 logger.error(
                     "Follow-up failed for patient %s (client %s): %s",
@@ -272,6 +287,7 @@ async def _process_lapsed_patient(
     client: dict,
     patient: dict,
     arq_pool=None,  # type: ignore[no-untyped-def]
+    redis=None,
 ) -> None:
     """Check do-not-contact, generate message, send SMS, log to appointments table.
 
@@ -282,6 +298,9 @@ async def _process_lapsed_patient(
     ``practitioner_name`` and ``claim_type`` into message generation, routes the SMS
     through the durable ``send_sms_task`` (C4), and writes followup/practitioner/
     claim columns to ``appointments`` (columns added in migration 016).
+
+    Pass *redis* to enable circuit breaker protection on the Claude message
+    generation call.
     """
     booking_system = client.get("booking_system", "cliniko")
     adapter = get_booking_adapter(booking_system)
@@ -354,15 +373,42 @@ async def _process_lapsed_patient(
         return
 
     # --- Message generation (thread practitioner_name + claim_type) ---
+    # Wrap Claude call in circuit breaker (outer layer); @retry_on_failure
+    # is the inner layer applied inside generate_followup_message.
     claim = patient.get("claim") or {}
-    message = await generate_followup_message(
-        patient_name=patient.get("patient_name", "Patient"),
-        last_treatment=patient.get("treatment_type", "treatment"),
-        business_name=client.get("business_name", ""),
-        channel="sms",
-        practitioner_name=patient.get("practitioner_name"),
-        claim_type=claim.get("type"),
-    )
+    try:
+        if redis is not None:
+            breaker = _get_followup_breaker(redis)
+            message = await breaker.execute(
+                fn=lambda: generate_followup_message(
+                    patient_name=patient.get("patient_name", "Patient"),
+                    last_treatment=patient.get("treatment_type", "treatment"),
+                    business_name=client.get("business_name", ""),
+                    channel="sms",
+                    practitioner_name=patient.get("practitioner_name"),
+                    claim_type=claim.get("type"),
+                ),
+                default_value="",
+            )
+            if not message:
+                # Breaker returned default — log and skip SMS
+                logger.warning(
+                    "Claude breaker on cooldown for patient %s — skipping message generation",
+                    patient_id,
+                )
+                return
+        else:
+            message = await generate_followup_message(
+                patient_name=patient.get("patient_name", "Patient"),
+                last_treatment=patient.get("treatment_type", "treatment"),
+                business_name=client.get("business_name", ""),
+                channel="sms",
+                practitioner_name=patient.get("practitioner_name"),
+                claim_type=claim.get("type"),
+            )
+    except Exception as e:
+        logger.error("generate_followup_message failed for patient %s: %s", patient_id, e)
+        return
 
     sid, sent, error = await _enqueue_sms(arq_pool, phone, message, client.get("state", "NSW"))
 

backend/jobs/competitor_watch.py

@@ -13,12 +13,51 @@
     diff_structured,
 )
 from utils.retry import retry_on_failure
+from utils.circuit_breaker import CircuitBreaker
 
 logger = logging.getLogger(__name__)
 
+# Module-level breaker caches
+_snapshot_breakers: dict[str, CircuitBreaker] = {}
+_brief_breaker: CircuitBreaker | None = None
+
+
+def _get_snapshot_breaker(redis, url: str) -> CircuitBreaker:
+    name = f"competitor-snapshot-{hashlib.md5(url.encode()).hexdigest()[:12]}"
+    if name not in _snapshot_breakers:
+        _snapshot_breakers[name] = CircuitBreaker(
+            name, redis, max_failures=2, cooldown_seconds=300, cache_ttl_seconds=3600,
+        )
+    return _snapshot_breakers[name]
+
+
+def _get_brief_breaker(redis) -> CircuitBreaker:
+    global _brief_breaker
+    if _brief_breaker is None:
+        _brief_breaker = CircuitBreaker(
+            "claude-competitor-brief", redis, max_failures=2, cooldown_seconds=300, cache_ttl_seconds=0,
+        )
+    return _brief_breaker
+
+
+async def snapshot_website(redis, url: str) -> tuple[str, str, str]:
+    """Fetch a competitor URL with circuit breaker protection.
+
+    ``@retry_on_failure`` remains on ``_raw_snapshot_website`` as the inner layer.
+    """
+    if redis is None:
+        return await _raw_snapshot_website(url)
+
+    breaker = _get_snapshot_breaker(redis, url)
+    return await breaker.execute(
+        fn=lambda: _raw_snapshot_website(url),
+        default_value=("", "", ""),
+        cache_key=url,
+    )
+
 
 @retry_on_failure()
-async def snapshot_website(url: str) -> tuple[str, str, str]:
+async def _raw_snapshot_website(url: str) -> tuple[str, str, str]:
     """Fetch a competitor URL, strip non-content elements, and return (md5_hash, clean_text, raw_html).
 
     The raw HTML is returned so the caller can run ``extract_structured`` on
@@ -69,7 +108,7 @@ def _format_structured_diff(diff: list[dict]) -> list[str]:
     return lines
 
 
-async def detect_changes(client_id: str, competitor_url: str) -> dict | None:
+async def detect_changes(redis, client_id: str, competitor_url: str) -> dict | None:
     """Compare the latest snapshot against a fresh fetch.
 
     Returns None when the content is unchanged or the fetch failed.
@@ -90,7 +129,7 @@ async def detect_changes(client_id: str, competitor_url: str) -> dict | None:
         .execute()
     )
 
-    new_hash, new_text, raw_html = await snapshot_website(competitor_url)
+    new_hash, new_text, raw_html = await snapshot_website(redis, competitor_url)
     if not new_hash:
         return None
 
@@ -143,12 +182,27 @@ def _parse_threat_level(brief: str) -> str:
     return "MEDIUM"
 
 
+async def _generate_brief_safe(redis, business_name: str, changes_summary: str) -> str:
+    """Generate competitor brief with circuit breaker + retry protection.
+
+    ``@retry_on_failure`` remains on ``_raw_generate_brief`` as the inner layer.
+    """
+    if redis is None:
+        return await _raw_generate_brief(business_name, changes_summary)
+
+    breaker = _get_brief_breaker(redis)
+    return await breaker.execute(
+        fn=lambda: _raw_generate_brief(business_name, changes_summary),
+        default_value="",
+    )
+
+
 @retry_on_failure()
-async def _generate_brief_safe(business_name: str, changes_summary: str) -> str:
+async def _raw_generate_brief(business_name: str, changes_summary: str) -> str:
     return await generate_competitor_brief(business_name, changes_summary)
 
 
-async def run_competitor_snapshots_all_clients() -> None:
+async def run_competitor_snapshots_all_clients(ctx: dict | None = None) -> None:
     """APScheduler job — runs Sunday 10pm AEST.
 
     For every client with ``'competitor_watch'`` in ``active_jobs``:
@@ -164,6 +218,7 @@ async def run_competitor_snapshots_all_clients() -> None:
     Each client is wrapped in its own try/except so one failure never crashes
     the entire job run.
     """
+    redis = ctx.get("redis") if ctx else None
     db = get_db()
 
     resp = (
@@ -194,7 +249,7 @@ async def run_competitor_snapshots_all_clients() -> None:
         try:
             changes = []
             for url in competitor_urls:
-                result = await detect_changes(client_id, url)
+                result = await detect_changes(redis, client_id, url)
                 if result is not None:
                     changes.append({"url": url, **result})
 
@@ -220,7 +275,7 @@ async def run_competitor_snapshots_all_clients() -> None:
                     )
             changes_summary = "\n\n".join(parts)
 
-            brief = await _generate_brief_safe(business_name, changes_summary)
+            brief = await _generate_brief_safe(redis, business_name, changes_summary)
             threat_level = _parse_threat_level(brief)
 
             for c in changes:

backend/jobs/menu_sync.py

@@ -15,6 +15,7 @@
 import httpx
 
 from db import get_db
+from utils.circuit_breaker import CircuitBreaker
 
 logger = logging.getLogger(__name__)
 
@@ -24,6 +25,28 @@
 RETRY_MAX_ATTEMPTS = 2
 RETRY_DELAY_SECONDS = 5
 
+# Module-level breaker caches
+_square_breakers: dict[str, CircuitBreaker] = {}
+_gbp_breakers: dict[str, CircuitBreaker] = {}
+
+
+def _get_square_breaker(redis, location_id: str) -> CircuitBreaker:
+    name = f"square-catalog-{location_id}"
+    if name not in _square_breakers:
+        _square_breakers[name] = CircuitBreaker(
+            name, redis, max_failures=2, cooldown_seconds=300, cache_ttl_seconds=0,
+        )
+    return _square_breakers[name]
+
+
+def _get_gbp_breaker(redis, location_id: str) -> CircuitBreaker:
+    name = f"gbp-menu-{location_id}"
+    if name not in _gbp_breakers:
+        _gbp_breakers[name] = CircuitBreaker(
+            name, redis, max_failures=2, cooldown_seconds=300, cache_ttl_seconds=0,
+        )
+    return _gbp_breakers[name]
+
 
 # ---------------------------------------------------------------------------
 # Content hash + canonical store
@@ -120,9 +143,13 @@ async def _retry_push(label: str, coro_fn):
 
 
 async def _push_to_square(
-    location: dict, client: dict, menu_item: dict, link: dict | None
+    redis, location: dict, client: dict, menu_item: dict, link: dict | None
 ) -> dict:
-    """Push a menu item to Square catalog via per-client OAuth token."""
+    """Push a menu item to Square catalog via per-client OAuth token.
+
+    Circuit breaker wraps the outbound Square API call. ``_retry_push``
+    remains as the inner layer (transient 5xx retries).
+    """
     from services.square_oauth import get_valid_token
     from services.square_catalog import upsert_item as square_upsert
 
@@ -153,13 +180,23 @@ async def _do():
             "external_version": result.get("version"),
         }
 
-    return await _retry_push("Square", _do)
+    if redis is None:
+        return await _retry_push("Square", _do)
+
+    breaker = _get_square_breaker(redis, location.get("id", "unknown"))
+    return await breaker.execute(
+        fn=lambda: _retry_push("Square", _do),
+        default_value={"synced": False, "message": "Circuit breaker: Square temporarily unavailable"},
+    )
 
 
 async def _push_to_gbp(
-    location: dict, client: dict, menu_item: dict, link: dict | None
+    redis, location: dict, client: dict, menu_item: dict, link: dict | None
 ) -> dict:
-    """Push a menu item to GBP Menu API v1 using location's gbp_account_id."""
+    """Push a menu item to GBP Menu API v1 using location's gbp_account_id.
+
+    Circuit breaker wraps the outbound GBP API call.
+    """
     from services.crypto import decrypt
 
     account_id = location.get("gbp_account_id")
@@ -191,14 +228,22 @@ async def _do():
             resp.raise_for_status()
         return {"synced": True, "message": "Synced to GBP"}
 
-    return await _retry_push("GBP", _do)
+    if redis is None:
+        return await _retry_push("GBP", _do)
+
+    breaker = _get_gbp_breaker(redis, location.get("id", "unknown"))
+    return await breaker.execute(
+        fn=lambda: _retry_push("GBP", _do),
+        default_value={"synced": False, "message": "Circuit breaker: GBP temporarily unavailable"},
+    )
 
 
 async def reconcile_item(
     location: dict,
     client: dict,
     menu_item: dict,
     skip_platform: str | None = None,
+    redis=None,
 ) -> dict:
     """Reconcile a canonical menu item to its configured sync targets.
 
@@ -241,9 +286,9 @@ async def reconcile_item(
 
         # Push to platform
         if target == "square":
-            result = await _push_to_square(location, client, menu_item, link)
+            result = await _push_to_square(redis, location, client, menu_item, link)
         elif target == "gbp":
-            result = await _push_to_gbp(location, client, menu_item, link)
+            result = await _push_to_gbp(redis, location, client, menu_item, link)
         elif target == "website":
             result = {"synced": False, "message": "Website CMS sync not yet implemented"}
         elif target in ("ubereats", "doordash", "lightspeed"):
@@ -317,12 +362,16 @@ async def sync_menu_item(
     location_id: str | None = None,
     item: dict | None = None,
     origin: str = "sheets",
+    redis=None,
 ) -> dict:
     """Sync a menu item: upsert canonical + reconcile to targets.
 
     New signature: ``sync_menu_item(client_id, location_id, item, origin='sheets')``.
     Backward-compat: ``sync_menu_item(client_id, item)`` resolves the client's
     ``is_default`` location.
+
+    Pass *redis* (arq's ``ctx["redis"]``) to enable circuit breaker protection
+    on outbound Square/GBP calls.
     """
     # Backward-compat: sync_menu_item(client_id, item) — 2 positional args
     if item is None and location_id is not None:
@@ -383,7 +432,7 @@ async def sync_menu_item(
     menu_item = await upsert_canonical(client_id, location["id"], item, origin)
 
     # Reconcile
-    result = await reconcile_item(location, client, menu_item)
+    result = await reconcile_item(location, client, menu_item, redis=redis)
 
     return {"status": "completed", "targets": result.get("targets", {})}
 
@@ -392,7 +441,7 @@ async def sync_menu_item(
 # Square inbound (bi-directional)
 # ---------------------------------------------------------------------------
 
-async def apply_square_inbound(client_id: str, changed_objects: list[dict]) -> dict:
+async def apply_square_inbound(client_id: str, changed_objects: list[dict], redis=None) -> dict:
     """Apply Square catalog changes to the canonical store.
 
     For each changed Square ITEM object:
@@ -475,7 +524,7 @@ async def apply_square_inbound(client_id: str, changed_objects: list[dict]) -> d
             .execute()
         )
         if loc_resp.data and client_resp.data:
-            await reconcile_item(loc_resp.data, client_resp.data, updated_item, skip_platform="square")
+            await reconcile_item(loc_resp.data, client_resp.data, updated_item, skip_platform="square", redis=redis)
 
         applied += 1