Emitting PREMIS Events for Preservation Actions

Every preservation action a repository performs — a fixity check, a format migration, a replication, a deletion — is only trustworthy if it leaves behind a PREMIS Event record that says what happened, to which object, by which agent, with what outcome, and when. This page belongs to the Preservation Action Logging area, and it solves a narrow but ruinous problem: events that are emitted incorrectly. An event that omits its linking agent, invents its own eventType vocabulary, or is written with a fresh random identifier on every retry does not strengthen provenance — it corrupts it, because a later auditor cannot tell one genuine action from three phantom duplicates of the same action. The goal here is a single emission path that produces complete, canonically-typed, and idempotent events so that re-running an action never forks the record.

# Root-Cause Analysis of Broken Provenance

Bad provenance is rarely one big mistake; it is an accumulation of small emission defects that each look survivable in isolation. Five recur across cultural-heritage repositories.

1. Missing linkingAgentIdentifier or linkingObjectIdentifier. A PREMIS Event with no linked object is unattributable — it records that something was validated but not what, and an unlinked agent record hides who or what performed the action. PREMIS treats these links as the connective tissue of the provenance graph; drop them and the event becomes an orphan that satisfies a schema but answers no audit question.
2. Non-canonical eventType vocabulary. One task writes "checksum", another writes "fixity", a third writes "Fixity Check". All three mean the PREMIS fixityCheck action, but a query for fixityCheck now returns a third of the truth. Free-text event types silently fragment the record and make longitudinal reporting impossible.
3. Non-deterministic eventIdentifier causing duplicates on retry. If the identifier is a fresh uuid4() generated inside the task body, then a task that runs, succeeds, and is redelivered by the queue (a real occurrence under at-least-once delivery, described in Implementing Celery for Asynchronous Ingestion Tasks) writes a second event for the same physical action. The ledger now shows two migrations that were really one.
4. eventOutcome not captured. An event that records only that a validation ran, without recording whether it passed, is evidentially worthless. Provenance must distinguish a successful fixity check from a detected corruption; omitting the outcome collapses that distinction.
5. Emission not atomic with the action. When the action completes but the process dies before the event is written — or the event is written but the action is later rolled back — the record and reality diverge. An event lost this way is indistinguishable from an action that never happened.

The fixes below share one architecture: derive the identifier deterministically from the action’s own identity so re-emission is a no-op, and treat writing-then-verifying as a single step so a half-written event is caught immediately. The sequence diagram makes that emission path concrete.

Emission is a four-step path: build a deterministically-identified event, append it once, read it back, and commit — a mismatch triggers a re-emission that is idempotent because the identifier never changes.

# A Canonical PREMIS Event Schema

Every defect above traces back to an unconstrained event shape. Fix the shape first: define one model that makes the required links mandatory, pins eventType to a controlled list, and normalizes the timestamp to UTC. The table states the requirement for each field so the model and the audit rules agree.

Encoded as a pydantic model, these requirements become build-time guarantees — an event that violates them cannot be constructed, let alone written. The controlled vocabulary is the same set the repository uses when it folds events into object metadata through PREMIS Metadata Mapping.

from __future__ import annotations

import hashlib
import logging
import uuid
from datetime import datetime, timezone
from enum import Enum

from pydantic import BaseModel, Field, field_validator, model_validator

logger = logging.getLogger("archival.premis")

# A fixed namespace makes uuid5 reproducible across processes and hosts.
PREMIS_EVENT_NAMESPACE = uuid.UUID("6f2c6d5e-7a1b-5c4d-9e0f-1a2b3c4d5e6f")


class EventType(str, Enum):
    """The PREMIS event-type controlled vocabulary this repository accepts."""

    CAPTURE = "capture"
    MESSAGE_DIGEST = "messageDigestCalculation"
    VALIDATION = "validation"
    VIRUS_CHECK = "virusCheck"
    INGESTION = "ingestion"
    REPLICATION = "replication"
    FIXITY_CHECK = "fixityCheck"
    MIGRATION = "migration"
    DELETION = "deletion"


class Outcome(str, Enum):
    SUCCESS = "success"
    FAILURE = "failure"


class LinkingAgent(BaseModel):
    identifier_type: str = "local"
    identifier_value: str
    role: str  # e.g. "executingProgram", "authorizer"


class LinkingObject(BaseModel):
    identifier_type: str = "local"
    identifier_value: str
    role: str = "source"


class PREMISEvent(BaseModel):
    """A canonical, self-validating PREMIS Event record.

    The event identifier is derived deterministically from the action's own
    identity, so emitting the same action twice yields the same event id.
    """

    event_type: EventType
    event_date_time: datetime
    event_detail: str = Field(min_length=1)
    event_outcome: Outcome
    event_outcome_detail: str = Field(min_length=1)
    linking_agents: list[LinkingAgent] = Field(min_length=1)
    linking_objects: list[LinkingObject] = Field(min_length=1)

    @field_validator("event_date_time")
    @classmethod
    def _require_utc(cls, value: datetime) -> datetime:
        """Reject naive timestamps; normalize everything to UTC."""
        if value.tzinfo is None:
            raise ValueError("event_date_time must be timezone-aware")
        return value.astimezone(timezone.utc)

    @model_validator(mode="after")
    def _log_construction(self) -> "PREMISEvent":
        logger.debug(
            "premis_event_built",
            extra={"event_type": self.event_type.value,
                   "outcome": self.event_outcome.value},
        )
        return self

    @property
    def event_identifier(self) -> str:
        """Deterministic uuid5 over event type, primary object, and time."""
        primary_object = self.linking_objects[0].identifier_value
        stamp = self.event_date_time.isoformat()
        seed = f"{self.event_type.value}|{primary_object}|{stamp}"
        return str(uuid.uuid5(PREMIS_EVENT_NAMESPACE, seed))

Because event_identifier is a pure function of the action’s type, its primary object, and its timestamp, two independent emissions of the same action collapse to the same identifier. That single property is what makes retries safe: the queue may deliver a task twice, but the second delivery cannot mint a second event.

# An Idempotent Emitter with Emit-Then-Verify

The model guarantees a well-formed event; the emitter guarantees it lands exactly once and is readable afterward. The store below is append-only JSON Lines — the simplest durable substrate that never mutates a prior record, and the same discipline scaled up in Designing an Immutable Audit Ledger. The emitter checks for the deterministic id before writing, appends if absent, and then reads the event back to confirm it is durable and equal before returning success.

import json
from pathlib import Path


class AppendOnlyLedger:
    """A minimal append-only PREMIS event store backed by JSON Lines."""

    def __init__(self, path: Path) -> None:
        self._path = path
        self._path.touch(exist_ok=True)

    def contains(self, event_id: str) -> bool:
        return self.read(event_id) is not None

    def append(self, event_id: str, record: dict) -> None:
        line = json.dumps({"eventIdentifier": event_id, **record},
                          sort_keys=True, separators=(",", ":"))
        with self._path.open("a", encoding="utf-8") as handle:
            handle.write(line + "\n")
            handle.flush()  # push the line to the OS before we claim success

    def read(self, event_id: str) -> dict | None:
        with self._path.open("r", encoding="utf-8") as handle:
            for line in handle:
                if not line.strip():
                    continue
                row = json.loads(line)
                if row.get("eventIdentifier") == event_id:
                    return row
        return None


def emit_event(event: PREMISEvent, ledger: AppendOnlyLedger) -> str:
    """Emit a PREMIS event idempotently, then verify it was persisted.

    Returns the deterministic event identifier. Safe to call repeatedly for
    the same action: a redelivery is a no-op that returns the same id.
    """
    event_id = event.event_identifier
    record = {
        "eventType": event.event_type.value,
        "eventDateTime": event.event_date_time.isoformat(),
        "eventDetailInformation": event.event_detail,
        "eventOutcomeInformation": {
            "eventOutcome": event.event_outcome.value,
            "eventOutcomeDetail": event.event_outcome_detail,
        },
        "linkingAgentIdentifier": [a.model_dump() for a in event.linking_agents],
        "linkingObjectIdentifier": [o.model_dump() for o in event.linking_objects],
    }

    if ledger.contains(event_id):
        logger.info("premis_event_already_present",
                    extra={"event_id": event_id, "event_type": event.event_type.value})
        return event_id  # idempotent short-circuit

    ledger.append(event_id, record)

    # Emit-then-verify: prove the event is durable and equal before returning.
    stored = ledger.read(event_id)
    if stored is None:
        logger.error("premis_event_missing_after_write", extra={"event_id": event_id})
        raise RuntimeError(f"event {event_id} not found after append")

    stored_fingerprint = hashlib.sha256(
        json.dumps({k: v for k, v in stored.items() if k != "eventIdentifier"},
                   sort_keys=True).encode("utf-8")
    ).hexdigest()
    expected_fingerprint = hashlib.sha256(
        json.dumps(record, sort_keys=True).encode("utf-8")
    ).hexdigest()
    if stored_fingerprint != expected_fingerprint:
        logger.error("premis_event_verify_mismatch", extra={"event_id": event_id})
        raise RuntimeError(f"event {event_id} read-back did not match")

    logger.info("premis_event_committed",
                extra={"event_id": event_id, "event_type": event.event_type.value,
                       "outcome": event.event_outcome.value})
    return event_id

A calling site ties the emission to the moment an action finishes. Crucially, the timestamp is fixed once and passed into the event, so a retry reuses it and lands on the same deterministic id rather than a new one derived from “now”.

def record_fixity_check(object_id: str, digest_ok: bool, action_time: datetime,
                        ledger: AppendOnlyLedger) -> str:
    """Emit the PREMIS event for a completed fixity check on one object."""
    event = PREMISEvent(
        event_type=EventType.FIXITY_CHECK,
        event_date_time=action_time,
        event_detail="SHA-256 fixity check via hashlib; profile audit-grade",
        event_outcome=Outcome.SUCCESS if digest_ok else Outcome.FAILURE,
        event_outcome_detail=("stored digest matched recomputed digest"
                              if digest_ok else "digest mismatch — object flagged"),
        linking_agents=[LinkingAgent(identifier_value="fixity-worker@1.4.0",
                                     role="executingProgram")],
        linking_objects=[LinkingObject(identifier_value=object_id, role="source")],
    )
    return emit_event(event, ledger)

# Validation and Verification

Confirm the emitter’s guarantees empirically rather than trusting the code path:

• Prove idempotency. Call record_fixity_check twice with the same object_id and action_time. Both calls must return the identical event identifier, and the ledger must gain exactly one line — the second call short-circuits on contains.
• Prove completeness. Assert every persisted record carries a non-empty linkingAgentIdentifier, a non-empty linkingObjectIdentifier, and an eventOutcome in {success, failure}. A row missing any of these is a provenance defect, not a warning.
• Prove canonical typing. Reject any eventType outside the EventType enum at the boundary; a scan of the ledger should never surface a free-text type such as "checksum".
• Prove durability under retry. Kill the process between append and the verify read, then re-run the action. The deterministic id makes the re-run find the already-written line and commit without duplicating it — the record survives a crash exactly as scheduled re-validation expects when it revisits objects under Scheduling Fixity Re-Validation Jobs.

Propagate event_id through structured logging so a single action’s provenance is reconstructable across workers and hosts. When emission is deterministic and verified, the ledger becomes a defensible chain of custody rather than a hopeful log.

# Edge Cases and Gotchas

Three archival-specific traps deserve extra care. First, timestamp granularity: if a migration and its immediately-following fixity check on the same object could land in the same second, the seed must carry sub-second precision or an explicit action counter, or two genuinely distinct events will alias to one identifier. Second, agent identity drift: pin the executing program’s version into linkingAgentIdentifier (fixity-worker@1.4.0, not just fixity-worker), because a later audit needs to know which build performed the action. Third, deletion events still need a linked object: a deletion event whose target has been purged must retain the object’s identifier even after its bytes are gone — the point of the event is to prove the deletion was deliberate and attributable.

For the canonical field semantics, consult the Library of Congress PREMIS Data Dictionary. Adhering to a deterministic, verified emission path yields provenance that satisfies ISO 16363 audit expectations and never double-counts a single preservation action.

# Frequently Asked Questions

# Why use uuid5 instead of uuid4 for the event identifier?

uuid4 is random, so it produces a different identifier every time the code runs — including when a queue redelivers a task that already succeeded, which writes a second event for one real action. uuid5 is a deterministic hash of a namespace and a seed string. By seeding it with the action’s own identity — event type, primary object, and fixed timestamp — the same action always resolves to the same identifier, so a re-emission is detected and skipped. Determinism is precisely what makes idempotent provenance possible.

# What belongs in the uuid5 seed to keep events distinct but idempotent?

Enough to uniquely name the real-world action and nothing that varies between retries of it. The reliable trio is the eventType, the primary linked object identifier, and the action’s timestamp captured once at completion. Do not seed on wall-clock “now” read inside the emitter, or each retry invents a new id. If two legitimately different actions could share that trio at second precision, add sub-second precision or an explicit action sequence number so they stay distinct.

# Is a JSON Lines file really enough for a preservation ledger?

For a single-node store it is a sound append-only substrate: records are never mutated, writes are flushed before success is claimed, and read-back verification catches truncation. At scale you replace the file with a tamper-evident, hash-chained store, but the emission contract — deterministic id, idempotent append, emit-then-verify — stays identical. The model and emitter here are the durable part; the storage backend is swappable.

# What happens if the process crashes between appending and verifying?

Nothing is lost and nothing is duplicated. On restart the action re-runs, rebuilds the event with the same deterministic identifier, and the emitter’s contains check finds the already-written line and commits without appending a second copy. If the crash happened before the append flushed, the re-run simply writes the line for the first time. Either way the ledger converges to exactly one event per action.

# Related

• PREMIS Metadata Mapping — how emitted events are folded into object-level provenance and the controlled vocabulary they share.
• Designing an Immutable Audit Ledger — scaling the append-only store into a tamper-evident, hash-chained record.
• Scheduling Fixity Re-Validation Jobs — the recurring actions that emit fixityCheck events through this exact path.