Skip to content
Notifly Notifly

Scheduled agent run report

You have an autonomous agent that runs on a schedule: overnight it clears the inbox, prunes stale branches, bumps dependencies, triages support tickets. In the morning you want one line of summary, not a dig through Cloud-function logs and a guess about whether it even ran.

This is not the morning digest of a dozen different signals — here there’s exactly one source: the report of a specific scheduled run. Collect a summary dict as it goes and, at the very end, send one compact push.

Wrapping the run: collect a summary and send one push

Section titled “Wrapping the run: collect a summary and send one push”

Set up a tiny accumulator. The agent calls mark() as it goes, and the wrapper’s finally guarantees the digest is sent — even if the run crashes.

import os, time, traceback, requests


class RunSummary:
    def __init__(self, name):
        self.name    = name
        self.t0      = time.time()
        self.items   = 0        # how many units processed
        self.changes = []       # what was actually changed
        self.errors  = []       # accumulated errors
        self.cost    = 0.0      # $ for the 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)

The wrapper around the run itself:

def run_scheduled_agent():
    s = RunSummary("nightly-inbox")
    try:
        for ticket in fetch_open_tickets():
            usd = agent.handle(ticket)                 # your agent
            s.mark(items=1, change=f"тикет #{ticket.id} → {ticket.label}", cost=usd)
    except Exception as e:
        s.fail(traceback.format_exc())
        raise
    finally:
        # success — quiet (priority 3), errors — loud (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()

The key is finally: even if the run dies halfway, you get a digest with what managed to get done and the error text. One push per run, no spam.

Priority escalation: quiet on success, loud on failure

Section titled “Priority escalation: quiet on success, loud on failure”

A scheduled run is background noise; it shouldn’t ring every morning. The logic is simple:

  • all clear → priority: 3 — the push arrives silently, you read it over coffee;
  • there were errors but the run survived → priority: 8 — worth a look today;
  • the run crashed entirely / never started → priority: 10 + a heartbeat.

A separate wrinkle is “the agent never started at all” (cron broke, the Cloud function died). A push from inside the code won’t help here: the code never ran. So on top of the digest, hang a heartbeat with an interval slightly longer than the schedule period — silence itself becomes a loud alert.

How to run it on a schedule

Section titled “How to run it on a schedule”

The wrapper is provider-agnostic — it drops into any scheduler:

# cron: every night at 03:00
0 3 * * *  /usr/bin/python3 /opt/agent/nightly.py >> /var/log/agent.log 2>&1
# systemd timer: same, but with auto-restart and journald
[Timer]
OnCalendar=*-*-* 03:00:00
Persistent=true

Or a Yandex Cloud function with a timer trigger 0 3 * * ? * — then run_scheduled_agent() becomes the body of handler, and serverless spins the environment up and down for you. Secrets (NOTIFLY_TOKEN, model keys) go into the function’s environment variables.

What to put in the digest

Section titled “What to put in the digest”

The rule: the line must answer “do I need to step in” within five seconds.

  • how much was processed — tickets / files / records;
  • what actually changed — the 3–5 most important items, collapse the rest into …и ещё N;
  • the run’s cost — $ for the run (see per-run spend);
  • errors — the count + the first line of the first one, to gauge the scale;
  • duration — an abnormally long run is a signal in itself.
Section titled “Related recipes”