Automated Escalation Alerts Before a FOIA Deadline Breach

Within Deadline Tracking & Escalation, computing the deadline is only half the obligation — the other half is making sure a human acts before the countdown reaches zero. An escalation system watches every open request’s days-remaining and fires progressively more senior alerts as the margin shrinks: a quiet nudge to the assigned owner while there is still a week, a louder warning to the supervisor at three days, a page to the agency records officer at one, and a logged breach event at zero. This page builds that threshold ladder in Python, makes the notifications idempotent so a fifteen-minute monitor does not become a pager storm, and records an alert history that proves the right person was warned on the right day.

The Escalation Ladder

The design principle is that escalation should be anticipatory, not reactive. A system that only fires at T-0 tells the records officer about a breach that has already happened; a system that escalates on a schedule gives each level of the organization time to intervene while intervention still matters. The tiers are keyed to days-remaining rather than elapsed time, because tolling and per-jurisdiction windows mean two requests received the same day can have very different margins. The exact thresholds are policy — an agency working a 5-business-day acknowledgment window will compress them — but the ladder always climbs: the closer to breach, the more senior and more insistent the notification.

Escalation ladder keyed to days-remaining A vertical countdown axis descends through four thresholds. At T-7 business days remaining, the assigned records owner is notified by email and a queue flag. At T-3 the unit supervisor is notified via the escalation dashboard. At T-1 the agency records officer is paged. At T-0 the deadline breach is logged with breach evidence. Each tier fires at most once per request and every alert is appended to the audit log. days remaining T-7 T-3 T-1 T-0 Notify assigned records owner email · queue flag Notify unit supervisor email · escalation dashboard Notify agency records officer page · priority escalation Deadline breach log breach · assemble evidence Alert history · one send per tier · append-only — AU-2 / AU-12

Idempotent Notification Scheduling

The property that makes an escalation monitor usable is idempotency: a tier fires exactly once per request, no matter how often the monitor runs. Without it, a sweep every fifteen minutes would deliver the T-1 page ninety-six times a day, staff would mute the channel, and the system would fail precisely when it mattered. The mechanism is a durable record of which tiers have already been sent for each request. On every sweep the monitor computes the current tier from days-remaining, and only sends — and only logs a delivery — if that tier is not already present in the request’s sent-set. This turns a continuously-running monitor into an edge-triggered one that acts on the transition into a tier, not on every observation of it.

Idempotency also makes the monitor safe to recover. If the host is down for a day and a request crosses two thresholds during the outage, the monitor on restart computes the current, more-severe tier and sends any un-sent tiers at or below the current margin — the request escalates to where it should be, immediately, rather than replaying stale intermediate alerts. Running the send step through the durable brokering in Async Queue Management further decouples delivery from the monitor’s own loop, so a slow email gateway never stalls the evaluation of the next request.

Implementation

The monitor below reads a deadline clock’s days-remaining, maps it to a tier, and sends idempotently, appending every delivery to a structured audit log. The days_remaining value and the escalation thresholds mirror the ones the parent tracker computes; here the focus is the scheduling and delivery contract.

python 
"""
escalation_monitor.py
Threshold-based, idempotent escalation alerts for FOIA deadlines.
Alerts fire before breach so staff can act while intervention still matters.
5 U.S.C. § 552(a)(6)(A)(i) sets the 20-business-day window the countdown tracks.
"""

import json
import logging
import sys
from dataclasses import dataclass, field
from datetime import datetime, timezone
from typing import Callable, Optional, Set

# ---------------------------------------------------------------------------
# Structured audit logging: every alert delivery is recorded so the agency can
# prove timely warning (AU-2 / AU-12).
# ---------------------------------------------------------------------------
class JSONFormatter(logging.Formatter):
    def format(self, record: logging.LogRecord) -> str:
        entry = {
            "timestamp": self.formatTime(record, self.datefmt),
            "level": record.levelname,
            "message": record.getMessage(),
            "request_id": getattr(record, "request_id", None),
            "tier": getattr(record, "tier", None),
            "recipient": getattr(record, "recipient", None),
            "days_remaining": getattr(record, "days_remaining", None),
        }
        return json.dumps(entry)


audit = logging.getLogger("foia_escalation_audit")
audit.setLevel(logging.INFO)
_handler = logging.StreamHandler(sys.stdout)
_handler.setFormatter(JSONFormatter())
audit.addHandler(_handler)

# Ordered most-severe first. Each tier: threshold (days remaining), recipient role.
TIERS = (
    ("T0_BREACH", 0, "records_officer"),
    ("T1_RECORDS_OFFICER", 1, "records_officer"),
    ("T3_SUPERVISOR", 3, "unit_supervisor"),
    ("T7_OWNER", 7, "assigned_owner"),
)


def current_tier(days_remaining: Optional[int]):
    """Return (tier_name, recipient) for the current margin, or None if on track/paused."""
    if days_remaining is None:
        return None  # clock is tolled; suppress escalation while paused
    for name, threshold, recipient in TIERS:
        if days_remaining <= threshold:
            return name, recipient
    return None


@dataclass
class TrackedRequest:
    request_id: str
    days_remaining: Optional[int]
    sent_tiers: Set[str] = field(default_factory=set)  # durable in production


def evaluate_and_alert(req: TrackedRequest, notify: Callable[[str, str, str], None]) -> Optional[str]:
    """Send the current tier at most once; return the tier sent or None."""
    tier = current_tier(req.days_remaining)
    if tier is None:
        return None
    name, recipient = tier
    # Idempotency guard: a tier already sent for this request is never re-sent,
    # so a monitor sweeping every few minutes does not become an alert storm.
    if name in req.sent_tiers:
        return None
    try:
        notify(req.request_id, name, recipient)          # side-effecting delivery
        req.sent_tiers.add(name)                          # mark only after success
        audit.info(
            f"escalation {name} sent for {req.request_id}",
            extra={"request_id": req.request_id, "tier": name,
                   "recipient": recipient, "days_remaining": req.days_remaining},
        )
        return name
    except Exception as exc:
        # A failed send is NOT marked sent, so the next sweep retries it rather
        # than silently skipping a statutory warning.
        audit.error(
            f"escalation delivery failed for {req.request_id}: {exc}",
            extra={"request_id": req.request_id, "tier": name, "recipient": recipient},
        )
        raise


def sweep(requests, notify: Callable[[str, str, str], None]) -> dict:
    """Evaluate every open request; idempotent and safe to re-run after downtime."""
    fired = {}
    for req in requests:
        try:
            sent = evaluate_and_alert(req, notify)
            if sent:
                fired[req.request_id] = sent
        except Exception:
            # One request's delivery failure must not halt the whole sweep.
            continue
    audit.info(
        f"sweep complete at {datetime.now(timezone.utc).isoformat()}",
        extra={"tier": "SWEEP", "days_remaining": None},
    )
    return fired


if __name__ == "__main__":
    def deliver(request_id: str, tier: str, recipient: str) -> None:
        # Production hands this to a durable notification broker.
        print(f"ALERT {tier} -> {recipient} for {request_id}")

    open_requests = [
        TrackedRequest("REQ-2026-0412", days_remaining=3),
        TrackedRequest("REQ-2026-0733", days_remaining=8),   # on track, no alert
        TrackedRequest("REQ-2026-0655", days_remaining=0),   # breach
    ]
    first = sweep(open_requests, deliver)
    second = sweep(open_requests, deliver)   # re-run: idempotent, sends nothing new
    print(json.dumps({"first_sweep": first, "second_sweep": second}, indent=2))

The two consecutive sweeps in the example demonstrate the core guarantee: the first fires the due tiers and the second, running against the same requests, sends nothing because every fired tier is already in its request’s sent-set. A delivery that raises is never marked sent, so it is retried on the next pass rather than silently dropped.

Proving Timely Warning

The alert history is the evidentiary payoff of the whole system. When a requester alleges the agency ignored their request, the append-only record that a T-7 email reached the assigned owner, a T-3 warning reached the supervisor, and a T-1 page reached the records officer — each on a specific, logged date — is what converts an apparent lapse into a documented, escalated near-miss. Write that history to the shared store described in Audit Logging Architecture rather than a local file, so it inherits the immutability and retention guarantees the rest of the compliance record relies on.

Two verification checks keep the history trustworthy. First, completeness: every request that reached a tier has a matching delivery line, and a daily reconciliation flags any open request whose days-remaining has crossed a threshold with no corresponding alert. Second, suppression correctness: a tolled request emits no escalation while paused, and resumes escalation from the correct tier once the clock restarts. Because escalation is derived from the days-remaining figure the deadline math produces, an error in the underlying business-day computation surfaces here as a mistimed alert — which is why the arithmetic in Deadline Tracking & Escalation and its per-jurisdiction windows must be correct before the ladder can be trusted.

Compliance Verification Checklist

FAQ

How do the alerts avoid firing repeatedly when the monitor runs every few minutes?

Each tier is guarded by a durable sent-set per request. On every sweep the monitor computes the current tier from days-remaining and delivers only if that tier is not already in the set, marking it sent only after a successful delivery. This makes a continuously-running monitor edge-triggered: it acts on the transition into a tier, not on every observation, so the T-1 page is delivered once even if the monitor sweeps ninety-six times that day.

What happens to escalation while a request is tolled?

It is suppressed. A tolled clock reports its days-remaining as undefined, and the tier function returns nothing for an undefined margin, so no alert fires while the clock is paused. When the tolling resumes and the deadline shifts forward, the days-remaining figure is defined again and escalation continues from whatever tier the new margin implies — never replaying alerts that were suppressed during the pause.

Why record every alert instead of only logging the final breach?

Because the value of the system is proving the agency acted before the breach. A record showing that the owner, supervisor, and records officer were each warned on schedule converts an apparent missed deadline into a documented, escalated near-miss — the difference between a defensible process failure and negligence. Logging only the breach discards exactly the evidence that demonstrates the escalation worked as designed.