Add drift fingerprint + incremental mint/supersede publishing

src/pubmate/__init__.py

@@ -1,5 +1,13 @@
 from pubmate.defining import DEFAULT_LICENSE, DefiningNanopubBuilder
+from pubmate.fingerprint import (
+    FP_SCHEME,
+    IdentityFields,
+    canonical_assertion,
+    fingerprint_term,
+    identity_fields,
+)
 from pubmate.idmap import IdMap, IdMapEntry
+from pubmate.incremental import IncrementalResult, publish_incremental
 from pubmate.introduction import build_introduction
 from pubmate.migrate import MigrationResult, MintedSupersession, migrate_terms
 from pubmate.mint import IdentifierGenerator
@@ -11,10 +19,13 @@
 
 __all__ = [
     "DEFAULT_LICENSE",
+    "FP_SCHEME",
     "DefiningNanopubBuilder",
     "IdMap",
     "IdMapEntry",
     "IdentifierGenerator",
+    "IdentityFields",
+    "IncrementalResult",
     "MintBatch",
     "MintedTerm",
     "MigrationResult",
@@ -25,8 +36,12 @@
     "SupersessionBuilder",
     "TermInput",
     "build_introduction",
+    "canonical_assertion",
+    "fingerprint_term",
+    "identity_fields",
     "migrate_terms",
     "order_terms",
+    "publish_incremental",
     "referenced_terms",
     "serialize_nanopub",
     "sign_and_publish",

src/pubmate/cli/mint_publish.py

@@ -7,7 +7,9 @@
 from pubmate.cli._signing import resolve_signing
 from pubmate.defining import DefiningNanopubBuilder
 from pubmate.idmap import IdMap
+from pubmate.incremental import publish_incremental
 from pubmate.minting import SequentialMinter, term_input_from_assertion
+from pubmate.supersede import SupersessionBuilder
 from pubmate.utils import serialize_nanopub
 
 logging.basicConfig(level=logging.INFO, format="%(asctime)s - %(name)s - %(levelname)s - %(message)s")
@@ -55,14 +57,18 @@ def cli(
     dry_run: bool,
     pattern: str,
 ) -> None:
-    """Sequentially mint defining nanopubs from per-term assertions and publish them.
-
-    Each assertion is re-keyed onto the artifact-code placeholder, signed (which
-    lands the code on the term's thing URI), and -- unless --dry-run -- published.
-    Minted nanopubs are written to --output-dir as <old_id-stem>.trig and the
-    old_id -> thing_uri/np_uri mapping is written/merged into --id-map-file.
-
-    Inter-term links (forward refs/cycles) are intentionally left to a later
+    """Incrementally mint/supersede defining nanopubs from per-term assertions.
+
+    Each assertion is re-keyed onto the artifact-code placeholder. Per term, this
+    compares its identity fingerprint against the one recorded in --id-map-file:
+    a new term is minted, an unchanged term is skipped, and a *drifted* term
+    (content or wrapper changed) is superseded -- re-stated against its existing
+    thing URI in a nanopub that supersedes the recorded one, keeping the term's
+    identity. Published nanopubs (unless --dry-run) are written to --output-dir as
+    <artifact-code>.trig and the updated old_id -> thing_uri/np_uri/fingerprint
+    map is written to --id-map-file.
+
+    Inter-term links (forward refs/cycles) are intentionally left to the migration
     superseding pass (see the migration tooling); this mints the assertions as
     given.
     """
@@ -82,6 +88,10 @@ def cli(
         namespace, profile=signing.profile, test_server=signing.test_server,
         nanopub_types=nanopub_types, template=template,
     )
+    supersession_builder = SupersessionBuilder(
+        profile=signing.profile, test_server=signing.test_server,
+        license=builder.license, nanopub_types=nanopub_types, template=template,
+    )
 
     files = sorted(assertion_folder.glob(pattern))
     if not files:
@@ -105,26 +115,33 @@ def cli(
     existing = IdMap.from_tsv(id_map_file.read_text(encoding="utf-8")) if id_map_file and id_map_file.exists() else IdMap()
 
     minter = SequentialMinter(builder, default_suggester_orcid=default_suggester)
-    batch = minter.mint_all(
+    result = publish_incremental(
         terms,
+        minter=minter,
+        supersession_builder=supersession_builder,
+        existing=existing,
         dry_run=dry_run,
-        already_minted=existing.np_uri_map,
     )
 
-    # Write each nanopub as <artifact-code>.trig (the thing/np code under scheme A).
+    # Write each minted/superseding nanopub as <artifact-code>.trig (its own code:
+    # for a defining nanopub that equals the thing code, for a supersession its own).
     output_dir.mkdir(parents=True, exist_ok=True)
-    for minted in batch.terms:
-        code = minted.thing_uri.removeprefix(namespace)
-        (output_dir / f"{code}.trig").write_text(serialize_nanopub(minted.nanopub), encoding="utf-8")
+    published = [(m.np_uri, m.nanopub) for m in result.minted.terms]
+    published += [(s.np_uri, s.nanopub) for s in result.superseded]
+    for np_uri, np in published:
+        code = np_uri.rsplit("/", 1)[-1]
+        (output_dir / f"{code}.trig").write_text(serialize_nanopub(np), encoding="utf-8")
 
     if id_map_file is not None:
-        merged = IdMap(list(existing))
-        merged.merge(IdMap.from_batch(batch), overwrite=True)
         id_map_file.parent.mkdir(parents=True, exist_ok=True)
-        merged.write_tsv(id_map_file)
-        logger.info("Wrote id-map (%d entries) -> %s", len(merged), id_map_file)
+        result.id_map.write_tsv(id_map_file)
+        logger.info("Wrote id-map (%d entries) -> %s", len(result.id_map), id_map_file)
 
-    logger.info("Minted %d new term(s)%s -> %s", len(batch.terms), " (dry-run)" if dry_run else "", output_dir)
+    logger.info(
+        "Minted %d, superseded %d, skipped %d term(s)%s -> %s",
+        len(result.minted.terms), len(result.superseded), len(result.skipped),
+        " (dry-run)" if dry_run else "", output_dir,
+    )
 
 
 if __name__ == "__main__":

src/pubmate/fingerprint.py

@@ -0,0 +1,149 @@
+"""Semantic drift fingerprint for defining nanopublications.
+
+A defining nanopub's trusty artifact code is a hash of the *whole* nanopub, so it
+cannot answer "does the already-published version need re-issuing?": it also moves
+with the build timestamp, the signing key and the toolchain, and it only exists
+*after* signing. This module computes a stable, key-independent fingerprint over
+just the **identity-defining** inputs, so a caller can tell an unchanged term from
+one whose content or wrapper has drifted -- and, on drift, supersede rather than
+mint a fresh (differently-identified) nanopub.
+
+Three tiers, and where each goes:
+
+* **No-op** -- build timestamp, signature, blank-node ids, triple order,
+  serialization prefixes. Excluded: the assertion is URDNA2015-canonicalized
+  (:func:`canonical_assertion`), and nothing time/signature-derived enters.
+* **Identity-defining** -- the assertion itself (which already carries
+  ``dcterms:isPartOf`` when minted with ``part_of``), the suggester and
+  ``derived_from`` provenance, and the pubinfo the vocabulary commits to:
+  ``dct:license``, ``npx:introduces``, ``npx:hasNanopubType``,
+  ``nt:wasCreatedFromTemplate``. This is the fingerprint domain.
+* **Build-provenance** -- signing key, ``nanopub``/pubmate version, trusty
+  algorithm. Deliberately **not** hashed here (a key rotation must not read as
+  content drift); record it beside the fingerprint and gate re-issue on policy.
+
+The fingerprint is computed on the *placeholder* form of the assertion (subject =
+``namespace + ~~~ARTIFACTCODE~~~``), i.e. the graph the builder holds before
+signing, so the thing URI -- which contains the code, itself a hash of the whole
+nanopub -- never feeds its own fingerprint.
+
+Keep :func:`identity_fields` in lockstep with
+:meth:`~pubmate.defining.DefiningNanopubBuilder.build`: if a future change adds an
+identity-bearing pubinfo triple there, mirror it here and bump :data:`FP_SCHEME`.
+"""
+
+from __future__ import annotations
+
+import hashlib
+import json
+from dataclasses import dataclass
+from typing import Any, Optional
+
+import rdflib
+from rdflib.compare import to_canonical_graph
+
+from pubmate._nanopub_build import UNSET as _UNSET
+from pubmate.defining import DefiningNanopubBuilder
+from pubmate.minting import TermInput
+
+#: Fingerprint scheme tag, embedded in every digest. Bump when the domain or the
+#: canonicalization changes, so digests from an older scheme are recomputed
+#: rather than silently compared across incompatible definitions.
+FP_SCHEME = "pubmate-fp-1"
+
+
+def canonical_assertion(graph: rdflib.Graph) -> str:
+    """Canonicalize ``graph`` (URDNA2015) to sorted N-Triples.
+
+    Stable across blank-node labels, triple order and serialization prefixes --
+    the no-op tier -- so cosmetic RDF churn does not move the fingerprint.
+    """
+    canonical = to_canonical_graph(graph)
+    lines = canonical.serialize(format="nt").splitlines()
+    return "\n".join(sorted(line for line in lines if line.strip()))
+
+
+@dataclass(frozen=True)
+class IdentityFields:
+    """The identity-defining inputs a defining nanopub commits to.
+
+    A change to any field means the published nanopub is materially stale and
+    should be re-issued by *superseding*. The signing key and toolchain are
+    intentionally absent (see the module docstring)."""
+
+    assertion: str
+    suggester: str
+    derived_from: str
+    license: str
+    introduces: str
+    nanopub_types: tuple[str, ...]
+    template: str
+
+    def digest(self) -> str:
+        """Hex SHA-256 over the canonical JSON of these fields plus the scheme."""
+        payload = {
+            "scheme": FP_SCHEME,
+            "assertion": self.assertion,
+            "suggester": self.suggester,
+            "derived_from": self.derived_from,
+            "license": self.license,
+            "introduces": self.introduces,
+            "nanopub_types": list(self.nanopub_types),
+            "template": self.template,
+        }
+        blob = json.dumps(payload, sort_keys=True, ensure_ascii=False)
+        return hashlib.sha256(blob.encode("utf-8")).hexdigest()
+
+
+def identity_fields(
+    term: TermInput,
+    builder: DefiningNanopubBuilder,
+    *,
+    default_suggester: Optional[str] = None,
+    license: Any = _UNSET,
+    introduces: Any = _UNSET,
+) -> IdentityFields:
+    """Extract ``term``'s identity-defining fields as ``builder`` would emit them.
+
+    Mirrors :meth:`DefiningNanopubBuilder.build` and
+    :meth:`SequentialMinter.mint` defaulting so the fingerprint tracks exactly
+    what gets signed: ``license`` falls back to the builder's license,
+    ``introduces`` to the placeholder thing URI, and the suggester to
+    ``default_suggester`` when the term carries none. ``part_of`` is already in
+    ``term.assertion`` (added by ``term_input_from_assertion``), so it is covered
+    by the assertion hash rather than a field here.
+    """
+    effective_license = builder.license if license is _UNSET else license
+    effective_introduces = builder.thing_uri if introduces is _UNSET else introduces
+    return IdentityFields(
+        assertion=canonical_assertion(term.assertion),
+        suggester=term.suggester_orcid or default_suggester or "",
+        derived_from=term.derived_from or "",
+        license=effective_license or "",
+        introduces="" if effective_introduces is None else str(effective_introduces),
+        nanopub_types=tuple(builder.nanopub_types),
+        template=builder.template or "",
+    )
+
+
+def fingerprint_term(
+    term: TermInput,
+    builder: DefiningNanopubBuilder,
+    *,
+    default_suggester: Optional[str] = None,
+    license: Any = _UNSET,
+    introduces: Any = _UNSET,
+) -> str:
+    """Hex SHA-256 drift fingerprint for ``term`` as built by ``builder``.
+
+    Convenience wrapper over :func:`identity_fields`; the same defaulting rules
+    apply. Two terms share a fingerprint iff they would sign to nanopubs that are
+    identical in every identity-defining respect (ignoring timestamp, signature,
+    key and blank-node/serialization noise)."""
+    return identity_fields(
+        term,
+        builder,
+        default_suggester=default_suggester,
+        license=license,
+        introduces=introduces,
+    ).digest()

src/pubmate/idmap.py

@@ -2,12 +2,16 @@
 
 Records, per term, the mapping from its old/local identifier to the new
 nanopub-based identifiers minted for it: the term's thing URI (its trusty
-artifact-code URI) and the URI of its defining nanopub.
+artifact-code URI) and the URI of its defining nanopub, plus an optional
+drift ``fingerprint`` (see :mod:`pubmate.fingerprint`) of the identity-defining
+inputs the published nanopub was built from -- so a later run can tell an
+unchanged term from one that has drifted and needs superseding.
 
 The map is meant to be kept permanently and grown incrementally, so old
 identifiers stay resolvable and re-runs can append without losing prior entries.
 It round-trips to a tab-separated file (a superset of a simple redirect table)
-and to JSON.
+and to JSON. The TSV gained a fourth ``fingerprint`` column; 3-column files
+written by older versions still read (their fingerprint is empty).
 """
 
 from __future__ import annotations
@@ -19,16 +23,20 @@
 
 from pubmate.minting import MintBatch
 
-_TSV_HEADER = ("old_id", "thing_uri", "np_uri")
+_TSV_HEADER = ("old_id", "thing_uri", "np_uri", "fingerprint")
 
 
 @dataclass(frozen=True)
 class IdMapEntry:
-    """One term's old identifier and its new nanopub-based identifiers."""
+    """One term's old identifier and its new nanopub-based identifiers.
+
+    ``fingerprint`` is the drift fingerprint of the identity-defining inputs the
+    nanopub was built from (empty when unknown, e.g. legacy 3-column rows)."""
 
     old_id: str
     thing_uri: str
     np_uri: str
+    fingerprint: str = ""
 
 
 class IdMap:
@@ -60,13 +68,21 @@ def merge(self, other: "IdMap", *, overwrite: bool = False) -> None:
             self.add(entry, overwrite=overwrite)
 
     @classmethod
-    def from_batch(cls, batch: MintBatch) -> "IdMap":
+    def from_batch(cls, batch: MintBatch, *, fingerprints: Optional[Dict[str, str]] = None) -> "IdMap":
         """Build a map from a :class:`~pubmate.minting.MintBatch`.
 
-        The minter's ``term_id`` is used as the old identifier.
+        The minter's ``term_id`` is used as the old identifier. ``fingerprints``,
+        if given, supplies each term's drift fingerprint keyed by ``term_id``
+        (missing terms get an empty fingerprint).
         """
+        fingerprints = fingerprints or {}
         return cls(
-            IdMapEntry(old_id=t.term_id, thing_uri=t.thing_uri, np_uri=t.np_uri)
+            IdMapEntry(
+                old_id=t.term_id,
+                thing_uri=t.thing_uri,
+                np_uri=t.np_uri,
+                fingerprint=fingerprints.get(t.term_id, ""),
+            )
             for t in batch.terms
         )
 
@@ -94,14 +110,19 @@ def np_uri_map(self) -> Dict[str, str]:
         """``old_id -> nanopub URI``."""
         return {e.old_id: e.np_uri for e in self}
 
+    @property
+    def fingerprint_map(self) -> Dict[str, str]:
+        """``old_id -> drift fingerprint`` (empty string when unknown)."""
+        return {e.old_id: e.fingerprint for e in self}
+
     def _sorted(self) -> List[IdMapEntry]:
         return sorted(self._entries.values(), key=lambda e: e.old_id)
 
     # -- serialization ----------------------------------------------------
 
     def to_tsv(self) -> str:
         lines = ["\t".join(_TSV_HEADER)]
-        lines += ["\t".join((e.old_id, e.thing_uri, e.np_uri)) for e in self._sorted()]
+        lines += ["\t".join((e.old_id, e.thing_uri, e.np_uri, e.fingerprint)) for e in self._sorted()]
         return "\n".join(lines) + "\n"
 
     @classmethod
@@ -110,10 +131,12 @@ def from_tsv(cls, text: str) -> "IdMap":
         lines = [ln for ln in text.splitlines() if ln.strip()]
         for line in lines:
             fields = line.split("\t")
-            if tuple(fields) == _TSV_HEADER:
-                continue
-            if len(fields) != 3:
-                raise ValueError(f"expected 3 tab-separated fields, got {len(fields)}: {line!r}")
+            if fields[0] == _TSV_HEADER[0]:
+                continue  # header row (3- or 4-column)
+            if len(fields) == 3:  # legacy row, no fingerprint
+                fields = (*fields, "")
+            if len(fields) != 4:
+                raise ValueError(f"expected 3 or 4 tab-separated fields, got {len(fields)}: {line!r}")
             id_map.add(IdMapEntry(*fields))
         return id_map