igdb: auto-refresh twitch token; category -> game_type filter

Static app tokens expire after ~60 days -> every lookup failed with
401. With igdb-client-secret set, the bot fetches the token via
client-credentials OAuth itself, refreshes a day before expiry and
retries once on 401; a static igdb-access-token still works.

IGDB renamed games.category to game_type: the category = 0 filter in
search_games silently matched nothing even with a valid token.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

IGDB_SETUP.md

@@ -20,13 +20,6 @@ The bot now supports real-time video game information through IGDB (Internet Gam
    - **Category**: Select appropriate category
 3. Note down your **Client ID**
 4. Generate a **Client Secret**
-5. Get an access token using this curl command:
-   ```bash
-   curl -X POST 'https://id.twitch.tv/oauth2/token' \
-   -H 'Content-Type: application/x-www-form-urlencoded' \
-   -d 'client_id=YOUR_CLIENT_ID&client_secret=YOUR_CLIENT_SECRET&grant_type=client_credentials'
-   ```
-6. Save the `access_token` from the response
 
 ### 2. Configure the Bot
 
@@ -35,6 +28,25 @@ Update your `config.toml` file:
 ```toml
 # IGDB Configuration for game information
 igdb-client-id = "your_actual_client_id_here"
+igdb-client-secret = "your_actual_client_secret_here"
+enable-game-info = true
+```
+
+With the client secret configured, the bot fetches an app access token from
+Twitch itself and refreshes it automatically before it expires (Twitch app
+tokens live ~60 days) — no manual token handling needed.
+
+Alternatively, a static token still works (legacy setup — it expires after
+~60 days and then game lookups fail with 401 until you replace it):
+
+```bash
+curl -X POST 'https://id.twitch.tv/oauth2/token' \
+-H 'Content-Type: application/x-www-form-urlencoded' \
+-d 'client_id=YOUR_CLIENT_ID&client_secret=YOUR_CLIENT_SECRET&grant_type=client_credentials'
+```
+
+```toml
+igdb-client-id = "your_actual_client_id_here"
 igdb-access-token = "your_actual_access_token_here"
 enable-game-info = true
 ```
@@ -99,7 +111,8 @@ The integration provides two OpenAI functions:
    - Verify client ID and access token are set
 
 2. **Authentication errors**
-   - Regenerate access token (they expire)
+   - Prefer `igdb-client-secret` — the bot then refreshes tokens itself
+   - With a static `igdb-access-token`: regenerate it (they expire)
    - Verify client ID matches your Twitch app
 
 3. **No game results**

Makefile

@@ -83,3 +83,6 @@ ci: install-dev all-checks ## Full CI pipeline (install deps and run all checks)
 # Deploy targets (SPEC-007)
 deploy: ## Deploy a tag to a host: make deploy HOST=ggg TAG=v3.0.0
 	bash deploy/deploy.sh $(HOST) $(TAG)
+
+backup: ## Back up a local bot.db: make backup DB=history/bot.db DIR=backups
+	uv run python deploy/backup_db.py $(DB) $(DIR) $(or $(KEEP),14)

config.toml

@@ -14,7 +14,11 @@ system = "You are a smart AI assistant with access to real-time video game infor
 
 # IGDB Configuration for game information
 igdb-client-id = "YOUR_IGDB_CLIENT_ID"
-igdb-access-token = "YOUR_IGDB_ACCESS_TOKEN"
+# With the Twitch app client secret set, the bot fetches and refreshes the
+# access token itself (recommended). A static igdb-access-token still works
+# but expires after ~60 days.
+igdb-client-secret = "YOUR_IGDB_CLIENT_SECRET"
+# igdb-access-token = "YOUR_IGDB_ACCESS_TOKEN"
 enable-game-info = true
 
 # --- operator / safety (SPEC-003, SPEC-006) ---
@@ -73,3 +77,6 @@ enable-game-info = true
 # url-max-chars = 6000          # text handed to the model
 # url-max-images = 2            # page images into the vision cache
 # url-daily-per-user = 20
+# Ops (SPEC-012): consecutive OpenAI failures before a staff alert
+# api-error-alert-threshold = 5
+# Backups: cron runs deploy/backup_db.py daily -> ~/backups/<bot>/ (keep 14)

deploy/backup_db.py

@@ -0,0 +1,80 @@
+#!/usr/bin/env python3
+"""Consistent, rotated bot.db backups (SPEC-012 OPS-13).
+
+Run from cron on each host. Uses the sqlite3 online-backup API so the
+snapshot is consistent even while the bot writes (WAL-safe), gzips it,
+and keeps the newest N. Stdlib only.
+
+Usage: python3 backup_db.py <bot.db> <backup-dir> [keep]
+"""
+
+import gzip
+import os
+import shutil
+import sqlite3
+import sys
+import tempfile
+import time
+from pathlib import Path
+
+DEFAULT_KEEP = 14
+BACKUP_GLOB = "bot-*.db.gz"
+
+
+def snapshot(src: Path, dest_gz: Path) -> None:
+    """Write a consistent gzipped snapshot of src to dest_gz (OPS-13)."""
+    fd, tmp_path = tempfile.mkstemp(suffix=".db", dir=str(dest_gz.parent))
+    os.close(fd)
+    tmp = Path(tmp_path)
+    try:
+        source = sqlite3.connect(str(src))
+        try:
+            target = sqlite3.connect(str(tmp))
+            try:
+                source.backup(target)  # atomic, WAL-safe online backup
+            finally:
+                target.close()
+        finally:
+            source.close()
+        with open(tmp, "rb") as raw, gzip.open(str(dest_gz), "wb") as gz:
+            shutil.copyfileobj(raw, gz)
+        os.chmod(dest_gz, 0o600)  # conversation data
+    finally:
+        tmp.unlink(missing_ok=True)
+
+
+def victims(existing: list, keep: int) -> list:
+    """Given backup paths (any order), return the ones to delete, oldest first (OPS-14)."""
+    ordered = sorted(existing)  # timestamped names sort chronologically
+    return ordered[: max(0, len(ordered) - keep)]
+
+
+def rotate(backup_dir: Path, keep: int) -> int:
+    removed = 0
+    for path in victims(list(backup_dir.glob(BACKUP_GLOB)), keep):
+        Path(path).unlink(missing_ok=True)
+        removed += 1
+    return removed
+
+
+def main() -> int:
+    if len(sys.argv) < 3:
+        print("usage: backup_db.py <bot.db> <backup-dir> [keep]", file=sys.stderr)
+        return 2
+    src = Path(sys.argv[1]).expanduser()
+    backup_dir = Path(sys.argv[2]).expanduser()
+    keep = int(sys.argv[3]) if len(sys.argv) > 3 else DEFAULT_KEEP
+    if not src.exists():
+        print(f"backup: source {src} missing", file=sys.stderr)
+        return 1
+    backup_dir.mkdir(parents=True, exist_ok=True)
+    stamp = time.strftime("%Y%m%d-%H%M%S", time.gmtime())
+    dest = backup_dir / f"bot-{stamp}.db.gz"
+    snapshot(src, dest)
+    removed = rotate(backup_dir, keep)
+    print(f"backup: wrote {dest.name} ({dest.stat().st_size} bytes), rotated {removed} old")
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())

fjerkroa_bot/discord_bot.py

@@ -89,6 +89,7 @@ class FjerkroaBot(commands.Bot):
         self.tasks_enabled = True
         self.quiet_until = 0.0
         self._staff_alert_times: deque = deque()
+        self._consecutive_api_errors = 0  # OPS-16
 
         self.init_observer()
         self.init_aichannels()
@@ -498,6 +499,14 @@ class FjerkroaBot(commands.Bot):
             return True, False
         return False, bool(verdict.get("factual", False))
 
+    async def _note_api_error(self, err: Exception) -> None:
+        """Count consecutive failures; alert staff once at threshold (OPS-16)."""
+        self._consecutive_api_errors += 1
+        logging.warning(f"responder call failed ({self._consecutive_api_errors} in a row): {repr(err)}")
+        threshold = int(self.config.get("api-error-alert-threshold", 5))
+        if self._consecutive_api_errors == threshold:
+            await self.send_staff_alert(f"⚠️ {threshold} consecutive API errors — the bot may be down. Last: {str(err)[:200]}")
+
     async def send_message_with_typing(self, airesponder, channel, message):
         """Send the user message to the AI responder with typing animation in discord"""
         async with channel.typing():
@@ -603,8 +612,15 @@ class FjerkroaBot(commands.Bot):
         # Get the AI responder based on the channel name
         airesponder = self.get_ai_responder(channel_name)
 
-        # Send the user message to the AI responder, with typing indicators
+        # Send the user message to the AI responder, with typing indicators.
+        # A raised call = a broken API path (cf. the gpt-5.6 tools incident):
+        # count it, alert staff at threshold, never crash the handler (OPS-16).
+        try:
             response = await self.send_message_with_typing(airesponder, channel, message)
+        except Exception as err:
+            await self._note_api_error(err)
+            return
+        self._consecutive_api_errors = 0
 
         # SAF/OPS gates between model proposal and delivery
         await self._apply_response_gates(message, response)

fjerkroa_bot/igdblib.py

@@ -1,27 +1,60 @@
 import logging
+import time
 from functools import cache
 from typing import Any, Dict, List, Optional
 
 import requests
 
+TWITCH_OAUTH_URL = "https://id.twitch.tv/oauth2/token"
+# Refresh this long before Twitch expires the token (app tokens live ~60 days)
+TOKEN_REFRESH_MARGIN = 86400
+
 
 class IGDBQuery(object):
-    def __init__(self, client_id, igdb_api_key):
+    def __init__(self, client_id, igdb_api_key=None, client_secret=None):
         self.client_id = client_id
         self.igdb_api_key = igdb_api_key
+        self.client_secret = client_secret
+        # Unknown for statically configured tokens; set after each refresh
+        self._token_expires_at = None
+
+    def _refresh_token(self):
+        response = requests.post(
+            TWITCH_OAUTH_URL,
+            params={"client_id": self.client_id, "client_secret": self.client_secret, "grant_type": "client_credentials"},
+        )
+        response.raise_for_status()
+        data = response.json()
+        self.igdb_api_key = data["access_token"]
+        self._token_expires_at = time.time() + data.get("expires_in", 0) - TOKEN_REFRESH_MARGIN
+        logging.info("IGDB: refreshed Twitch app access token")
+
+    def _ensure_token(self):
+        if not self.client_secret:
+            return
+        if not self.igdb_api_key or (self._token_expires_at is not None and time.time() >= self._token_expires_at):
+            self._refresh_token()
 
     def send_igdb_request(self, endpoint, query_body):
         igdb_url = f"https://api.igdb.com/v4/{endpoint}"
-        headers = {"Client-ID": self.client_id, "Authorization": f"Bearer {self.igdb_api_key}"}
 
         try:
-            response = requests.post(igdb_url, headers=headers, data=query_body)
+            self._ensure_token()
+            response = self._post_igdb(igdb_url, query_body)
+            if self.client_secret and response.status_code == 401:
+                # Token expired server-side (e.g. statically configured) — refresh and retry once
+                self._refresh_token()
+                response = self._post_igdb(igdb_url, query_body)
             response.raise_for_status()
             return response.json()
         except requests.RequestException as e:
             print(f"Error during IGDB API request: {e}")
             return None
 
+    def _post_igdb(self, igdb_url, query_body):
+        headers = {"Client-ID": self.client_id, "Authorization": f"Bearer {self.igdb_api_key}"}
+        return requests.post(igdb_url, headers=headers, data=query_body)
+
     @staticmethod
     def build_query(fields, filters=None, limit=10, offset=None):
         query = f"fields {','.join(fields) if fields is not None and len(fields) > 0 else '*'}; limit {limit};"
@@ -79,7 +112,7 @@ class IGDBQuery(object):
                 "id",
                 "name",
                 "alternative_names",
-                "category",
+                "game_type",
                 "release_dates",
                 "franchise",
                 "language_supports",
@@ -120,7 +153,7 @@ class IGDBQuery(object):
                     "themes.name",
                     "cover.url",
                 ],
-                additional_filters={"category": "= 0"},  # Main games only
+                additional_filters={"game_type": "= 0"},  # Main games only (IGDB renamed category -> game_type)
                 limit=limit,
             )
 

fjerkroa_bot/openai_responder.py

@@ -137,16 +137,20 @@ class OpenAIResponder(AIResponder, LeonardoAIDrawMixIn):
 
         # Initialize IGDB if enabled
         self.igdb = None
+        igdb_client_id = self.config.get("igdb-client-id")
+        igdb_client_secret = self.config.get("igdb-client-secret")
+        igdb_access_token = self.config.get("igdb-access-token")
         logging.info("IGDB Configuration Check:")
         logging.info(f"  enable-game-info: {self.config.get('enable-game-info', 'NOT SET')}")
-        logging.info(f"  igdb-client-id: {'SET' if self.config.get('igdb-client-id') else 'NOT SET'}")
+        logging.info(f"  igdb-client-id: {'SET' if igdb_client_id else 'NOT SET'}")
-        logging.info(f"  igdb-access-token: {'SET' if self.config.get('igdb-access-token') else 'NOT SET'}")
+        logging.info(f"  igdb-client-secret: {'SET' if igdb_client_secret else 'NOT SET'}")
+        logging.info(f"  igdb-access-token: {'SET' if igdb_access_token else 'NOT SET'}")
 
-        if self.config.get("enable-game-info", False) and self.config.get("igdb-client-id") and self.config.get("igdb-access-token"):
+        if self.config.get("enable-game-info", False) and igdb_client_id and (igdb_client_secret or igdb_access_token):
             try:
-                self.igdb = IGDBQuery(self.config["igdb-client-id"], self.config["igdb-access-token"])
+                self.igdb = IGDBQuery(igdb_client_id, igdb_access_token, client_secret=igdb_client_secret)
                 logging.info("✅ IGDB integration SUCCESSFULLY enabled for game information")
-                logging.info(f"   Client ID: {self.config['igdb-client-id'][:8]}...")
+                logging.info(f"   Client ID: {igdb_client_id[:8]}...")
                 logging.info(f"   Available functions: {len(self.igdb.get_openai_functions())}")
             except Exception as e:
                 logging.error(f"❌ Failed to initialize IGDB: {e}")

manual-verification.md

@@ -12,3 +12,4 @@ with date + result.
 | DEP-04 | 2026-07-13 | Smoke gate exercised on ggg: RUNNING + fresh login line. |
 | DEP-05 | 2026-07-13 | Live-verified: kroa deploy attempt ~15h Oslo refused without DEPLOY_FORCE=1. |
 | DEP-06 | 2026-07-13 | Rollback documented (older tag + db backup restore); live drill pending — next release. |
+| OPS-15 | 2026-07-13 | Backup cron installed on both hosts (daily 03:17 UTC → ~/backups/<bot>/, keep 14); first snapshots written + verified 0600 (kroa 10965 B, luma 25871 B). |

specs/SPEC-012-ops.md

@@ -0,0 +1,32 @@
+# SPEC-012 — Operations hardening
+
+Runtime + host operability (FDB-012). Backups and host wiring are
+`manual` coverage; the in-process alerting is `test`.
+
+### OPS-13 — Consistent DB backups (coverage: test)
+
+`deploy/backup_db.py` writes a gzipped snapshot of `bot.db` using the
+sqlite3 online-backup API — consistent even while the bot writes
+(WAL-safe) — with 0600 permissions. Restoring a snapshot yields a
+readable database with the same rows.
+
+### OPS-14 — Backups are rotated (coverage: test)
+
+The newest `backup-keep` (default 14) snapshots are kept; older ones
+are deleted. Timestamped names sort chronologically so rotation is a
+pure list operation.
+
+### OPS-15 — Backup cron on each host (coverage: manual)
+
+Each host runs `backup_db.py` daily via cron, writing to
+`~/backups/<bot>/` (outside `~/fjerkroa_bot`, so deploys and service
+restarts never touch it). Verified by presence of the cron line and a
+fresh snapshot.
+
+### OPS-16 — Repeated API errors alert staff (coverage: test)
+
+The responder counts consecutive OpenAI request failures; at
+`api-error-alert-threshold` (default 5) in a row it fires one staff
+alert (rate-limited like all staff alerts) so a silently-broken bot
+(cf. the gpt-5.6 tools/reasoning incident) surfaces within minutes
+instead of hours. A success resets the counter.

tests/test_igdb_integration.py

@@ -28,7 +28,7 @@ class TestIGDBIntegration(unittest.IsolatedAsyncioTestCase):
 
             responder = OpenAIResponder(self.config_with_igdb)
 
-            mock_igdb.assert_called_once_with("test_client", "test_token")
+            mock_igdb.assert_called_once_with("test_client", "test_token", client_secret=None)
             self.assertEqual(responder.igdb, mock_igdb_instance)
 
     def test_igdb_initialization_disabled(self):

tests/test_igdblib.py

@@ -160,7 +160,7 @@ class TestIGDBQuery(unittest.TestCase):
             "id",
             "name",
             "alternative_names",
-            "category",
+            "game_type",
             "release_dates",
             "franchise",
             "language_supports",
@@ -174,5 +174,85 @@ class TestIGDBQuery(unittest.TestCase):
         self.assertEqual(result, [{"id": 1, "name": "Super Mario Bros"}])
 
 
+class TestIGDBTokenRefresh(unittest.TestCase):
+    @staticmethod
+    def _oauth_response(token="fresh_token", expires_in=5_000_000):
+        response = Mock()
+        response.json.return_value = {"access_token": token, "expires_in": expires_in}
+        response.raise_for_status.return_value = None
+        return response
+
+    @staticmethod
+    def _api_response(payload, status_code=200):
+        response = Mock()
+        response.status_code = status_code
+        response.json.return_value = payload
+        response.raise_for_status.return_value = None
+        return response
+
+    @patch("fjerkroa_bot.igdblib.requests.post")
+    def test_fetches_token_when_only_secret_configured(self, mock_post):
+        """Without a static token, the first request fetches one via Twitch OAuth."""
+        mock_post.side_effect = [self._oauth_response(), self._api_response([{"id": 1}])]
+        igdb = IGDBQuery("cid", client_secret="secret")
+
+        result = igdb.send_igdb_request("games", "fields name; limit 1;")
+
+        self.assertEqual(result, [{"id": 1}])
+        oauth_call, api_call = mock_post.call_args_list
+        self.assertEqual(oauth_call.args[0], "https://id.twitch.tv/oauth2/token")
+        self.assertEqual(
+            oauth_call.kwargs["params"],
+            {"client_id": "cid", "client_secret": "secret", "grant_type": "client_credentials"},
+        )
+        self.assertEqual(api_call.kwargs["headers"]["Authorization"], "Bearer fresh_token")
+
+    @patch("fjerkroa_bot.igdblib.requests.post")
+    def test_refreshes_and_retries_on_401(self, mock_post):
+        """A 401 with a configured secret triggers one refresh and retry."""
+        mock_post.side_effect = [
+            self._api_response(None, status_code=401),
+            self._oauth_response(),
+            self._api_response([{"id": 2}]),
+        ]
+        igdb = IGDBQuery("cid", "expired_token", client_secret="secret")
+
+        result = igdb.send_igdb_request("games", "fields name; limit 1;")
+
+        self.assertEqual(result, [{"id": 2}])
+        self.assertEqual(igdb.igdb_api_key, "fresh_token")
+        self.assertEqual(mock_post.call_args_list[2].kwargs["headers"]["Authorization"], "Bearer fresh_token")
+
+    @patch("fjerkroa_bot.igdblib.time.time")
+    @patch("fjerkroa_bot.igdblib.requests.post")
+    def test_proactive_refresh_before_expiry(self, mock_post, mock_time):
+        """An expired self-fetched token is refreshed before the request."""
+        mock_time.return_value = 1_000_000.0
+        mock_post.side_effect = [self._oauth_response("token_a", expires_in=5_000_000), self._api_response([])]
+        igdb = IGDBQuery("cid", client_secret="secret")
+        igdb.send_igdb_request("games", "fields name;")
+
+        # jump past the token expiry -> next request refreshes first
+        mock_time.return_value = 1_000_000.0 + 5_000_000
+        mock_post.side_effect = [self._oauth_response("token_b"), self._api_response([])]
+        igdb.send_igdb_request("games", "fields name;")
+
+        self.assertEqual(igdb.igdb_api_key, "token_b")
+
+    @patch("fjerkroa_bot.igdblib.requests.post")
+    def test_no_refresh_without_secret(self, mock_post):
+        """Static-token setups keep the old behavior: no OAuth calls, error -> None."""
+        response = Mock()
+        response.status_code = 401
+        response.raise_for_status.side_effect = requests.RequestException("401 Client Error")
+        mock_post.return_value = response
+        igdb = IGDBQuery("cid", "expired_token")
+
+        result = igdb.send_igdb_request("games", "fields name; limit 1;")
+
+        self.assertIsNone(result)
+        mock_post.assert_called_once()
+
+
 if __name__ == "__main__":
     unittest.main()

tests/test_spec_ops_backup.py

@@ -0,0 +1,100 @@
+"""Unit coverage for SPEC-012 ops hardening (OPS-13/14/16)."""
+
+import gzip
+import sqlite3
+import stat
+import sys
+import tempfile
+import unittest
+from pathlib import Path
+from unittest.mock import AsyncMock, MagicMock
+
+from fjerkroa_bot.persistence import PersistentStore
+
+from .test_spec_ops import OpsBase
+
+sys.path.insert(0, str(Path(__file__).resolve().parent.parent / "deploy"))
+import backup_db  # noqa: E402
+
+
+class TestSnapshotConsistency(unittest.TestCase):
+    def test_snapshot_roundtrips(self):
+        """OPS-13: a gzipped snapshot restores to a readable DB with the same rows, 0600."""
+        with tempfile.TemporaryDirectory() as tmp:
+            db = Path(tmp) / "bot.db"
+            store = PersistentStore(db)
+            store.save_history("chat", [{"role": "user", "content": "hei"}])
+            store.add_user_fact("alice", "likes espresso", "self")
+
+            dest = Path(tmp) / "snap.db.gz"
+            backup_db.snapshot(db, dest)
+            self.assertEqual(stat.S_IMODE(dest.stat().st_mode), 0o600)
+
+            restored = Path(tmp) / "restored.db"
+            with gzip.open(dest, "rb") as gz, open(restored, "wb") as out:
+                out.write(gz.read())
+            conn = sqlite3.connect(restored)
+            try:
+                rows = conn.execute("SELECT content FROM history WHERE channel='chat'").fetchall()
+                facts = conn.execute("SELECT fact FROM user_facts").fetchall()
+            finally:
+                conn.close()
+            self.assertEqual(rows, [("hei",)])
+            self.assertEqual(facts, [("likes espresso",)])
+
+    def test_snapshot_during_writes(self):
+        """OPS-13: snapshot succeeds while another connection holds the DB open (WAL)."""
+        with tempfile.TemporaryDirectory() as tmp:
+            db = Path(tmp) / "bot.db"
+            store = PersistentStore(db)
+            store.save_history("chat", [{"role": "user", "content": "x"}])
+            live = sqlite3.connect(db)  # simulate the running bot's open handle
+            live.execute("PRAGMA journal_mode=WAL")
+            try:
+                dest = Path(tmp) / "snap.db.gz"
+                backup_db.snapshot(db, dest)  # must not raise
+                self.assertTrue(dest.exists())
+            finally:
+                live.close()
+
+
+class TestRotation(unittest.TestCase):
+    def test_victims_keeps_newest(self):
+        """OPS-14: only the oldest beyond `keep` are selected for deletion."""
+        names = [f"bot-2026070{d}-000000.db.gz" for d in range(1, 8)]  # 7 chronological
+        victims = backup_db.victims(list(reversed(names)), keep=3)
+        self.assertEqual(victims, names[:4])  # oldest 4 removed, newest 3 kept
+
+    def test_victims_under_keep_deletes_nothing(self):
+        """OPS-14: fewer than `keep` backups -> nothing deleted."""
+        self.assertEqual(backup_db.victims(["bot-20260701-000000.db.gz"], keep=14), [])
+
+    def test_rotate_on_disk(self):
+        """OPS-14: rotate removes the right files from a real dir."""
+        with tempfile.TemporaryDirectory() as tmp:
+            for d in range(1, 6):
+                (Path(tmp) / f"bot-2026070{d}-000000.db.gz").write_bytes(b"x")
+            removed = backup_db.rotate(Path(tmp), keep=2)
+            self.assertEqual(removed, 3)
+            self.assertEqual(len(list(Path(tmp).glob("bot-*.db.gz"))), 2)
+
+
+class TestApiErrorAlert(OpsBase):
+    async def test_threshold_alert_and_reset(self):
+        """OPS-16: N consecutive failures fire one staff alert; success resets."""
+        self.bot.config["api-error-alert-threshold"] = 3
+        self.bot.send_message_with_typing = AsyncMock(side_effect=RuntimeError("boom"))
+        origin = MagicMock()
+        from fjerkroa_bot.ai_responder import AIMessage
+
+        for _ in range(3):
+            await self.bot.respond(AIMessage("alice", "hei", "chat"), origin)
+        self.assertEqual(self.bot.staff_channel.send.await_count, 1)  # exactly one alert at threshold
+        self.assertEqual(self.bot._consecutive_api_errors, 3)
+
+        # a success resets the counter
+        from fjerkroa_bot.ai_responder import AIResponse
+
+        self.bot.send_message_with_typing = AsyncMock(return_value=AIResponse(None, False, "chat", None, None, False, False))
+        await self.bot.respond(AIMessage("alice", "hei", "chat"), origin)
+        self.assertEqual(self.bot._consecutive_api_errors, 0)