FAIR Data Automation Research Data Management

Automating Data Retention Policy Enforcement

Retention policies written in a governance document are unenforceable until something evaluates them against real datasets on a schedule. The gap between “our policy says clinical data is kept ten years, then reviewed” and a system that actually flags the dataset the day it crosses that line is where most research-data governance quietly fails: schedules live in a PDF, nobody runs them, and storage grows until an audit or a subject-access request forces a manual scramble. This page turns the schedule into policy-as-code — each retention rule expressed as a validated data structure carrying a retention period, an optional legal hold, and a disposition action — and builds an evaluator that runs over every dataset, decides what should happen to each, and records the decision in an append-only audit log. It is the enforcement layer for the schedules you design with the Data Governance Frameworks overview, and it assumes you already have a dataset inventory and are comfortable with Python 3.10+ and the Pydantic V2 API.

The design goal is that enforcement is deterministic and reversible-by-review. Deterministic means the same dataset and the same policy always yield the same action, so a run is auditable and reproducible. Reversible-by-review means the evaluator never deletes anything directly: it emits a disposition action — retain, review, or purge — and routes purges through a human or a delayed queue, so a misconfigured rule causes a flagged review, not silent data loss. The single hardest requirement, and the one that trips up naive implementations, is the legal hold: an active hold must override every disposition, no matter how far past its retention date a dataset is. The evaluator below enforces that ordering structurally.

Retention policy evaluation flow A top-down evaluation flow. Each dataset is evaluated against its policy. A first decision asks whether a legal hold is active; if yes, the dataset is retained under hold. If no, a second decision asks whether the dataset is past its retention period; if no, it is retained as active. If yes, a disposition action is emitted, routed to a review or purge queue, and the decision is written to an append-only audit log. Evaluate dataset against its policy Legal hold active? yes Retain — under hold disposition suspended no Past retention period? no Retain — active re-check next run yes Emit disposition review or purge Route to queue review / purge worker Append audit log immutable, hash-chained

Retention Rule Fields

A retention rule is policy-as-code: a small set of fields that together fully determine what the evaluator does with a dataset. The table below is the core reference — each field, what it means, and the enforcement behaviour it drives. There are no optional-but-undefined fields; every one has a concrete effect.

Field Meaning Enforcement action it drives
policy_id Stable identifier for the rule, cited in the audit log Ties every emitted action back to the rule that produced it
category Data class the rule applies to (e.g. clinical, derived, instrument-raw) Selects which datasets the rule matches
retention_years Minimum years to keep from the reference date Sets the disposition date; before it, the action is always RETAIN
reference_event What starts the clock: created, published, or last_accessed Determines which timestamp the retention period is measured from
disposition Action once the period elapses: REVIEW, PURGE, or ARCHIVE The action emitted for an expired, un-held dataset
legal_hold Boolean or hold reference freezing all disposition When active, overrides disposition and forces RETAIN
grace_days Buffer after the disposition date before action fires Prevents same-day churn; action waits until the grace window closes

Two fields carry most of the risk. legal_hold must sit above the retention calculation in the control flow, not beside it — a dataset ten years past its retention date but under an active litigation hold must be retained, and any implementation that checks the date first and the hold second will eventually purge held data. reference_event is the quiet one: measuring a ten-year period from last_accessed rather than created produces wildly different disposition dates, and choosing the wrong anchor is a policy error the code cannot catch for you. The building-blocks for expressing these schedules in a researcher-facing plan are covered in building a data management plan template for researchers.

Production Implementation

The policy is a Pydantic V2 model so a malformed rule fails at load time, not mid-run over live data. The evaluator is a pure function of a dataset record and a policy: it returns a decision object and never mutates storage, which keeps it testable and keeps destructive effects behind the queue and audit boundary. Legal hold is checked first and short-circuits, so the override is structural rather than a convention a future edit might break.

python
from __future__ import annotations

import hashlib
import json
from datetime import date, timedelta
from enum import Enum

from pydantic import BaseModel, Field, field_validator


class Disposition(str, Enum):
    REVIEW = "REVIEW"
    PURGE = "PURGE"
    ARCHIVE = "ARCHIVE"


class ReferenceEvent(str, Enum):
    CREATED = "created"
    PUBLISHED = "published"
    LAST_ACCESSED = "last_accessed"


class Action(str, Enum):
    RETAIN = "RETAIN"           # keep; nothing due yet, or under legal hold
    REVIEW = "REVIEW"           # expired; route to human review
    PURGE = "PURGE"             # expired; route to deletion worker
    ARCHIVE = "ARCHIVE"         # expired; route to cold archival tier


class RetentionPolicy(BaseModel):
    policy_id: str = Field(min_length=1)
    category: str = Field(min_length=1)
    retention_years: int = Field(ge=0, le=100)
    reference_event: ReferenceEvent = ReferenceEvent.CREATED
    disposition: Disposition = Disposition.REVIEW
    legal_hold: bool = False
    grace_days: int = Field(default=0, ge=0, le=365)

    @field_validator("policy_id")
    @classmethod
    def no_whitespace_id(cls, v: str) -> str:
        if v != v.strip() or " " in v:
            raise ValueError("policy_id must not contain whitespace")
        return v


class DatasetRecord(BaseModel):
    dataset_id: str = Field(min_length=1)
    category: str
    created: date
    published: date | None = None
    last_accessed: date | None = None
    under_hold: bool = False   # dataset-level hold, independent of the policy


class Decision(BaseModel):
    dataset_id: str
    policy_id: str
    action: Action
    disposition_date: date
    reason: str


def _reference_date(record: DatasetRecord, event: ReferenceEvent) -> date:
    """Resolve the timestamp the retention clock is measured from."""
    if event is ReferenceEvent.PUBLISHED:
        return record.published or record.created
    if event is ReferenceEvent.LAST_ACCESSED:
        return record.last_accessed or record.created
    return record.created


def evaluate(
    record: DatasetRecord,
    policy: RetentionPolicy,
    *,
    today: date,
) -> Decision:
    """Decide the action for one dataset under one policy on a given day.

    Legal hold is evaluated FIRST and overrides everything: a held dataset is
    always retained, regardless of how far past its disposition date it is.
    """
    anchor = _reference_date(record, policy.reference_event)
    disposition_date = anchor.replace(year=anchor.year + policy.retention_years)
    effective_date = disposition_date + timedelta(days=policy.grace_days)

    if policy.legal_hold or record.under_hold:
        return Decision(
            dataset_id=record.dataset_id,
            policy_id=policy.policy_id,
            action=Action.RETAIN,
            disposition_date=disposition_date,
            reason="legal hold active — disposition suspended",
        )

    if today < effective_date:
        return Decision(
            dataset_id=record.dataset_id,
            policy_id=policy.policy_id,
            action=Action.RETAIN,
            disposition_date=disposition_date,
            reason=f"within retention until {effective_date.isoformat()}",
        )

    action = Action[policy.disposition.value]  # REVIEW / PURGE / ARCHIVE
    return Decision(
        dataset_id=record.dataset_id,
        policy_id=policy.policy_id,
        action=action,
        disposition_date=disposition_date,
        reason=f"past retention (+{policy.grace_days}d grace) — {action.value}",
    )


class AuditLog:
    """Append-only, hash-chained log of every disposition decision."""

    def __init__(self) -> None:
        self._entries: list[dict[str, str]] = []
        self._head = "0" * 64  # genesis hash

    def append(self, decision: Decision, *, at: date) -> str:
        body = {
            "prev": self._head,
            "timestamp": at.isoformat(),
            "dataset_id": decision.dataset_id,
            "policy_id": decision.policy_id,
            "action": decision.action.value,
            "reason": decision.reason,
        }
        serialized = json.dumps(body, sort_keys=True, separators=(",", ":"))
        digest = hashlib.sha256(serialized.encode("utf-8")).hexdigest()
        self._entries.append({**body, "hash": digest})
        self._head = digest
        return digest

    def verify(self) -> bool:
        """Recompute the chain; any tampering breaks the linkage."""
        prev = "0" * 64
        for entry in self._entries:
            if entry["prev"] != prev:
                return False
            body = {k: entry[k] for k in
                    ("prev", "timestamp", "dataset_id", "policy_id", "action", "reason")}
            serialized = json.dumps(body, sort_keys=True, separators=(",", ":"))
            if hashlib.sha256(serialized.encode("utf-8")).hexdigest() != entry["hash"]:
                return False
            prev = entry["hash"]
        return True


def enforce(
    records: list[DatasetRecord],
    policies: dict[str, RetentionPolicy],
    *,
    today: date,
    log: AuditLog,
) -> list[Decision]:
    """Evaluate every dataset against its category policy and log each decision."""
    decisions: list[Decision] = []
    for record in records:
        policy = policies.get(record.category)
        if policy is None:
            continue  # no rule for this category — left for a catch-all pass
        decision = evaluate(record, policy, today=today)
        log.append(decision, at=today)
        decisions.append(decision)
    return decisions

The enforce loop is intentionally boring: match each dataset to the policy for its category, evaluate, log, collect. The interesting invariants — hold overrides disposition, grace defers the action, the reference event selects the anchor — all live in evaluate, which is a pure function you can exhaustively test. The audit log’s hash chain is what makes the record defensible: each entry commits to the previous entry’s hash, so removing or editing any past decision breaks verify. The access-control boundary that gates who may run enforce or read the log is covered in Security & Access Control.

Verification

The tests pin the three behaviours most likely to regress: a legal hold overrides an expired dataset, the grace window defers an otherwise-due purge, and the audit chain detects tampering.

python
from datetime import date


def _policy(**kw: object) -> RetentionPolicy:
    base = dict(policy_id="clin-10y", category="clinical", retention_years=10)
    return RetentionPolicy(**{**base, **kw})


def test_legal_hold_overrides_expiry() -> None:
    rec = DatasetRecord(dataset_id="D1", category="clinical", created=date(2000, 1, 1))
    pol = _policy(disposition=Disposition.PURGE, legal_hold=True)
    decision = evaluate(rec, pol, today=date(2026, 7, 16))
    assert decision.action is Action.RETAIN


def test_grace_window_defers_action() -> None:
    rec = DatasetRecord(dataset_id="D2", category="clinical", created=date(2016, 7, 16))
    pol = _policy(disposition=Disposition.PURGE, grace_days=30)
    # exactly on the disposition date, still inside the grace buffer
    assert evaluate(rec, pol, today=date(2026, 7, 16)).action is Action.RETAIN
    assert evaluate(rec, pol, today=date(2026, 8, 20)).action is Action.PURGE


def test_audit_chain_detects_tampering() -> None:
    log = AuditLog()
    rec = DatasetRecord(dataset_id="D3", category="clinical", created=date(2010, 1, 1))
    log.append(evaluate(rec, _policy(disposition=Disposition.REVIEW), today=date(2026, 7, 16)),
               at=date(2026, 7, 16))
    assert log.verify() is True
    log._entries[0]["action"] = "PURGE"   # forge a past decision
    assert log.verify() is False

Run it with pytest -q test_retention.py. A passing run prints 3 passed; the tampering test is the one auditors care about, because it demonstrates that a purge decision cannot be rewritten after the fact without the chain flagging it.

Gotchas

  • Checking the retention date before the legal hold. The most dangerous ordering bug: a dataset past its date is purged before the hold is consulted, destroying evidence under litigation. Fix: evaluate legal_hold (and any dataset-level hold) first and short-circuit to RETAIN, exactly as evaluate does — never as a filter applied after the disposition is chosen.
  • Purging directly from the evaluator. Wiring evaluate straight to a DELETE makes a single bad policy irreversible and unauditable. Fix: emit an action and route PURGE through a queue with a review or delay step, and write the decision to the append-only log before the worker acts, so every deletion has a prior committed record.
  • Ambiguous reference events across time zones and leap years. Measuring retention_years by naively adding to a year can raise on 29 February, and mixing local and UTC dates shifts disposition by a day at boundaries. Fix: store reference timestamps as explicit dates, keep the anchor selection in one function, and test the leap-day and grace-boundary cases directly.

FAQ

Why emit an action instead of deleting expired data directly?

Because retention enforcement must be reversible until a human or a delayed process confirms it. A policy typo — the wrong reference_event, a retention_years off by a decade — should surface as a queue of flagged datasets a curator can inspect, not as data that is already gone. Emitting REVIEW, PURGE, or ARCHIVE and routing through a worker keeps the destructive step behind a boundary, and writing the decision to the append-only log before the worker acts means every deletion is preceded by a committed, verifiable record.

Structurally, not by convention. The evaluator checks legal_hold and any dataset-level hold as its first branch and returns RETAIN immediately, so the retention-date arithmetic never even runs for a held dataset. That ordering is the whole point: a dataset can be twenty years past its disposition date, but while a hold is active the only possible action is retention. Encoding the override as the first short-circuit rather than a later filter is what prevents a future refactor from reintroducing the purge-then-check bug.

What makes the audit log trustworthy?

Each entry commits to the hash of the entry before it, forming a chain from a fixed genesis value. Recomputing the chain with verify detects any insertion, deletion, or edit, because changing one entry changes its hash and breaks every link after it. It is not a substitute for write-once storage or access controls, but it turns the log into tamper-evident evidence: you can prove the sequence of disposition decisions was not rewritten after an audit began.

How often should the evaluator run?

Run it on a fixed schedule matched to the granularity of your dispositions — daily is typical, because grace_days and disposition dates are day-resolution. A daily pass over the inventory is idempotent: a dataset still within retention returns RETAIN every day until the effective date passes, so re-running does not double-process anything. Align the cadence with the governance schedule you define in the data-management-plan template rather than running it ad hoc.

See the Open Science Infrastructure Planning overview for how retention enforcement fits the wider governance and deposit pipeline.