From e289e3a7f9c83ae32f1649e958edbd8c21355bd9 Mon Sep 17 00:00:00 2001 From: Your Name Date: Thu, 30 Jul 2026 09:05:36 +0500 Subject: [PATCH] Alert on AI worker auto-disable --- PROJECT.md | 4 +- STATE.md | 2 +- src/vk_parser_app/admin.py | 8 ++- src/vk_parser_app/templates/workers.html | 7 ++- src/vk_parser_app/workers/ai_alerts.py | 71 +++++++++++++++++++++++ src/vk_parser_app/workers/ai_qualifier.py | 12 +++- src/vk_parser_app/workers/ai_writer.py | 12 +++- 7 files changed, 106 insertions(+), 10 deletions(-) create mode 100644 src/vk_parser_app/workers/ai_alerts.py diff --git a/PROJECT.md b/PROJECT.md index 3ecaee3..a247912 100644 --- a/PROJECT.md +++ b/PROJECT.md @@ -796,7 +796,9 @@ AI получает JSON-массив: посты нужно возвращать в `pending` только явным ручным действием. После любого failed AI batch worker дополнительно выключает свой флаг `ai_writer_enabled=false` / `ai_qualifier_enabled=false`; включать обратно -только после просмотра ошибки. +только после просмотра ошибки. Ошибка показывается на `/workers` из heartbeat +meta и отправляется Telegram-уведомлением получателям `daily_report_recipient_ids` +через `daily_report_bot_token`, затем `tg_poster_bot_token`, затем `TG_BOT_TOKEN`. ### 10.2. Вход в модель diff --git a/STATE.md b/STATE.md index 9a89212..7c4cbd1 100644 --- a/STATE.md +++ b/STATE.md @@ -102,7 +102,7 @@ As of 2026-07-29: - Current live writer model setting checked on 2026-07-29: `ai_writer_model = anthropic/claude-sonnet-4-6`. - Sonnet 4.6 can return `rewrites` as a JSON string instead of a list. `validate_rewrites()` in `src/vk_parser_app/workers/ai_writer.py` now parses stringified `rewrites` before validation. -- 2026-07-30 incident: post `2348` was retried 983 times by AI writer because automatic claim included `rewrite_status='failed'`; estimated internal cost was `$16.465977`. Live setting `ai_writer_enabled` was switched to `false` to stop spend. Automatic AI qualifier/writer claims must not include `failed`; failed posts require an explicit manual reset to `pending`. AI writer/qualifier now also set their own `*_enabled=false` after any failed AI batch, so one broken prompt/model/post cannot burn money indefinitely. +- 2026-07-30 incident: post `2348` was retried 983 times by AI writer because automatic claim included `rewrite_status='failed'`; estimated internal cost was `$16.465977`. Live setting `ai_writer_enabled` was switched to `false` to stop spend. Automatic AI qualifier/writer claims must not include `failed`; failed posts require an explicit manual reset to `pending`. AI writer/qualifier now also set their own `*_enabled=false` after any failed AI batch, show the error on `/workers`, and send a Telegram alert via daily-report/TG-poster bot recipients, so one broken prompt/model/post cannot burn money indefinitely or fail silently. ## Prompt Architecture diff --git a/src/vk_parser_app/admin.py b/src/vk_parser_app/admin.py index 9de54aa..734cc1c 100644 --- a/src/vk_parser_app/admin.py +++ b/src/vk_parser_app/admin.py @@ -3410,7 +3410,13 @@ async def workers(request: Request): base_context( request, user, - workers=[dict(r) for r in rows], + workers=[ + { + **dict(r), + "meta_json": json.loads(r["meta_json"]) if isinstance(r["meta_json"], str) else (r["meta_json"] or {}), + } + for r in rows + ], settings=settings, provider_options=PROVIDER_OPTIONS, ai_provider=ai_provider, diff --git a/src/vk_parser_app/templates/workers.html b/src/vk_parser_app/templates/workers.html index 316bf68..63de50e 100644 --- a/src/vk_parser_app/templates/workers.html +++ b/src/vk_parser_app/templates/workers.html @@ -38,7 +38,12 @@ {% endif %} {{ w.heartbeat_at or "—" }} - {{ w.status or "—" }} + +
{{ w.status or "—" }}
+ {% if w.meta_json and w.meta_json.error %} +
{{ w.meta_json.error }}
+ {% endif %} + {{ w.current_job_id or "—" }}
diff --git a/src/vk_parser_app/workers/ai_alerts.py b/src/vk_parser_app/workers/ai_alerts.py new file mode 100644 index 0000000..2fdbf4f --- /dev/null +++ b/src/vk_parser_app/workers/ai_alerts.py @@ -0,0 +1,71 @@ +from __future__ import annotations + +import asyncio +import json +import re +from typing import Any + +from aiogram import Bot +from aiogram.exceptions import TelegramRetryAfter +from loguru import logger + +from ..config import settings +from ..db import fetch_setting + + +def parse_recipients(value: Any) -> list[int]: + if isinstance(value, list): + raw_items = value + elif isinstance(value, str): + raw = value.strip() + if not raw: + return [] + try: + parsed = json.loads(raw) + raw_items = parsed if isinstance(parsed, list) else [parsed] + except json.JSONDecodeError: + raw_items = re.split(r"[\s,;]+", raw) + else: + raw_items = [value] + + recipients: list[int] = [] + for item in raw_items: + try: + recipient_id = int(str(item).strip()) + except Exception: + continue + if recipient_id not in recipients: + recipients.append(recipient_id) + return recipients + + +async def send_ai_worker_disabled_alert(worker_name: str, model: str, post_ids: list[int], error: str) -> None: + token = ( + str(await fetch_setting("daily_report_bot_token", "") or "").strip() + or str(await fetch_setting("tg_poster_bot_token", "") or "").strip() + or settings.tg_bot_token + ) + recipients = parse_recipients(await fetch_setting("daily_report_recipient_ids", [442509142])) + if not token or not recipients: + logger.warning("AI worker alert skipped: token or recipients are empty") + return + + post_part = ", ".join(str(post_id) for post_id in post_ids) if post_ids else "-" + text = ( + "AI worker auto-disabled\n" + f"worker: {worker_name}\n" + f"model: {model or '-'}\n" + f"posts: {post_part}\n" + f"error: {error[:1000]}" + ) + bot = Bot(token=token) + try: + for recipient_id in recipients: + while True: + try: + await bot.send_message(recipient_id, text, disable_web_page_preview=True) + break + except TelegramRetryAfter as exc: + await asyncio.sleep(float(exc.retry_after) + 1) + finally: + await bot.session.close() diff --git a/src/vk_parser_app/workers/ai_qualifier.py b/src/vk_parser_app/workers/ai_qualifier.py index 5423ef5..4eec999 100644 --- a/src/vk_parser_app/workers/ai_qualifier.py +++ b/src/vk_parser_app/workers/ai_qualifier.py @@ -15,6 +15,7 @@ from ..constants import WORKER_AI_QUALIFIER from ..db import fetch_bool_setting, fetch_float_setting, fetch_int_setting, fetch_setting, get_pool from ..heartbeat import HeartbeatReporter from ..jobs import is_worker_enabled +from .ai_alerts import send_ai_worker_disabled_alert def now_utc() -> datetime: @@ -291,7 +292,7 @@ class AIQualifierWorker: error[:1000], ) - async def disable_after_error(self, error: str) -> None: + async def disable_after_error(self, model: str, post_ids: list[int], error: str) -> None: await self.pool.execute( """ UPDATE app_settings @@ -300,7 +301,12 @@ class AIQualifierWorker: WHERE key='ai_qualifier_enabled' """ ) - await self.heartbeat.beat(self.pool, status="disabled_after_error", meta={"error": error[:300]}, force=True) + meta = {"error": error[:300], "post_ids": post_ids, "model": model} + await self.heartbeat.beat(self.pool, status="disabled_after_error", meta=meta, force=True) + try: + await send_ai_worker_disabled_alert(WORKER_AI_QUALIFIER, model, post_ids, error) + except Exception as alert_exc: + logger.warning("AI qualifier alert failed: {}", alert_exc) async def apply_results(self, batch_id: int, results: list[dict], model: str, prompt: str, min_score: int) -> None: for item in results: @@ -423,7 +429,7 @@ class AIQualifierWorker: except Exception as exc: await self.mark_posts_failed(post_ids, str(exc)) await self.finish_batch(batch_id, "failed", error=str(exc)) - await self.disable_after_error(str(exc)) + await self.disable_after_error(normalize_model(provider, model), post_ids, str(exc)) logger.exception("AI qualifier batch failed: id={} error={}", batch_id, exc) return True diff --git a/src/vk_parser_app/workers/ai_writer.py b/src/vk_parser_app/workers/ai_writer.py index 2fa2c8d..843cc72 100644 --- a/src/vk_parser_app/workers/ai_writer.py +++ b/src/vk_parser_app/workers/ai_writer.py @@ -17,6 +17,7 @@ from ..db import fetch_bool_setting, fetch_float_setting, fetch_int_setting, fet from ..heartbeat import HeartbeatReporter from ..jobs import is_worker_enabled from ..text_utils import build_publication_text, normalize_hash_tag, parse_categories +from .ai_alerts import send_ai_worker_disabled_alert NON_TARGET_CATEGORY_ID = 18 @@ -372,7 +373,7 @@ class AIWriterWorker: error[:1000], ) - async def disable_after_error(self, error: str) -> None: + async def disable_after_error(self, model: str, post_ids: list[int], error: str) -> None: await self.pool.execute( """ UPDATE app_settings @@ -381,7 +382,12 @@ class AIWriterWorker: WHERE key='ai_writer_enabled' """ ) - await self.heartbeat.beat(self.pool, status="disabled_after_error", meta={"error": error[:300]}, force=True) + meta = {"error": error[:300], "post_ids": post_ids, "model": model} + await self.heartbeat.beat(self.pool, status="disabled_after_error", meta=meta, force=True) + try: + await send_ai_worker_disabled_alert(WORKER_AI_WRITER, model, post_ids, error) + except Exception as alert_exc: + logger.warning("AI writer alert failed: {}", alert_exc) async def apply_rewrites(self, batch_id: int, rewrites: list[dict], model: str, prompt: str) -> None: for item in rewrites: @@ -529,7 +535,7 @@ class AIWriterWorker: except Exception as exc: await self.mark_posts_failed(post_ids, str(exc)) await self.finish_batch(batch_id, "failed", response=response, error=str(exc), usage=usage) - await self.disable_after_error(str(exc)) + await self.disable_after_error(normalize_model(provider, model), post_ids, str(exc)) logger.exception("AI writer batch failed: id={} error={}", batch_id, exc) return True