Automating effort certification reminders for PIs

Problem statement #

You need a Python job that runs at every effort-period close, works out which principal investigators still owe a certification, emails each of them a reminder on a fixed cadence without ever double-sending inside a window, escalates the ones who go overdue, and records every notification and the eventual sign-off with a reproducible hash — so a stalled certification is chased deterministically instead of slipping past a federal deadline.

This task sits under Effort Reporting & Certification, part of the broader Compliance Reporting & Budget Reconciliation practice. The reminder job is intentionally narrow: it reads the reconciled certification worklist produced by the parent cluster, decides who to nudge and when, and appends an auditable record of each contact. It does not reconcile effort or capture the attestation itself — the parent cluster owns the salary-cap-aware reconciliation and the immutable sign-off ledger; this page owns the cadence that drives a PI to sign.

Prerequisites #

Before deploying the reminder job, confirm the following environment and policy configuration:

• Python 3.10+ (the code uses modern type hints and datetime.now(timezone.utc)).
• Libraries: SQLAlchemy>=2.0 with psycopg[binary] for the certification and notification tables. The email transport is abstracted behind a callable so the same job runs against a real SMTP relay or a dry-run sink. Install with pip install "SQLAlchemy>=2.0" "psycopg[binary]".
• Environment variables (never hard-code credentials, per Security Boundary Configuration):
  • EFFORT_DB_URL — the SQLAlchemy connection string for the certification store.
  • SMTP_DSN — the relay endpoint the transport callable dials.
• Policy config: a version-controlled reminder cadence (offsets in days from period close), an escalation threshold, and the effort-period calendar, kept alongside your University Policy Mapping Frameworks. Reconciliation of committed against charged effort is done upstream by the parent cluster; this page assumes a worklist of reconciled certifications is already in the store.

Step-by-step implementation #

The flow below is enforced by the job: at period close it loads commitments and the payroll distribution, computes each PI’s variance, builds a per-PI worklist of uncertified statements, sends reminders on a deterministic cadence with an idempotency guard, escalates overdue items, and records every notification and the eventual sign-off with a hash. The (person_id, award_id, period) key and the send-window guard are what make the whole job safe to re-run.

Figure: the reminder cadence turns a reconciled worklist into deterministic, non-duplicating notifications.

Step 1 — Build the per-PI certification worklist at period close #

At period close, join the commitment of record against the payroll distribution to compute each PI’s variance, then select only the statements that are not yet certified. The worklist is the input to every later step, and it is built deterministically so the same close always yields the same list.

import hashlib
import json
import logging
from dataclasses import dataclass
from datetime import datetime, timedelta, timezone
from decimal import Decimal

logging.basicConfig(level=logging.INFO, format="%(asctime)s [REMINDER] %(message)s")
logger = logging.getLogger(__name__)


@dataclass(frozen=True)
class WorklistItem:
    person_id: str
    award_id: str
    period: str
    pi_email: str
    committed_pct: Decimal
    charged_pct: Decimal        # cap-adjusted fraction from the parent cluster
    variance: Decimal
    cert_status: str            # 'reconciled' | 'escalated' — never 'certified'
    period_close: datetime      # UTC instant the effort period closed


def build_worklist(rows: list[dict]) -> list[WorklistItem]:
    """Select uncertified statements and sort deterministically. Rows arrive
    already reconciled (variance and cap adjustment applied) from the parent
    cluster; this job never re-computes effort, it only decides who to nudge."""
    items = [
        WorklistItem(
            person_id=r["person_id"], award_id=r["award_id"], period=r["period"],
            pi_email=r["pi_email"], committed_pct=Decimal(r["committed_pct"]),
            charged_pct=Decimal(r["charged_pct"]), variance=Decimal(r["variance"]),
            cert_status=r["cert_status"], period_close=r["period_close"],
        )
        for r in rows
        if r["cert_status"] != "certified"   # certified statements need no reminder
    ]
    # Stable order so a re-run produces the same sequence of sends.
    return sorted(items, key=lambda i: (i.person_id, i.award_id, i.period))

Step 2 — Decide whether a reminder is due on a deterministic cadence #

The cadence is policy, not code: a list of day offsets from period close at which a reminder should fire, plus an escalation threshold. Computing the due offset from period_close and the current time — rather than from “when we last happened to run” — makes the decision independent of the scheduler’s jitter.

# Policy config (version-controlled): reminders at these days after close,
# escalation once past the final offset.
REMINDER_OFFSETS_DAYS = (0, 7, 14)
ESCALATION_OFFSET_DAYS = 21
SEND_WINDOW = timedelta(hours=20)   # no two sends for the same item inside this


def due_offset(item: WorklistItem, now: datetime) -> int | None:
    """Return the cadence offset (in days) whose reminder is due now, or None.
    Deterministic: derived from period_close and now, not from run time."""
    elapsed_days = (now - item.period_close).days
    # Fire the largest offset that has been reached but not yet passed the next.
    due = [d for d in REMINDER_OFFSETS_DAYS if d <= elapsed_days]
    return max(due) if due else None


def is_overdue(item: WorklistItem, now: datetime) -> bool:
    return (now - item.period_close).days >= ESCALATION_OFFSET_DAYS

Step 3 — Send reminders idempotently and record each notification #

Before sending, check the append-only notification log for an existing send to the same (person_id, award_id, period, offset) inside the current window. This guard is what prevents a double-run — or a scheduler that fires twice — from spamming a PI. Every send is written with a SHA-256 hash over its content, so the log is auditable.

from sqlalchemy import Column, String, Integer, DateTime, select
from sqlalchemy.orm import declarative_base, Session
from sqlalchemy.dialects.postgresql import insert as pg_insert

Base = declarative_base()


class NotificationLog(Base):
    __tablename__ = "effort_notifications"
    person_id = Column(String(64), primary_key=True)
    award_id = Column(String(32), primary_key=True)
    period = Column(String(16), primary_key=True)
    offset_days = Column(Integer, primary_key=True)   # cadence offset of this send
    kind = Column(String(16), nullable=False)          # 'reminder' | 'escalation'
    sent_at = Column(DateTime(timezone=True), nullable=False)
    notice_hash = Column(String(64), nullable=False)


def notice_digest(item: WorklistItem, kind: str, offset: int, sent_at: datetime) -> str:
    canonical = json.dumps(
        {"person_id": item.person_id, "award_id": item.award_id, "period": item.period,
         "kind": kind, "offset": offset, "variance": str(item.variance),
         "sent_at": sent_at.isoformat()},
        sort_keys=True, separators=(",", ":"),
    )
    return hashlib.sha256(canonical.encode("utf-8")).hexdigest()


def _already_sent(session: Session, item: WorklistItem, offset: int, now: datetime) -> bool:
    row = session.execute(
        select(NotificationLog).where(
            NotificationLog.person_id == item.person_id,
            NotificationLog.award_id == item.award_id,
            NotificationLog.period == item.period,
            NotificationLog.offset_days == offset,
        )
    ).scalar_one_or_none()
    return row is not None and (now - row.sent_at) < SEND_WINDOW


def process(session: Session, worklist: list[WorklistItem], send_email, now: datetime) -> dict:
    """Send due reminders and escalations, each at most once per window.
    send_email(to, subject, body) is injected so the job is dry-run testable."""
    stats = {"reminders": 0, "escalations": 0, "skipped": 0}

    for item in worklist:
        overdue = is_overdue(item, now)
        offset = ESCALATION_OFFSET_DAYS if overdue else due_offset(item, now)
        if offset is None:
            continue  # not yet due on the cadence
        if _already_sent(session, item, offset, now):
            stats["skipped"] += 1
            continue  # idempotent guard: already contacted inside this window

        kind = "escalation" if overdue else "reminder"
        digest = notice_digest(item, kind, offset, now)
        recipient = item.pi_email if not overdue else "effort-compliance@dept.internal"
        send_email(
            recipient,
            f"Effort certification due: {item.award_id} / {item.period}",
            f"Committed {item.committed_pct}% vs charged {item.charged_pct}% "
            f"(variance {item.variance}). Please certify.",
        )
        # Append-only: the primary key includes offset_days, so a genuinely new
        # cadence step inserts while a replay of the same step conflicts and is
        # left untouched.
        stmt = pg_insert(NotificationLog).values(
            person_id=item.person_id, award_id=item.award_id, period=item.period,
            offset_days=offset, kind=kind, sent_at=now, notice_hash=digest,
        ).on_conflict_do_nothing(
            index_elements=["person_id", "award_id", "period", "offset_days"],
        )
        session.execute(stmt)
        stats["escalations" if overdue else "reminders"] += 1

    session.commit()
    return stats

The on_conflict_do_nothing keyed on (person_id, award_id, period, offset_days) is the second idempotency layer: even if the in-memory window check is bypassed, the database refuses to log the same cadence step twice. The eventual attestation itself is captured by the parent cluster’s hashed sign-off ledger, and the signed statements ultimately flow into the Immutable Audit Report Generation deliverables.

Schema and field reference #

The reminder job reads the worklist fields and writes the notification log. Widen the cadence in version-controlled policy config, not in code.

Verification #

Confirm a run behaved correctly before trusting its output:

1. Dry-run idempotency: run process twice with the same worklist and now. The second pass must report every item under skipped and insert zero new NotificationLog rows.
2. Reproduce a notice hash: re-run notice_digest for a logged send with the stored sent_at and confirm it equals the persisted notice_hash — proof the log row matches the notification that was sent.
3. Cadence boundary check: for an item with period_close set so elapsed_days is 6 vs 7, confirm due_offset returns 0 then 7 respectively, so a reminder fires exactly at the policy offset and not a day early.
4. Escalation routing: set now past ESCALATION_OFFSET_DAYS and confirm the recipient becomes the department address and kind is escalation.

Troubleshooting #

Three failure modes specific to reminder cadences:

• Cost-shared effort makes a PI appear delinquent. Committed cost-shared effort is not charged to the federal fund, so it never appears in payroll distribution; if the worklist is built from payroll alone, the PI’s committed award shows 0% charged and gets nagged for a certification that does not apply. Build the worklist from the reconciled parent-cluster output, which carries cost-share on a separate record excluded from the payroll variance, rather than from a raw payroll join.
• Over-the-cap effort inflates the apparent variance. For a PI above the NIH salary cap, the raw payroll fraction understates their effort because salary above the cap is on non-federal funds. A reminder that quotes the raw variance alarms the PI with a number that is not a real shortfall. Always read the cap-adjusted charged_pct and variance from the parent cluster’s reconciliation; never recompute variance from raw payroll in this job.
• Effort-period boundary and timezone drift double-fire the cadence. If period_close is stored in local time but now is UTC, elapsed_days can shift by a day near the boundary and fire an offset early or late — and a scheduler that runs just before and after midnight can log two sends. Store period_close as a timezone-aware UTC instant, derive elapsed_days in UTC, and rely on the SEND_WINDOW guard plus the composite-key on_conflict_do_nothing to absorb any residual double-run.

Frequently asked questions #

Related #

• Up to the parent topic: Effort Reporting & Certification
• Automating Calibration Reminder Emails for Lab Equipment — the same deterministic reminder pattern for calibration due dates
• Escalating Overdue Calibration to Lab Managers — a sibling escalation cadence
• Reconciling F&A Indirect-Cost Recovery Rates in Python — a related reporting reconciliation