Tracking State Primacy Notification Deadlines

The clock that governs a public notice does not start when an operator notices a problem; it starts the moment a violation becomes known, and it runs against a deadline that depends on both the tier of the violation and the primacy agency that oversees your system. This page, part of the Public Notification Workflows for SDWA Violations section, is about the bookkeeping layer that keeps those clocks honest: how to encode each agency’s deadline rules as versioned configuration, compute the exact instant a notice is due, measure the time still remaining, and escalate the moment a deadline slips past. It deliberately leaves the question of what fires a notice to automating Tier 1 public notification triggers and the question of how a message is delivered to dispatching multi-channel notifications. Here the single concern is the deadline itself, and getting it wrong is its own reportable violation.

Federal public-notification rules under 40 CFR Part 141 Subpart Q set the floor: Tier 1 acute-risk notices within 24 hours, Tier 2 notices within 30 days, and Tier 3 notices bundled into the annual report. But states with primacy routinely adopt tighter windows, shorter repeat-notice intervals, and their own definitions of when the clock begins. A deadline engine that hard-codes the federal numbers will quietly miss the very state rules that primacy agencies enforce most aggressively.

Prerequisites & Environment Setup

The implementation targets Python 3.11+ so that the standard-library zoneinfo module is available for timezone-aware arithmetic; every deadline is computed against a real IANA zone, never a naive datetime. Deadline definitions are validated with pydantic so that a malformed configuration fails loudly at load time rather than silently producing a wrong due date. The only third-party dependency is pydantic; zoneinfo and datetime ship with the interpreter.

Before writing any code you need one non-package input: the current public-notification rule set for each primacy agency your utility reports to. Most systems answer to a single state, but wholesalers, multi-state authorities, and tribal systems under direct EPA primacy may track several at once. Treat that rule set the way the Monitoring Frequency Scheduling module treats sampling calendars: as reviewed, version-controlled configuration with an effective date, not as constants buried in code. When a primacy agency amends a window, the change should arrive as a new configuration revision with its own effective date, so that a deadline computed last quarter is still reproducible against the rule that was in force at the time. That reproducibility is what an auditor asks for when a notice is disputed, and it is far easier to guarantee from a version history than to reconstruct from memory.

python3 -m venv .venv && source .venv/bin/activate
pip install "pydantic==2.7.*"
# tzdata is bundled on most platforms; install it explicitly where the OS omits it
pip install "tzdata==2024.1"

Step-by-Step Implementation

Step 1 — Model the per-agency deadline configuration

Each rule is a small record: which tier it applies to, the maximum interval to first notice, the interval at which the notice must be repeated until the violation is resolved, and the timezone whose local midnight or business hours the agency measures against. Encoding this as a pydantic model turns a spreadsheet of state rules into a validated table. The tighter_of semantics matter: when a state window is shorter than the federal floor, the state value governs; the loader below stores whatever the agency published and never assumes the federal number.

from __future__ import annotations

from datetime import timedelta
from enum import IntEnum
from typing import Optional

from pydantic import BaseModel, Field, field_validator


class Tier(IntEnum):
    TIER_1 = 1  # acute risk to health
    TIER_2 = 2  # non-acute risk
    TIER_3 = 3  # informational / annual


class DeadlineRule(BaseModel):
    """One primacy agency's public-notification deadline for one tier."""

    agency: str = Field(..., description="Primacy agency code, e.g. 'US-EPA', 'CA-SWRCB'")
    tier: Tier
    first_notice: timedelta = Field(..., description="Max interval from known-at to first notice")
    repeat_interval: Optional[timedelta] = Field(
        None, description="Cadence of repeat notices until resolved; None = no repeat"
    )
    timezone: str = Field("UTC", description="IANA zone the agency measures deadlines in")
    citation: str = Field("", description="Regulatory citation for audit provenance")

    @field_validator("first_notice", "repeat_interval")
    @classmethod
    def _must_be_positive(cls, v: Optional[timedelta]) -> Optional[timedelta]:
        if v is not None and v <= timedelta(0):
            raise ValueError("deadline intervals must be positive")
        return v


# Version-controlled table. Federal floor (Subpart Q) plus two state overrides
# that tighten the Tier 2 window. Load this from YAML/DB in production.
DEADLINE_RULES: list[DeadlineRule] = [
    DeadlineRule(agency="US-EPA", tier=Tier.TIER_1, first_notice=timedelta(hours=24),
                 repeat_interval=None, timezone="UTC",
                 citation="40 CFR 141.202(b)"),
    DeadlineRule(agency="US-EPA", tier=Tier.TIER_2, first_notice=timedelta(days=30),
                 repeat_interval=timedelta(days=90), timezone="UTC",
                 citation="40 CFR 141.203(b)"),
    DeadlineRule(agency="US-EPA", tier=Tier.TIER_3, first_notice=timedelta(days=365),
                 repeat_interval=None, timezone="UTC",
                 citation="40 CFR 141.204(b)"),
    # State override: 10-day Tier 2 window, tighter than the 30-day floor.
    DeadlineRule(agency="CA-SWRCB", tier=Tier.TIER_2, first_notice=timedelta(days=10),
                 repeat_interval=timedelta(days=90), timezone="America/Los_Angeles",
                 citation="22 CCR 64463.4"),
    DeadlineRule(agency="TX-TCEQ", tier=Tier.TIER_1, first_notice=timedelta(hours=24),
                 repeat_interval=timedelta(days=1), timezone="America/Chicago",
                 citation="30 TAC 290.122"),
]

Step 2 — Resolve the governing rule and compute the due instant

Given a violation’s tier and the agency that oversees it, the tracker selects the matching rule and adds the interval to the moment the violation became known — not the moment it occurred. Subpart Q keys the clock to the point of knowledge, so the tracker takes an explicit known_at timestamp that upstream detection supplies. All arithmetic runs in the agency’s declared zone and the result is normalized back to UTC for storage, which keeps the due instant stable across daylight-saving transitions.

from dataclasses import dataclass
from datetime import datetime, timezone
from zoneinfo import ZoneInfo


class NoRuleError(LookupError):
    """No configured deadline rule matched the agency/tier pair."""


def resolve_rule(agency: str, tier: Tier,
                 rules: list[DeadlineRule] = DEADLINE_RULES) -> DeadlineRule:
    matches = [r for r in rules if r.agency == agency and r.tier == tier]
    if not matches:
        raise NoRuleError(f"no deadline rule for agency={agency} tier={tier}")
    if len(matches) > 1:
        raise ValueError(f"ambiguous rules for agency={agency} tier={tier}")
    return matches[0]


@dataclass(frozen=True)
class NoticeDeadline:
    agency: str
    tier: Tier
    known_at: datetime      # UTC, when the violation became known
    due_at: datetime        # UTC, first-notice deadline
    citation: str


def compute_deadline(agency: str, tier: Tier, known_at: datetime,
                     rules: list[DeadlineRule] = DEADLINE_RULES) -> NoticeDeadline:
    if known_at.tzinfo is None:
        raise ValueError("known_at must be timezone-aware")
    rule = resolve_rule(agency, tier, rules)

    # Add the interval in the agency's local wall-clock frame, then store UTC.
    local = known_at.astimezone(ZoneInfo(rule.timezone))
    due_local = local + rule.first_notice
    due_at = due_local.astimezone(timezone.utc)

    return NoticeDeadline(
        agency=agency, tier=tier,
        known_at=known_at.astimezone(timezone.utc),
        due_at=due_at, citation=rule.citation,
    )

Step 3 — Measure remaining time and classify status

The time still available on a deadline is simply the difference between the due instant and now:

Δt=tdeadline−tnow\Delta t = t_{deadline} - t_{now}

When Δt\Delta t is positive the notice is still pending; when it crosses zero the deadline is breached. A small band before zero — a configurable warning threshold — lets the engine surface a notice that is at risk while there is still time to act. The classifier below returns a status so that dashboards and the escalation layer share one definition of “overdue.”

from enum import Enum


class DeadlineStatus(str, Enum):
    PENDING = "PENDING"        # comfortably ahead of the deadline
    AT_RISK = "AT_RISK"        # inside the warning band, not yet overdue
    OVERDUE = "OVERDUE"        # deadline passed, notice not confirmed sent


def time_remaining(deadline: NoticeDeadline,
                   now: Optional[datetime] = None) -> timedelta:
    now = now or datetime.now(timezone.utc)
    return deadline.due_at - now              # Δt = t_deadline − t_now


def classify(deadline: NoticeDeadline,
             notice_sent_at: Optional[datetime] = None,
             warning_band: timedelta = timedelta(hours=4),
             now: Optional[datetime] = None) -> DeadlineStatus:
    now = now or datetime.now(timezone.utc)
    # A confirmed send before the due instant closes the obligation.
    if notice_sent_at is not None and notice_sent_at <= deadline.due_at:
        return DeadlineStatus.PENDING
    remaining = deadline.due_at - now
    if remaining <= timedelta(0):
        return DeadlineStatus.OVERDUE
    if remaining <= warning_band:
        return DeadlineStatus.AT_RISK
    return DeadlineStatus.PENDING

The warning band should scale with the tier: a four-hour band is meaningful against a 24-hour Tier 1 clock but pointless against an annual Tier 3 deadline, where a band of several weeks gives compliance staff room to assemble the notice. The status vocabulary here maps cleanly onto the alert codes produced by Violation Code Classification, so an overdue deadline can be raised as its own distinct alert type rather than being folded into the original water-quality violation.

Deadline status state machine: PENDING, AT_RISK, OVERDUE and SATISFIED A new deadline starts in PENDING. When remaining time falls inside the warning band it becomes AT_RISK. A confirmed send from PENDING or AT_RISK transitions to SATISFIED. If the due instant passes with no confirmed send the deadline becomes OVERDUE, which fires an escalation and re-escalates on each repeat interval until a confirmed send transitions it to SATISFIED. PENDING AT_RISK OVERDUE SATISFIED ahead of clock in warning band escalates + repeats confirmed send Δt ≤ band Δt ≤ 0 send send send repeat

Step 4 — Track many deadlines and escalate the overdue

A production system carries dozens of open deadlines at once, each on its own clock and often against a different agency. The tracker below holds the live set, re-classifies them on every poll, and yields two actionable lists: those that have entered the warning band and those already overdue. For overdue Tier 1 and repeating Tier 2 obligations it also computes the next repeat-notice instant from the rule’s repeat_interval, so the escalation layer knows not just that a notice is late but when the next one is owed. The register and resolve methods bound the set: a deadline enters when detection reports a known violation and leaves only when the obligation is closed, either by a confirmed send or by the violation being cleared. Keeping that set small and explicit is what lets a single scan call drive both an operator dashboard and an automated escalation without either one re-deriving the deadline math.

class DeadlineTracker:
    def __init__(self, rules: list[DeadlineRule] = DEADLINE_RULES):
        self._rules = rules
        self._open: dict[str, NoticeDeadline] = {}

    def register(self, violation_id: str, agency: str, tier: Tier,
                 known_at: datetime) -> NoticeDeadline:
        dl = compute_deadline(agency, tier, known_at, self._rules)
        self._open[violation_id] = dl
        return dl

    def resolve(self, violation_id: str) -> None:
        self._open.pop(violation_id, None)

    def next_repeat_at(self, deadline: NoticeDeadline) -> Optional[datetime]:
        rule = resolve_rule(deadline.agency, deadline.tier, self._rules)
        if rule.repeat_interval is None:
            return None
        return deadline.due_at + rule.repeat_interval

    def scan(self, now: Optional[datetime] = None,
             warning_band: timedelta = timedelta(hours=4)):
        now = now or datetime.now(timezone.utc)
        at_risk, overdue = [], []
        for vid, dl in self._open.items():
            status = classify(dl, warning_band=warning_band, now=now)
            if status is DeadlineStatus.OVERDUE:
                overdue.append((vid, dl, self.next_repeat_at(dl)))
            elif status is DeadlineStatus.AT_RISK:
                at_risk.append((vid, dl))
        return at_risk, overdue

Configuration Reference

The federal floor and two representative state overrides are shown below; every primacy agency you report to needs its own reviewed row with an effective date and citation.

Agency Tier First notice Repeat interval Timezone Citation
US-EPA 1 24 hours none UTC 40 CFR 141.202(b)
US-EPA 2 30 days 90 days UTC 40 CFR 141.203(b)
US-EPA 3 365 days none UTC 40 CFR 141.204(b)
CA-SWRCB 2 10 days 90 days America/Los_Angeles 22 CCR 64463.4
TX-TCEQ 1 24 hours 1 day America/Chicago 30 TAC 290.122
Field / status Type Default Meaning
first_notice timedelta — Maximum interval from known_at to first notice
repeat_interval timedelta None Cadence of repeat notices until the violation is resolved
timezone str UTC IANA zone the agency measures the deadline in
warning_band timedelta 4 hours Lead time before due that flips status to AT_RISK
PENDING status — Ahead of the deadline, or a confirmed send closed the obligation
AT_RISK status — Inside the warning band, not yet overdue
OVERDUE status — Due instant passed with no confirmed send; escalate

Verification & Testing

Deadline math must be deterministic under a fixed now, and the daylight-saving behavior must be pinned by test, because a one-hour drift on a 24-hour clock is the difference between compliant and late.

from datetime import datetime, timezone


def _rule_ca_tier2():
    return [r for r in DEADLINE_RULES
            if r.agency == "CA-SWRCB" and r.tier == Tier.TIER_2]


def test_state_override_beats_federal_floor():
    known = datetime(2026, 3, 1, 12, 0, tzinfo=timezone.utc)
    dl = compute_deadline("CA-SWRCB", Tier.TIER_2, known)
    assert dl.due_at - known == timedelta(days=10)  # not the 30-day floor


def test_remaining_and_overdue_classification():
    known = datetime(2026, 7, 1, 0, 0, tzinfo=timezone.utc)
    dl = compute_deadline("US-EPA", Tier.TIER_1, known)   # 24h window
    before = datetime(2026, 7, 1, 20, 0, tzinfo=timezone.utc)
    after = datetime(2026, 7, 2, 1, 0, tzinfo=timezone.utc)
    assert time_remaining(dl, now=before) == timedelta(hours=4)
    assert classify(dl, now=before) is DeadlineStatus.AT_RISK
    assert classify(dl, now=after) is DeadlineStatus.OVERDUE


def test_confirmed_send_closes_obligation():
    known = datetime(2026, 7, 1, 0, 0, tzinfo=timezone.utc)
    dl = compute_deadline("US-EPA", Tier.TIER_1, known)
    sent = datetime(2026, 7, 1, 10, 0, tzinfo=timezone.utc)
    late = datetime(2026, 7, 3, 0, 0, tzinfo=timezone.utc)
    assert classify(dl, notice_sent_at=sent, now=late) is DeadlineStatus.PENDING

Acceptance criteria before promoting the tracker to production:

Troubleshooting & Gotchas

  • The clock starts at the wrong moment. Adding the interval to when the violation occurred rather than when it became known silently shortens or lengthens every deadline. Feed compute_deadline the point-of-knowledge timestamp that detection records, and store both timestamps for audit.
  • Federal numbers quietly override a state rule. If the configuration only carries the Subpart Q floor, a 10-day state Tier 2 window is missed by 20 days. Load every agency’s published rule and let resolve_rule raise NoRuleError when a pair is unconfigured, rather than defaulting to federal values.
  • Naive datetimes drift by an hour twice a year. Computing a 24-hour deadline on a naive local timestamp across a daylight-saving boundary lands the due instant 60 minutes off. Require tzinfo on every input and do the arithmetic in the agency’s IANA zone before normalizing to UTC.
  • Escalation storms on a repeating deadline. A Tier 1 rule with a one-day repeat interval will re-fire every scan once overdue. Debounce by recording the last escalation instant and only re-alerting once next_repeat_at has passed.
  • A sent notice never clears the deadline. If the dispatch layer does not report a confirmed-send timestamp back to the tracker, classify keeps returning OVERDUE. Close the loop from the delivery path described in the multi-channel dispatch section so resolve or notice_sent_at is set on success.

Related pages