Approval stalled and is about to expire

Human-in-the-loop covers the moment “the agent stopped, it needs your approval right now”. But the real pain comes later: the push arrived, you were driving, then you forgot — and the approval is still hanging. An hour later the pipeline never moved, the task is about to expire (auto-deny or a blocked deploy), and you don’t even remember it was waiting on you.

Here the focus is not “approval needed” but the deadline and escalation: the approval has been pending too long and will soon expire. One push when it’s enqueued isn’t enough. You need a sweeper that watches overdue approvals and reminds you ever louder as the deadline approaches.

A pending-approvals store with deadlines

Every approval is a row with its own deadline and a ladder of already-sent reminders. The agent files a request, then goes about its business.

import os, time, json, sqlite3, requests

DB = os.environ.get("APPROVALS_DB", "/state/approvals.db")

def _db():
    db = sqlite3.connect(DB)
    db.execute("""CREATE TABLE IF NOT EXISTS approvals(
        id TEXT PRIMARY KEY, what TEXT, created REAL,
        deadline REAL, status TEXT, rung INTEGER)""")
    return db

def request_approval(req_id, what, ttl_sec=3600):
    """The agent parks a task: approval needed by created+ttl."""
    now = time.time()
    db = _db()
    db.execute("INSERT OR REPLACE INTO approvals VALUES (?,?,?,?,?,?)",
               (req_id, what, now, now + ttl_sec, "pending", 0))
    db.commit()
    notify("🤖 Нужен approve",
           f"{what}\nОтветить до {time.strftime('%H:%M', time.localtime(now + ttl_sec))}.",
           priority=6)

def resolve(req_id, verdict):
    """Called from your callback (approve/deny), see human-in-the-loop."""
    db = _db()
    db.execute("UPDATE approvals SET status=? WHERE id=?", (verdict, req_id))
    db.commit()


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)

How you capture the verdict itself (a button / callback URL / WebScript) is covered end to end in human-in-the-loop. Here all that matters is that resolve() moves the row from pending to approve/deny.

The sweeper: escalate overdue approvals with rising priority

A separate process (cron / a timer-triggered Yandex Cloud function) runs every few minutes over the pending rows and decides who to poke. The closer the deadline, the louder.

def sweep():
    now = time.time()
    db = _db()
    rows = db.execute(
        "SELECT id, what, deadline, rung FROM approvals WHERE status='pending'"
    ).fetchall()

    for req_id, what, deadline, rung in rows:
        left = deadline - now

        # reminder ladder: less time left -> higher priority and rung
        if left <= 0:
            ladder = (3, 10)      # deadline blown — loud finale
        elif left <= 5 * 60:
            ladder = (2, 9)       # < 5 min — almost expired
        elif left <= 15 * 60:
            ladder = (1, 7)       # < 15 min — time to answer
        else:
            continue              # too early to remind

        want_rung, prio = ladder
        if rung >= want_rung:
            continue              # this rung was already sent

        mins = max(int(left // 60), 0)
        notify("⏳ Approve скоро протухнет",
               f"{what}\nОсталось ~{mins} мин. Ответьте, иначе авто-deny.",
               priority=prio)
        db.execute("UPDATE approvals SET rung=? WHERE id=?", (want_rung, req_id))
    db.commit()

if __name__ == "__main__":
    sweep()

The rung field is the same idea as the _warned flag in the budget guard: each rung of the ladder is sent exactly once, so the sweeper doesn’t spam the same push every five minutes.

Auto-deny on expiry (fail-safe)

Silence is a “no”. If the approval gates an irreversible action (git push --force, a migration, a rollout), it’s safer to auto-deny on expiry and unblock the pipeline with a rejection than to hang forever:

def expire_overdue():
    now = time.time()
    db = _db()
    dead = db.execute(
        "SELECT id, what FROM approvals WHERE status='pending' AND deadline < ?",
        (now,)).fetchall()
    for req_id, what in dead:
        db.execute("UPDATE approvals SET status='auto-deny' WHERE id=?", (req_id,))
        notify("🛑 Approve авто-отклонён",
               f"{what}\nНикто не ответил вовремя — задача отменена (safe default).",
               priority=8)
    db.commit()

Wire expire_overdue() into the same sweeper right after sweep(). The agent, polling its store, will see the auto-deny status and cleanly roll the task back instead of waiting forever.

A reminder ladder instead of a wall

If you’d rather be gentle — don’t tie escalation strictly to the deadline; remind on a schedule instead: the first ping quiet/normal, then louder.

Since the request	Priority	Meaning
immediately	6	”there’s a task to approve”
+20 min	7	”still hanging”
+45 min	9	”deadline close, answer”
deadline	10	”expired, auto-deny”

Such a ladder respects your attention: you don’t drown in loud pushes, but the task is physically impossible to sleep through.

What to put in the push

• exactly what’s awaiting approval — command, file, branch, migration name;
• how much time is left and what happens on expiry (auto-deny?);
• req_id — so you can find the request in logs and in the store;
• an approve/deny link — to answer straight from the push, no laptop needed.

Related recipes

• Human-in-the-loop required — how to catch an approval right now; this recipe is about its deadline.
• Stalled agent / loop — the agent is silent not because it waits on you, but because it’s stuck.
• Heartbeat (dead-man-switch) — to run the sweeper on a timer.