Normalize KEGG compound xref annotations without changing identities

docs/reviews/kegg-xref-annotations-1259.md

@@ -0,0 +1,41 @@
+# KEGG compound annotation syntax (#1259)
+
+The shared chemical reader now emits `kegg.compound:Cnnnnn` instead of the exact
+historical `kegg.compound:cpd:Cnnnnn` spelling in node xref annotations. The rule
+requires uppercase `C` and exactly five ASCII digits. It deduplicates and sorts
+emitted xrefs, including optional ingredient-bundle enrichment. Other KEGG
+subspaces, identifiers and malformed forms are unchanged. KEGG documents `cpd`
+as the compound database abbreviation and `C` plus five digits as the entry
+format. [Official KEGG API manual](https://www.kegg.jp/kegg/rest/keggapi.html).
+
+This is an annotation-only repair. Original unified SSSOM rows, their provenance
+and multiplicity, literal identity lookup keys, query behavior, supported MIM
+export and immutable pin are unchanged. Canonical and historical alias queries
+can therefore still have different historical targets; this change neither
+endorses those equivalences nor resolves their scientific conflicts.
+
+The accepted unified artifact with SHA-256
+`67c48e1bf6bed1f1fef03a0da1d7d1b56af9fc72374dddd703c222fb36df3cd4`
+contains 15,370 affected rows. Of these, 15,355 have exact full-row canonical
+counterparts after subject normalization, while 15 do not. Blind normalization
+before lookup indexing would change five canonical choices; routing aliases
+through canonical choices would change 35 historical alias-query choices and
+alias-only fallback would newly resolve two canonical keys. None of those
+lookup changes is part of this fix.
+
+The current native ChEBI JSON and node TSV have no such double-prefixed compound
+xrefs; each has 17,446 canonical compound xref values. The primary-mapping
+importer already removes the `cpd:` abbreviation before prefixing new entries.
+The malformed retained rows can be reproduced by historical seed paths; no
+exporter or physical SSSOM cleanup is claimed here. Such cleanup needs its own
+reviewed, stable identity-precedence contract, because existing exporters sort
+by target and simple row relocation cannot preserve canonical precedence.
+
+The full original-row inventory, counterpart ordinals, five canonical conflicts,
+35 alias-query differences and native-byte evidence are retained under
+`data/issue1224-quarantine-20260929.xjuS9H/issue1259-diagnosis.7T9Jn8/`.
+Hermetic tests cover conflicting rows in either order, reloads, alias-only
+unresolved canonical queries, weak-relation exclusion and actual MediaDive
+producer enrichment through unified, legacy and embedded identity routes.
+The chemical-mapping skill and reviewed-MIM runbook governed this bounded scope;
+no MIM regeneration or mapping promotion is required for this code change.

kg_microbe/utils/chemical_mapping_utils.py

@@ -21,6 +21,7 @@
 
 import pandas as pd
 
+from kg_microbe.transform_utils.constants import XREF_COLUMN
 from kg_microbe.utils.cas import invalid_cas_identifier
 from kg_microbe.utils.ingredient_identity import (
     ingredient_authority_label,
@@ -755,9 +756,19 @@ def get_synonyms(chebi_id: str) -> List[str]:
     return list(_PRIMARY_SYNONYMS_INDEX.get(chebi_id, ()))
 
 
+def _xref_annotation(identifier: str) -> str:
+    """Repair only the historical doubled KEGG compound prefix in output annotations."""
+    match = re.fullmatch(r"kegg\.compound:cpd:(C[0-9]{5})", identifier)
+    return f"kegg.compound:{match.group(1)}" if match else identifier
+
+
 def get_xrefs(chebi_id: str) -> List[str]:
     """
-    Get all cross-references for a given CURIE.
+    Get cross-reference annotations, without changing literal identity lookup keys.
+
+    The historical doubled KEGG compound prefix is presentation-only (#1259).
+    Keep original SSSOM rows and the independently keyed xref lookup index intact:
+    canonical and alias keys can have different historical mapping winners.
 
     :param chebi_id: Primary CURIE (e.g., "CHEBI:12345")
     :return: List of xrefs (e.g., ["cas:50-00-0", "kegg.compound:C00001"])
@@ -768,7 +779,7 @@ def get_xrefs(chebi_id: str) -> List[str]:
         load_unified_mappings()
     if _PRIMARY_XREFS_INDEX is None:
         return []
-    return list(_PRIMARY_XREFS_INDEX.get(chebi_id, ()))
+    return sorted({_xref_annotation(value) for value in _PRIMARY_XREFS_INDEX.get(chebi_id, ())})
 
 
 def get_parents(curie: str) -> List[str]:
@@ -894,7 +905,10 @@ def get_node_enrichment(curie: str, *, ingredient_bundle=None) -> Dict[str, str]
         "synonym": "|".join(synonyms) if synonyms else "",
         "name": name,
     }
-    return ingredient_bundle.enrich_node(curie, result) if ingredient_bundle is not None else result
+    if ingredient_bundle is not None:
+        result = ingredient_bundle.enrich_node(curie, result)
+        result[XREF_COLUMN] = "|".join(sorted({_xref_annotation(value) for value in result[XREF_COLUMN].split("|")}))
+    return result
 
 
 class ChemicalMappingLoader:

tests/test_chemical_xref_annotations.py

@@ -0,0 +1,181 @@
+"""Correct annotation spelling without re-arbitrating historical identities (#1259)."""
+
+import csv
+import gzip
+import hashlib
+import json
+from copy import deepcopy
+
+import pytest
+
+from kg_microbe.transform_utils.constants import ID_COLUMN, OBJECT_COLUMN, SOURCE_RECORD_COLUMN, XREF_COLUMN
+from kg_microbe.utils import chemical_mapping_utils as runtime
+from kg_microbe.utils.ingredient_bundle import ReviewedIngredientBundle
+from tests.test_chemical_mapping_utils import reset_cache as reset_cache
+from tests.test_mediadive_material_scope_audit import _producer
+from tests.test_mim_conservative_refresh import FIELDS, _metadata, _row, _table
+
+CANONICAL = "kegg.compound:C00865"
+ALIAS = "kegg.compound:cpd:C00865"
+
+
+def _load(tmp_path, rows, filename="mappings.tsv"):
+    """Load a tiny original-row fixture through the actual shared reader."""
+    path = tmp_path / filename
+    _table(path, FIELDS, rows, _metadata())
+    return runtime.ChemicalMappingLoader(path), path
+
+
+@pytest.mark.parametrize("alias_first", [False, True])
+def test_annotation_does_not_change_conflicting_literal_lookup_winners(tmp_path, alias_first):
+    """Explicit canonical and alias choices survive either interleaving and reload."""
+    alias = _row(ALIAS, "CHEBI:136912", "Alias fixture")
+    canonical = _row(CANONICAL, "CHEBI:16247", "Canonical fixture")
+    rows = [alias, canonical] if alias_first else [canonical, alias]
+    rows.append(_row(CANONICAL, "CHEBI:1", "Later canonical fixture"))
+    loader, path = _load(tmp_path, rows)
+    original = path.read_bytes()
+    original_indices = deepcopy(runtime._XREF_INDEX)
+    original_xrefs = deepcopy(runtime._PRIMARY_XREFS_INDEX)
+    for identifier in ("CHEBI:136912", "CHEBI:16247", "CHEBI:1"):
+        assert runtime.get_xrefs(identifier) == [CANONICAL]
+        assert loader.get_xrefs(identifier) == [CANONICAL]
+        assert loader.get_node_enrichment(identifier)[XREF_COLUMN] == CANONICAL
+    assert runtime._PRIMARY_XREFS_INDEX == original_xrefs
+    assert runtime._XREF_INDEX == original_indices
+    assert runtime.find_chebi_by_xref(CANONICAL) == "CHEBI:16247"
+    assert runtime.find_chebi_by_xref(ALIAS) == "CHEBI:136912"
+    _load(tmp_path, [_row("cas:7732-18-5", "CHEBI:15377", "Water")], "other.tsv")
+    runtime.load_unified_mappings(path)
+    assert runtime._XREF_INDEX == original_indices
+    assert runtime._PRIMARY_XREFS_INDEX == original_xrefs
+    assert path.read_bytes() == original
+    assert runtime.get_xrefs("CHEBI:136912") == [CANONICAL]
+
+
+def test_alias_only_does_not_add_an_unasserted_canonical_lookup(tmp_path):
+    """Annotation formatting is not an extra identity fallback."""
+    loader, _ = _load(tmp_path, [_row(ALIAS, "CHEBI:136912", "Alias fixture")])
+    assert loader.get_xrefs("CHEBI:136912") == [CANONICAL]
+    assert loader.find_chebi_by_xref(ALIAS) == "CHEBI:136912"
+    assert loader.find_chebi_by_xref(CANONICAL) is None
+
+
+def test_same_target_annotations_deduplicate_without_erasing_original_rows(tmp_path):
+    """All repeated claims and original query keys remain available to the reader."""
+    rows = [_row(value, "CHEBI:7070", "Fixture salt") for value in (ALIAS, CANONICAL, ALIAS)]
+    loader, path = _load(tmp_path, rows)
+    before = hashlib.sha256(path.read_bytes()).hexdigest()
+    assert loader.get_xrefs("CHEBI:7070") == [CANONICAL]
+    assert list(runtime._iter_sssom_rows(path)) == rows
+    assert runtime._PRIMARY_XREFS_INDEX["CHEBI:7070"] == [CANONICAL, ALIAS]
+    assert loader.find_chebi_by_xref(ALIAS) == loader.find_chebi_by_xref(CANONICAL) == "CHEBI:7070"
+    returned = loader.get_xrefs("CHEBI:7070")
+    returned.append("fixture:cannot_mutate_cache")
+    assert loader.get_xrefs("CHEBI:7070") == [CANONICAL]
+    assert hashlib.sha256(path.read_bytes()).hexdigest() == before
+
+
+@pytest.mark.parametrize(
+    "identifier",
+    [
+        "kegg.compound:C12345",
+        "cpd:C12345",
+        "KEGG:C12345",
+        "kegg.drug:D12345",
+        "kegg.glycan:G12345",
+        "kegg.drug:cpd:C12345",
+        "kegg.compound:cpd:D12345",
+        "kegg.compound:cpd:C1234",
+        "kegg.compound:cpd:C123456",
+        "kegg.compound:cpd:C12345extra",
+        "kegg.compound:cpd:cpd:C12345",
+        "kegg.compound:cpd:c12345",
+        "KEGG.COMPOUND:cpd:C12345",
+        "kegg.compound:CPD:C12345",
+        "kegg.compound:cpd:C１２３４５",
+        "hsa:12345",  # codespell:ignore hsa
+        "pdb-ccd:https://example.org/entry",
+        "https://example.org/kegg.compound:cpd:C12345",
+        " kegg.compound:cpd:C12345",
+        "kegg.compound:cpd:C12345\n",
+        "",
+    ],
+)
+def test_normalizer_does_not_rewrite_other_namespaces_or_malformed_forms(identifier):
+    """Only the exact ASCII grammar is in scope, not arbitrary nested colons."""
+    assert runtime._xref_annotation(identifier) == identifier
+
+
+def test_exact_annotation_rule_is_idempotent():
+    """Normalization of the selected syntax is stable on repeated output."""
+    assert runtime._xref_annotation(ALIAS) == CANONICAL
+    assert runtime._xref_annotation(runtime._xref_annotation(ALIAS)) == CANONICAL
+
+
+@pytest.mark.parametrize(
+    "predicate,comment",
+    [
+        ("skos:broadMatch", ""),
+        ("skos:narrowMatch", ""),
+        ("skos:closeMatch", ""),
+        ("oboInOwl:hasDbXref", ""),
+        ("skos:closeMatch", "recipe_equivalent_hydrate"),
+    ],
+)
+def test_nonidentity_rows_are_not_promoted_into_xrefs(tmp_path, predicate, comment):
+    """The syntax fix does not alter the existing identity-route admission."""
+    loader, _ = _load(tmp_path, [_row(ALIAS, "CHEBI:7070", "Fixture", predicate=predicate, comment=comment)])
+    assert loader.get_xrefs("CHEBI:7070") == []
+    assert loader.find_chebi_by_xref(ALIAS) is None
+    assert loader.find_chebi_by_xref(CANONICAL) is None
+
+
+def test_actual_bundle_enrichment_preserves_annotation_boundary(tmp_path):
+    """The real bundle merger cannot leak the old spelling after node enrichment."""
+    loader, _ = _load(tmp_path, [_row(ALIAS, "CHEBI:7070", "Fixture salt")])
+    # Exercise the real merger on an isolated, already-selected annotation state;
+    # this is not a bypass for bundle admission or an identity assertion.
+    bundle = ReviewedIngredientBundle.__new__(ReviewedIngredientBundle)
+    bundle._covered = {"CHEBI:7070"}
+    bundle._identities = {}
+    bundle._xrefs = {"CHEBI:7070": {ALIAS, CANONICAL, "cas:75-57-0"}}
+    result = runtime.get_node_enrichment("CHEBI:7070", ingredient_bundle=bundle)
+    assert result[XREF_COLUMN] == "cas:75-57-0|" + CANONICAL
+    assert bundle._xrefs["CHEBI:7070"] == {ALIAS, CANONICAL, "cas:75-57-0"}
+    assert loader.find_chebi_by_xref(CANONICAL) is None
+
+
+@pytest.mark.parametrize("route", ["unified", "legacy", "embedded"])
+def test_actual_mediadive_producer_uses_canonical_annotations_on_every_identity_route(tmp_path, monkeypatch, route):
+    """The real recipe producer emits fixed xrefs without changing endpoint/context."""
+    name = "Fixture salt" if route == "unified" else "Fixture source name"
+    raw = {"compound_id": 99, "compound": name, "amount": 1, "unit": "g"}
+    producer, mapping = _producer(
+        tmp_path,
+        monkeypatch,
+        recipes={"1": {"recipe": [raw]}},
+        unified=0,
+        legacy=False,
+        embedded={"99": {"name": name, **({"ChEBI": "7070"} if route == "embedded" else {})}},
+    )
+    source = tmp_path / "tiny.tsv"
+    _table(source, FIELDS, [_row(ALIAS, "CHEBI:7070", "Fixture salt")], _metadata())
+    with gzip.open(mapping, "wt", encoding="utf-8") as stream:
+        stream.write(source.read_text())
+    producer.chemical_loader = runtime.ChemicalMappingLoader(mapping)
+    if route == "legacy":
+        producer.compound_mappings[name.lower()] = "CHEBI:7070"
+    producer.run(show_status=False)
+    with producer.output_node_file.open() as stream:
+        nodes = [row for row in csv.DictReader(stream, delimiter="\t") if row[ID_COLUMN] == "CHEBI:7070"]
+    assert len(nodes) == 1
+    assert nodes[0][XREF_COLUMN] == CANONICAL
+    with producer.output_edge_file.open() as stream:
+        edges = [row for row in csv.DictReader(stream, delimiter="\t") if row.get("source_assertion_id")]
+    assert len(edges) == 1
+    assert edges[0][OBJECT_COLUMN] == "CHEBI:7070"
+    assert edges[0]["value"] == "1.0" and edges[0]["unit"] == "g"
+    assert json.loads(edges[0][SOURCE_RECORD_COLUMN]) == raw
+    assert runtime.find_chebi_by_xref(ALIAS) == "CHEBI:7070"
+    assert runtime.find_chebi_by_xref(CANONICAL) is None