Отчёт ночного агента

У вас есть автономный агент, который крутится по расписанию: ночью разгребает inbox, чистит стейл-ветки, обновляет зависимости, разбирает support-тикеты. Утром хочется одну строку итога, а не лезть в логи Cloud-функции и не гадать, отработал ли он вообще.

Это не утренний дайджест из десятка разных сигналов — здесь ровно один источник: отчёт конкретного планового run-а. Собрали по ходу словарь-итог и в самом конце отправили один компактный push.

Оборачиваем run: собираем summary и шлём один push

Заводим маленький аккумулятор. Агент по ходу дёргает mark(), а обёртка в finally гарантированно шлёт дайджест — даже если run упал.

import os, time, traceback, requests

class RunSummary:
    def __init__(self, name):
        self.name    = name
        self.t0      = time.time()
        self.items   = 0        # сколько единиц обработали
        self.changes = []       # что реально изменили
        self.errors  = []       # накопленные ошибки
        self.cost    = 0.0      # $ за run

    def mark(self, items=0, change=None, cost=0.0):
        self.items += items
        self.cost  += cost
        if change:
            self.changes.append(change)

    def fail(self, err):
        self.errors.append(str(err))

    def digest(self):
        mins = (time.time() - self.t0) / 60
        lines = [f"⏱ {mins:.0f} мин · обработано: {self.items}"]
        lines += [f"• {c}" for c in self.changes[:5]]
        if len(self.changes) > 5:
            lines.append(f"…и ещё {len(self.changes) - 5}")
        lines.append(f"💵 ${self.cost:.2f}")
        if self.errors:
            lines.append(f"❌ ошибок: {len(self.errors)} — {self.errors[0][:80]}")
        return "\n".join(lines)


def notify(title, message, priority):
    requests.post(f"{os.environ['NOTIFLY_URL']}/message",
                  params={"token": os.environ["NOTIFLY_TOKEN"]},
                  json={"title": title, "message": message, "priority": priority},
                  timeout=5)

Сама обёртка вокруг run-а:

def run_scheduled_agent():
    s = RunSummary("nightly-inbox")
    try:
        for ticket in fetch_open_tickets():
            usd = agent.handle(ticket)                 # ваш агент
            s.mark(items=1, change=f"тикет #{ticket.id} → {ticket.label}", cost=usd)
    except Exception as e:
        s.fail(traceback.format_exc())
        raise
    finally:
        # успех — тихо (priority 3), есть ошибки — громко (priority 8)
        prio  = 8 if s.errors else 3
        emoji = "🌙❌" if s.errors else "🌙✅"
        notify(f"{emoji} Ночной агент: {s.name}", s.digest(), priority=prio)

if __name__ == "__main__":
    run_scheduled_agent()

Ключевое — finally: даже если run упадёт на середине, вы получите дайджест с тем, что успело сделаться, и с текстом ошибки. Один push на run, без спама.

Эскалация приоритета: успех тихо, провал громко

Плановый run — это фон, он не должен звенеть каждое утро. Логика простая:

• всё чисто → priority: 3 — push приходит, но беззвучно, прочитаете за кофе;
• были ошибки, но run дожил → priority: 8 — стоит глянуть сегодня;
• run упал целиком / вообще не стартовал → priority: 10 + heartbeat.

Отдельный нюанс — «агент вообще не запустился» (упал cron, умерла Cloud-функция). Push из кода тут не поможет: код не выполнился. Поэтому поверх дайджеста повесьте heartbeat с интервалом чуть больше периода расписания — молчание само по себе станет громким алёртом.

Как запускать по расписанию

Обёртка провайдер-агностична — заворачивается в любой планировщик:

# cron: каждую ночь в 03:00
0 3 * * *  /usr/bin/python3 /opt/agent/nightly.py >> /var/log/agent.log 2>&1

# systemd-timer: то же, но с автоперезапуском и journald
[Timer]
OnCalendar=*-*-* 03:00:00
Persistent=true

Или Yandex Cloud function с timer-триггером 0 3 * * ? * — тогда run_scheduled_agent() становится телом handler, а serverless сам поднимет и погасит окружение. Секреты (NOTIFLY_TOKEN, ключи модели) — в переменных функции.

Что класть в дайджест

Правило: строка должна отвечать на «нужно ли мне вмешаться» за пять секунд.

• сколько обработано — тикетов / файлов / записей;
• что реально изменилось — 3–5 самых важных пунктов, остальное свернуть в …и ещё N;
• стоимость run-а — $ за прогон (см. расход на run);
• ошибки — количество + первая строка первой, чтобы понять масштаб;
• длительность — аномально долгий run сам по себе сигнал.

Связанные рецепты

• Утренний дайджест — агрегат многих сигналов; здесь — отчёт одного run-а.
• Завершение долгой задачи — разовый «готово», а не плановый прогон.
• Агент проедает бюджет за run — откуда брать $ в строке дайджеста.