Within the broader rights metadata mapping and licensing automation pipeline, this stage is the decision engine that turns a validated asset record into a single, defensible copyright verdict and a routing destination. It sits after ingestion and before publication: every object that a digitization campaign produces must pass through a compliance gate that decides whether the institution may open the asset, must clear rights first, must hold it under embargo, or lacks the data to decide at all. This page specifies the record contract that feeds the engine, the jurisdictional term arithmetic that is easy to get subtly wrong, the routing map that carries the verdict downstream, and the verification harness a Python automation engineer runs before a batch of forty thousand records goes live.

The problem is narrow and unforgiving. A human registrar can read a catalogue card, weigh a donor agreement, and reason about a creation date; a pipeline cannot improvise. It needs an explicit rule for the United States rolling publication cutoff, an explicit rule for the European Union life + 70 term, and — most importantly — an explicit, fail-safe behaviour for the records where the data simply is not there. A silent default that guesses “public domain” is a compliance incident waiting to happen. The engine described here never guesses: incomplete temporal data resolves to an AMBIGUOUS verdict that routes to human review, never to open publication.

Workflow Context

Museum digital asset pipelines routinely ingest thousands of records whose rights metadata is fragmented across legacy CMS fields, donor contracts, and digitization logs. Manual verification of each record introduces unacceptable latency and, worse, inconsistent judgement calls that fail an audit. The role that owns this stage is usually a Python automation engineer working alongside a collections manager who defines institutional policy; the engineer encodes that policy as deterministic rules so that the same input always yields the same verdict.

This stage is downstream of the ingestion and normalization work described in the parent pipeline and is a sibling to three other rights subsystems. Assets that arrive already carrying an open license are handed off to routing Creative Commons licenses; records that resolve as still-protected but destined for eventual release flow into implementing embargo workflows; and the sensitivity of the US publication cutoff — the single date that flips a work into the public domain — is examined in depth under threshold tuning for public domain. The engine on this page is the common decision point all three depend on.

The design goal is a single-pass, idempotent evaluation. Each record is hashed on its rights-relevant fields, evaluated exactly once, logged, and dispatched to precisely one queue. Re-running a batch produces identical hashes and identical verdicts, so a retried run after a crash never double-publishes an asset or contradicts its earlier decision.

Prerequisites

Before wiring the status engine, confirm the following are in place:

  • Python 3.9+ for the PEP 604 | union syntax, tuple[...] generics, and the asyncio features used in the batch processor.
  • pydantic (v2) for the record contract — this page uses the v2 API (field_validator, model_dump, ConfigDict); see the pydantic v2 documentation for the validator signatures.
  • httpx (0.27+) as the async client for dispatching verdicts to downstream queue endpoints, with connection-pool limits and per-request timeouts.
  • Normalized input records. The engine assumes rights fields have already been resolved to canonical URIs upstream; the mapping that produces them is the subject of the child guide mapping RightsStatements.org to collection fields.
  • A jurisdictional policy source. Statutory term tables must be externalized, not hard-coded per deployment — the authoritative US durations are published by the U.S. Copyright Office.
  • Four downstream destinations reachable at dispatch time: an open-publication queue, a rights-clearance queue, a curator-review queue, and a fallback/error queue.
  • Interchange standards for the surviving payload. Verdicts attach to records that must remain conformant to the LIDO schema lido:rights element and the IIIF Presentation API 3.0 rights property.

Schema & Verdict Reference

The record contract is the single point of truth for what the engine may reason about. Raw collection exports rarely conform to interoperability standards, so a strict Pydantic v2 model coerces types, bounds year values to a plausible range, and rejects unsupported jurisdiction codes before any rule runs. The verdict model, in turn, fixes the closed set of statuses the engine may emit and pairs each with its routing destination and the basis string that makes the decision auditable.

Field Type Constraint Purpose
object_id str required Idempotency and audit key
creation_year int | None 1000 ≤ y ≤ now+10 US publication-term input
creator_death_year int | None 1000 ≤ y ≤ now+10 EU life+70 input
jurisdiction str one of US EU UK CA Selects the rule branch
rights_statement_uri str | None RightsStatements.org URI Explicit assertion overrides computation
donor_restrictions str | None free text Highest-priority restriction gate

The status vocabulary is closed and maps one-to-one onto a routing destination:

Status Meaning Routing destination
PUBLIC_DOMAIN Term lapsed or explicit no-known-copyright open_publish
COPYRIGHTED Statutory term still active rights_clearance
RESTRICTED Active donor or contractual restriction curator_review
AMBIGUOUS Insufficient data to decide fallback_chain

Step-by-Step Implementation

Production implementations require asynchronous batch processing to handle high-throughput DAM integrations without blocking on downstream dispatch. The stack below uses pydantic for schema enforcement, asyncio with a semaphore for controlled concurrency, and explicit routing logic. The engine is split into three responsibilities — the record contract, the rule evaluation, and the async pipeline — each of which is presented in turn and then assembled in main().

1. Define and validate the record contract

The ingestion boundary maps legacy fields to standard properties and enforces strict schema limits during deserialization. Invalid date formats or unsupported jurisdiction codes trigger immediate rejection, so a malformed record can never corrupt the decision graph.

python
import asyncio
import hashlib
import logging
from datetime import date, datetime, timezone
from typing import Optional, List, Any, Literal
from pydantic import BaseModel, Field, field_validator, ValidationError
import httpx

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s | %(levelname)s | %(message)s"
)
logger = logging.getLogger(__name__)

class AssetRecord(BaseModel):
    object_id: str
    title: str
    creator_death_year: Optional[int] = None
    creation_year: Optional[int] = None
    jurisdiction: str = Field(default="US", pattern="^(US|EU|UK|CA)$")
    rights_statement_uri: Optional[str] = None
    donor_restrictions: Optional[str] = None
    iiif_manifest_url: Optional[str] = None

    @field_validator("creation_year", "creator_death_year")
    @classmethod
    def validate_year_range(cls, v: Optional[int]) -> Optional[int]:
        if v is not None and not (1000 <= v <= date.today().year + 10):
            raise ValueError("Year must be within plausible historical range")
        return v

    def compute_payload_hash(self) -> str:
        raw = f"{self.object_id}|{self.creation_year}|{self.creator_death_year}|{self.jurisdiction}"
        return hashlib.sha256(raw.encode()).hexdigest()

class CopyrightStatus(BaseModel):
    object_id: str
    status: Literal["PUBLIC_DOMAIN", "COPYRIGHTED", "AMBIGUOUS", "RESTRICTED"]
    basis: str
    routing_destination: str
    payload_hash: str
    validated_at: datetime = Field(default_factory=lambda: datetime.now(timezone.utc))

The payload_hash deliberately covers only the rights-relevant fields. Two records that differ solely in, say, their title produce the same hash and therefore the same idempotency key — cosmetic edits never force a re-decision.

2. Encode the jurisdictional rules

Status determination relies on configurable jurisdictional rulesets. US federal law applies a rolling publication cutoff — currently 95 years, advancing every January 1 — while European directives calculate duration from creator lifespan plus seventy years. The engine evaluates RightsStatements.org URIs against institutional policy before falling back to computation, and treats missing temporal data as a reason to defer, never to publish.

python
class CopyrightEngine:
    US_COPYRIGHT_TERM = 95  # years from publication for pre-1978 US works
    EU_LIFE_PLUS = 70

    @classmethod
    def evaluate(cls, record: AssetRecord) -> CopyrightStatus:
        status, basis, destination = cls._apply_rules(record)
        return CopyrightStatus(
            object_id=record.object_id,
            status=status,
            basis=basis,
            routing_destination=destination,
            payload_hash=record.compute_payload_hash()
        )

    @classmethod
    def _apply_rules(cls, record: AssetRecord) -> tuple[str, str, str]:
        current_year = date.today().year

        if record.donor_restrictions:
            return "RESTRICTED", "Donor agreement active", "curator_review"

        if record.rights_statement_uri and "NoKnownCopyright" in record.rights_statement_uri:
            return "PUBLIC_DOMAIN", "RightsStatements.org assertion", "open_publish"

        if record.jurisdiction == "US" and record.creation_year:
            # The cutoff is not fixed: it advances every January 1 as the
            # 95-year term lapses. Compute it from the current year.
            latest_pd_year = current_year - cls.US_COPYRIGHT_TERM - 1
            if record.creation_year <= latest_pd_year:
                return "PUBLIC_DOMAIN", f"US 95-year term (published ≤ {latest_pd_year})", "open_publish"
            return "COPYRIGHTED", "US 95-year term active", "rights_clearance"

        if record.jurisdiction == "EU" and record.creator_death_year:
            if record.creator_death_year + cls.EU_LIFE_PLUS < current_year:
                return "PUBLIC_DOMAIN", "EU life+70 expiry", "open_publish"

        if record.creation_year and record.creator_death_year:
            return "COPYRIGHTED", "Active statutory term", "rights_clearance"

        return "AMBIGUOUS", "Insufficient temporal data", "fallback_chain"

Rule order encodes priority. The donor-restriction gate runs first because a contractual hold overrides any term calculation — a work can be in the public domain and still be legally embargoed by a bequest. The explicit NoKnownCopyright assertion runs next because a curator’s reviewed judgement outranks a computed guess. Only then does term arithmetic apply, and the final unconditional return guarantees that any record slipping past every branch lands on the fail-safe AMBIGUOUS verdict rather than an implicit None.

3. Drive the batch through an async pipeline

The pipeline caps concurrency with a semaphore so a large batch never overruns the downstream DAM API, evaluates each record, and dispatches its verdict. asyncio.gather(..., return_exceptions=True) ensures that one poisoned record cannot cancel its siblings — failures return as values the caller inspects by type.

python
class RightsPipeline:
    def __init__(self, batch_size: int = 50, max_concurrency: int = 10):
        self.batch_size = batch_size
        self.semaphore = asyncio.Semaphore(max_concurrency)
        self.client = httpx.AsyncClient(timeout=15.0)

    async def process_batch(self, records: List[AssetRecord]) -> List[Any]:
        # return_exceptions=True means the result list may hold CopyrightStatus
        # objects or, for any unhandled error, the raised exception — callers
        # must check the type (see main()).
        tasks = [self._process_single(r) for r in records]
        return await asyncio.gather(*tasks, return_exceptions=True)

    async def _process_single(self, record: AssetRecord) -> CopyrightStatus:
        async with self.semaphore:
            try:
                status = CopyrightEngine.evaluate(record)
                logger.info(f"Evaluated {record.object_id} -> {status.status}")
                await self._dispatch(status)
                return status
            except ValidationError as e:
                logger.error(f"Schema validation failed for {record.object_id}: {e}")
                return CopyrightStatus(
                    object_id=record.object_id,
                    status="AMBIGUOUS",
                    basis="Invalid input schema",
                    routing_destination="error_queue",
                    payload_hash=record.compute_payload_hash()
                )

    async def _dispatch(self, status: CopyrightStatus) -> None:
        # Simulate downstream API call or queue push
        endpoint = f"https://workflow.internal/api/v1/rights/{status.routing_destination}"
        try:
            await self.client.post(endpoint, json=status.model_dump(mode="json"))
        except httpx.HTTPError as e:
            logger.warning(f"Dispatch failed for {status.object_id}: {e}")

    async def close(self) -> None:
        await self.client.aclose()

async def main() -> None:
    sample_records = [
        AssetRecord(object_id="OBJ-001", title="Landscape Study", creation_year=1890, jurisdiction="US"),
        AssetRecord(object_id="OBJ-002", title="Modern Sculpture", creator_death_year=1950, jurisdiction="EU"),
        AssetRecord(object_id="OBJ-003", title="Restricted Archive", donor_restrictions="Active until 2030", jurisdiction="US"),
    ]

    pipeline = RightsPipeline(batch_size=50, max_concurrency=5)
    try:
        results = await pipeline.process_batch(sample_records)
        for r in results:
            if isinstance(r, CopyrightStatus):
                logger.info(f"Final: {r.object_id} | {r.status} | {r.routing_destination}")
    finally:
        await pipeline.close()

if __name__ == "__main__":
    asyncio.run(main())

Note the edge-case handling across input variants: a CSV export that carries year columns as strings is coerced by Pydantic before _apply_rules ever sees them; an XML payload with a stray unsupported jurisdiction code is rejected at construction by the pattern constraint; and an API record missing both year fields flows deterministically to fallback_chain instead of raising.

Decision Flow

The rule precedence above is easiest to read as a branch diagram: donor restrictions short-circuit first, then explicit assertions, then jurisdiction-specific term arithmetic, with every unresolved path converging on the fail-safe fallback.

Copyright verdict decision tree — the first matching gate wins Top to bottom, rules are applied in the same priority order the engine uses. Gate one: donor restrictions short-circuit to RESTRICTED / curator_review. Gate two: an explicit NoKnownCopyright URI short-circuits to PUBLIC_DOMAIN / open_publish. Gate three splits on jurisdiction — US works route on the rolling 95-year publication cutoff to either PUBLIC_DOMAIN / open_publish or COPYRIGHTED / rights_clearance; EU works route on the life-plus-70 test to PUBLIC_DOMAIN / open_publish or the fail-safe AMBIGUOUS / fallback_chain; every other jurisdiction and every unresolved path lands on the fail-safe AMBIGUOUS / fallback_chain. Rule precedence — the first matching gate wins; unresolved paths fail safe AssetRecord Donor restrictions? NoKnownCopyright URI? Jurisdiction? Published ≤ cutoff year? Death + 70 before now? RESTRICTED → curator_review PUBLIC_DOMAIN → open_publish AMBIGUOUS → fallback_chain PUBLIC_DOMAIN → open_publish COPYRIGHTED → rights_clearance PUBLIC_DOMAIN → open_publish AMBIGUOUS → fallback_chain yes no yes no US EU other yes no yes no

Rights and Access Routing

The verdict is not the end of the story — it is a routing signal that determines the access tier every downstream consumer honours. Public domain assets route to open-access publication and surface in public portals and IIIF viewers with a permissive rights property. COPYRIGHTED assets divert to a rights-clearance queue and stay non-resolvable to anonymous users until a license is secured; where that license already exists as a Creative Commons grant, the clearance step hands off to routing Creative Commons licenses. RESTRICTED assets — those under a donor or contractual hold — enter curator review and, when the hold has a defined end date, are scheduled through implementing embargo workflows so they release automatically when the term lapses. AMBIGUOUS assets never become publicly resolvable; they sit in the fallback queue for a registrar to supply the missing creation or death year.

This routing map is the reason the status vocabulary is closed. A downstream access-control layer can switch on exactly four values, and a new status can never appear without a corresponding routing rule being added deliberately.

Verification and Testing

Because the engine’s correctness rests on two boundary details — the rolling US cutoff and the strict EU inequality — the test suite must pin the exact year arithmetic rather than trust it. The following assert-based cases exercise each branch, including the fail-safe path, and run under pytest -q or as a plain script.

python
from datetime import date

def test_us_rolling_cutoff():
    # A US work published exactly at the rolling boundary is public domain.
    boundary = date.today().year - CopyrightEngine.US_COPYRIGHT_TERM - 1
    rec = AssetRecord(object_id="T1", title="Boundary", creation_year=boundary, jurisdiction="US")
    assert CopyrightEngine.evaluate(rec).status == "PUBLIC_DOMAIN"
    # One year later is still under copyright.
    rec2 = AssetRecord(object_id="T2", title="After", creation_year=boundary + 1, jurisdiction="US")
    assert CopyrightEngine.evaluate(rec2).status == "COPYRIGHTED"

def test_eu_strict_inequality():
    # life+70 uses strict '<': a work in its final year of protection is NOT public domain.
    final_year = date.today().year - CopyrightEngine.EU_LIFE_PLUS
    rec = AssetRecord(object_id="T3", title="Final year", creator_death_year=final_year, jurisdiction="EU")
    assert CopyrightEngine.evaluate(rec).status != "PUBLIC_DOMAIN"

def test_missing_data_is_fail_safe():
    rec = AssetRecord(object_id="T4", title="No dates", jurisdiction="UK")
    result = CopyrightEngine.evaluate(rec)
    assert result.status == "AMBIGUOUS"
    assert result.routing_destination == "fallback_chain"

def test_donor_restriction_overrides_term():
    # Public-domain-era work under an active donor hold must still be RESTRICTED.
    rec = AssetRecord(object_id="T5", title="Held bequest", creation_year=1850,
                      jurisdiction="US", donor_restrictions="Sealed until 2040")
    assert CopyrightEngine.evaluate(rec).status == "RESTRICTED"

All four cases should pass. For a live dry run, call process_batch against a sample of real identifiers and tally the routing destinations without letting _dispatch reach production endpoints — a pre-flight that surfaces a drifted export or a policy misconfiguration before it publishes anything.

In This Section

One focused guide goes deeper on the mapping work the engine depends on:

  • Mapping RightsStatements.org to Collection Fields — resolving the ValueError, silent-truncation, and namespace-collision failures that occur when RightsStatements.org v1.0 URIs are written into legacy VARCHAR-bounded CMS columns, so the URIs the engine reads are canonical.

FAQ

Why is the US public domain cutoff computed instead of hard-coded?

Because it is not a fixed year — it advances every January 1 as another year of works ages out of the 95-year term. Hard-coding “published before 1929” bakes in a value that is wrong twelve months later and quietly withholds newly-public works from the open portal. The engine computes current_year - 95 - 1 at evaluation time so the boundary is always correct. The sensitivity of this single date is explored further under threshold tuning for public domain.

Why does the EU rule use a strict < rather than <=?

The life + 70 term protects a work through the full seventieth year after the creator’s death; it lapses on the following January 1. A <= comparison would declare the work public domain during its final protected year, exposing the institution to an infringement claim. The strict < is a deliberate one-year safety margin on the side of caution.

What happens when a record has no creation or death year?

It resolves to AMBIGUOUS and routes to fallback_chain for a registrar to complete. This is the single most important behaviour in the engine: missing data never produces a PUBLIC_DOMAIN verdict. A false “in copyright” call merely delays access; a false “public domain” call is an unauthorized disclosure, so the fail-safe always defers rather than guesses.

The donor gate runs first and wins. A work whose statutory term lapsed a century ago can still be under a binding bequest condition, so donor_restrictions short-circuits to RESTRICTED before any term arithmetic runs. Copyright expiry and contractual access control are independent constraints, and the more restrictive one governs.

Why hash only some fields for idempotency?

The payload_hash covers object_id, both year fields, and jurisdiction — the inputs that can actually change a verdict. Excluding title, manifest URL, and other descriptive fields means a cosmetic metadata edit does not invalidate a prior decision or trigger a needless re-publish, while any change to a rights-relevant field correctly produces a new hash and a fresh evaluation.