This guide is part of the Audit Trail & Evidence Automation section, and it solves one narrow problem: given a grant record, compute exactly when it becomes eligible for destruction under federal rules — and guarantee that no automated job ever deletes it while a hold is active.
Record retention looks like a solved problem until an auditor asks you to prove that a document destroyed last quarter was actually eligible for destruction. A hardcoded “keep everything seven years” cron job is not a defense: it over-retains sensitive records that should have been purged, and it will happily delete a document that a pending audit still needs. This guide builds a retention engine that anchors the clock on the correct start event, extends it when litigation or an audit intervenes, respects an explicit legal hold, and writes an immutable decision record for every judgment it makes.
When to Use This Approach
Reach for a computed retention engine — rather than a fixed calendar rule — when all three conditions hold:
- Your records fall under the federal Uniform Guidance. 2 CFR §200.334 sets a three-year retention period for financial records, supporting documents, statistical records, and all other records pertinent to a federal award. The period generally starts from the date the recipient submits its final expenditure or financial report — not from the award date and not from the calendar year-end. If your grants are federal pass-through or direct awards, the start event is a data point you must capture, not assume.
- The retention clock can be extended by events you do not control. Under 2 CFR §200.334(a), if any litigation, claim, or audit is started before the three-year period expires, the records must be retained until all such actions are resolved and final action is taken. A static schedule cannot model this; you need a gate that inspects open actions at the moment of decision.
- Destruction is itself an auditable act. Deleting a record you were required to keep is a finding; retaining PII you should have purged is a separate exposure. Either way, the destruction decision needs an immutable record, which is why every judgment here is written to the ledger described in Building an Append-Only Audit Ledger in Postgres.
Special start events — real property and equipment records run from disposition, and some program-income and indirect-cost records key off different anchors — are modeled here as an overridable start-event type, but the disposition-tracking workflow that supplies those dates is out of scope. So is the storage tier itself: where the bytes physically live and how deletion is executed against object storage belongs to your data platform. This engine computes whether and when, and emits the decision; it does not own the delete syscall.
Step-by-Step Implementation
The reference implementation targets Python 3.10+ and uses only the standard library plus python-dateutil for calendar-correct year arithmetic:
python-dateutil==2.9.0.post0
One contract to internalize first: the engine never deletes; it decides. Every function returns a typed decision object. A separate purge job acts on that decision, and it treats an active hold as an absolute veto — no accuracy score, no override flag, no “force” argument can talk it into destroying a held record.
Step 1: Anchor the retention clock on the correct start event
The single most common retention error is starting the clock at the wrong event. 2 CFR §200.334 fixes the general anchor at the submission of the final expenditure or financial report, but real property, equipment, and a handful of other categories start elsewhere. Model the anchor as a typed enum and resolve the actual start date from the record, refusing to guess when the required date is missing.
import logging
from dataclasses import dataclass, field
from datetime import date
from enum import Enum
from typing import Optional
AUDIT_LOGGER = logging.getLogger("grant_retention.audit")
AUDIT_LOGGER.setLevel(logging.INFO)
_handler = logging.StreamHandler()
_handler.setFormatter(logging.Formatter("%(asctime)s | %(levelname)s | %(message)s"))
AUDIT_LOGGER.addHandler(_handler)
class StartEvent(Enum):
"""Retention anchor per 2 CFR 200.334; general case is final report submission."""
FINAL_REPORT = "final_report_submitted"
PROPERTY_DISPOSITION = "real_property_or_equipment_disposition"
INDIRECT_COST_PROPOSAL = "indirect_cost_rate_proposal_submitted"
class RetentionError(Exception):
"""Raised when a start event cannot be resolved to a concrete date."""
@dataclass(frozen=True)
class GrantRecord:
record_id: str
start_event: StartEvent
start_date: Optional[date]
contains_pii: bool = False
def resolve_start_date(record: GrantRecord) -> date:
"""Return the concrete anchor date, refusing to proceed if it is unknown."""
if record.start_date is None:
AUDIT_LOGGER.error(
"Missing start date | record=%s | event=%s", record.record_id, record.start_event.value
)
raise RetentionError(
f"record {record.record_id}: {record.start_event.value} date is required"
)
return record.start_date
Making start_date Optional and raising RetentionError when it is absent is deliberate: a None anchor must halt the pipeline loudly, never silently default to date.today() and start a three-year clock from the wrong day.
Step 2: Compute the base expiry with a typed function
With the anchor resolved, the base expiry is the anchor plus three years. Use dateutil.relativedelta rather than timedelta(days=365 * 3) so leap years land on the correct calendar date. Keep this function pure — no I/O, no hold logic — so it is trivially testable and the tolling rules in the next step compose on top of it.
from dateutil.relativedelta import relativedelta
RETENTION_YEARS: int = 3 # 2 CFR 200.334 general period
def compute_base_expiry(record: GrantRecord, years: int = RETENTION_YEARS) -> date:
"""Anchor + statutory period. Pure function: no holds, no clock reads."""
anchor = resolve_start_date(record)
expiry = anchor + relativedelta(years=years)
AUDIT_LOGGER.info(
"Base expiry computed | record=%s | anchor=%s | expiry=%s",
record.record_id, anchor.isoformat(), expiry.isoformat(),
)
return expiry
relativedelta(years=3) maps 2024-02-29 to 2027-02-28 rather than drifting a day, which matters when a report is filed on a leap day and an auditor recomputes the expiry by hand.
Step 3: Apply tolling and legal holds
This is the compliance core. 2 CFR §200.334(a) requires that if litigation, a claim, or an audit is started before the base period expires, records be retained until that action is fully resolved. Separately, an organization may impose an explicit legal hold that suspends any purge indefinitely regardless of the statutory clock. Model both: a tolling event only counts if it opened before the base expiry, and any open hold — statutory or explicit — pushes the effective expiry out to None (indefinite).
from typing import Sequence
class HoldKind(Enum):
LITIGATION = "litigation"
CLAIM = "claim"
AUDIT = "audit"
LEGAL_HOLD = "explicit_legal_hold"
@dataclass(frozen=True)
class Hold:
kind: HoldKind
opened_on: date
resolved_on: Optional[date] = None # None => still open
def effective_expiry(
record: GrantRecord,
holds: Sequence[Hold],
as_of: date,
) -> Optional[date]:
"""Return the enforceable expiry, or None when a hold suspends purge indefinitely."""
base = compute_base_expiry(record)
# An explicit legal hold always suspends purge while open.
if any(h.kind is HoldKind.LEGAL_HOLD and h.resolved_on is None for h in holds):
AUDIT_LOGGER.info("Legal hold active | record=%s | expiry suspended", record.record_id)
return None
# 200.334(a): litigation/claim/audit tolls only if started before base expiry.
tolling = [
h for h in holds
if h.kind in {HoldKind.LITIGATION, HoldKind.CLAIM, HoldKind.AUDIT}
and h.opened_on < base
and h.resolved_on is None
]
if tolling:
AUDIT_LOGGER.info(
"Tolled under 200.334(a) | record=%s | open=%s",
record.record_id, [h.kind.value for h in tolling],
)
return None
return base
The h.opened_on < base test is the literal encoding of “started before the period expires”: an audit opened after the record already aged out does not resurrect a retention obligation, while one opened on day 1,000 of the three years does. An unresolved resolved_on of None is what keeps the record held until final action is taken.
Step 4: Assign a retention class and schedule the purge job
Now fold the effective expiry into a retention_class and a purge decision. The class drives storage tiering and PII handling; the decision drives the job. The rule the job obeys without exception: it destroys only when the effective expiry is a real date that has passed. None — the sentinel for an active hold or tolling — is never “expired,” so a held record can never be selected for deletion.
class RetentionClass(Enum):
ACTIVE = "active" # within base period
HELD = "held" # suspended by hold or tolling
ELIGIBLE_FOR_PURGE = "eligible_for_purge"
@dataclass(frozen=True)
class RetentionDecision:
record_id: str
retention_class: RetentionClass
effective_expiry: Optional[date]
as_of: date
reason: str
def classify(record: GrantRecord, holds: Sequence[Hold], as_of: date) -> RetentionDecision:
"""Map a record to its retention class as of a given date."""
expiry = effective_expiry(record, holds, as_of)
if expiry is None:
cls, reason = RetentionClass.HELD, "hold or 200.334(a) tolling active"
elif as_of >= expiry:
cls, reason = RetentionClass.ELIGIBLE_FOR_PURGE, f"base period ended {expiry.isoformat()}"
else:
cls, reason = RetentionClass.ACTIVE, f"within retention until {expiry.isoformat()}"
return RetentionDecision(record.record_id, cls, expiry, as_of, reason)
def purge_if_eligible(decision: RetentionDecision) -> bool:
"""Destroy ONLY on an eligible decision. An active hold is an absolute veto."""
if decision.retention_class is not RetentionClass.ELIGIBLE_FOR_PURGE:
AUDIT_LOGGER.info(
"Purge skipped | record=%s | class=%s | %s",
decision.record_id, decision.retention_class.value, decision.reason,
)
return False
# ... hand the record id to the storage layer's delete primitive here ...
AUDIT_LOGGER.warning(
"Record purged | record=%s | expired=%s",
decision.record_id, decision.effective_expiry.isoformat(),
)
return True
Guarding on decision.retention_class is not ELIGIBLE_FOR_PURGE — rather than re-deriving expiry inside the job — means the veto lives in exactly one place. There is no code path where a HELD record reaches the delete primitive.
Step 5: Log every retention decision
A destruction you cannot explain is worse than one you never made. Serialize each RetentionDecision to a structured event and append it to the tamper-evident ledger; that record — not the absence of the file — is what proves to an auditor that the purge was authorized. Signing the serialized decision so it cannot be backdated is covered in Cryptographically Signing Grant Compliance Artifacts.
import json
from typing import Callable
def emit_decision(
decision: RetentionDecision,
sink: Callable[[str], None],
) -> None:
"""Serialize a retention decision to an append-only audit sink."""
payload = {
"record_id": decision.record_id,
"retention_class": decision.retention_class.value,
"effective_expiry": decision.effective_expiry.isoformat() if decision.effective_expiry else None,
"as_of": decision.as_of.isoformat(),
"reason": decision.reason,
"authority": "2 CFR 200.334",
}
try:
sink(json.dumps(payload, sort_keys=True))
except Exception as exc:
AUDIT_LOGGER.critical("Decision log failed | record=%s | %s", decision.record_id, exc)
raise RetentionError(f"could not persist decision for {decision.record_id}") from exc
If the sink write fails, emit_decision raises rather than swallowing the error — an unlogged purge decision must abort the whole operation, because proceeding would destroy the record and lose the only evidence that the destruction was lawful.
Verification
Confirm the engine behaves correctly with four checks:
- The base clock uses the right anchor and leap-year math. Build a record with
start_event=FINAL_REPORTandstart_date=date(2024, 2, 29)and assertcompute_base_expiryreturnsdate(2027, 2, 28), provingrelativedeltais used rather than a 365-day multiple. - A missing anchor halts loudly. Pass a record with
start_date=Noneand assertresolve_start_dateraisesRetentionErrorand that no expiry is produced. - Tolling suspends purge only when opened in time. Assert that an
AUDIThold opened before the base expiry withresolved_on=Noneyieldseffective_expiry == Noneand classHELD, while the same audit opened after the base expiry leaves the recordELIGIBLE_FOR_PURGE. - The purge veto is absolute. Assert
purge_if_eligiblereturnsFalsefor every non-eligible class, including a record whose base date passed years ago but carries an openLEGAL_HOLD.
A compliant run emits one INFO line per decision and a single WARNING only when a record is actually destroyed. Ship those to the append-only ledger so the trail itself satisfies the three-year period under 2 CFR §200.334.
rec = GrantRecord("G-2024-118", StartEvent.FINAL_REPORT, date(2021, 3, 15))
audit = [Hold(HoldKind.AUDIT, opened_on=date(2023, 9, 1))] # opened before 2024-03-15 expiry
decision = classify(rec, audit, as_of=date(2026, 7, 10))
assert decision.retention_class is RetentionClass.HELD
assert decision.effective_expiry is None
assert purge_if_eligible(decision) is False
Common Errors & Fixes
| Error | Cause | Fix |
|---|---|---|
| Records purged three years after the award date | Clock anchored on award, not the final report submission | Resolve the anchor via StartEvent/resolve_start_date; the general case starts at the final expenditure report per §200.334. |
| A record needed for an open audit was deleted | Purge job re-derived expiry and ignored tolling | Route every deletion through purge_if_eligible; an open litigation/claim/audit under §200.334(a) returns None expiry and class HELD. |
| Expiry lands one day early on leap-year filings | timedelta(days=365*3) used instead of calendar arithmetic |
Use dateutil.relativedelta(years=3) so 02-29 maps to 02-28, not a drifted day. |
| Legal hold silently overridden by a “force purge” flag | Deletion logic accepts an override argument | Remove the override; the only gate is retention_class is ELIGIBLE_FOR_PURGE, and an open legal hold makes that impossible. |
| Cannot prove a past destruction was authorized | Decisions logged with print or not at all |
Serialize each RetentionDecision via emit_decision to the append-only ledger, and raise if the sink write fails. |
Related
- Parent section: Audit Trail & Evidence Automation
- Where retention decisions are stored immutably: Building an Append-Only Audit Ledger in Postgres
- Signing each decision so it cannot be backdated: Cryptographically Signing Grant Compliance Artifacts
- The broader compliance-mapping framework: Core Architecture & Compliance Mapping