fix: fail closed on incoherent monthly citations

citeforge/bibtex_build.py

@@ -29,7 +29,7 @@
 _ARTICLE_TYPES = {"journal-article", "journal_article", "article"}
 _CONFERENCE_TYPES = {"proceedings-article", "paper-conference", "inproceedings", "conference"}
 _CHAPTER_TYPES = {"book-chapter", "book_chapter", "incollection"}
-_BOOK_TYPES = {"book", "edited-book", "monograph", "reference-book"}
+_BOOK_TYPES = {"book", "edited-book", "monograph", "proceedings", "reference-book"}
 
 _BOOK_SERIES_KEYWORDS = (
     "lecture notes",
@@ -169,12 +169,12 @@ def _classify_type_string(typ: str) -> str | None:
     match is found."""
     if "journal" in typ or typ in _ARTICLE_TYPES:
         return "article"
+    if typ in _BOOK_TYPES:
+        return "book"
     if "proceed" in typ or typ in _CONFERENCE_TYPES:
         return "inproceedings"
     if "chapter" in typ or typ in _CHAPTER_TYPES:
         return "incollection"
-    if typ in _BOOK_TYPES:
-        return "book"
     return None
 
 

citeforge/io_utils.py

@@ -33,11 +33,18 @@
     get_min_year,
 )
 from .exceptions import CSV_ERRORS, FILE_READ_ERRORS
-from .fsscan import iter_author_bibs, iter_output_dirs
+from .fsscan import iter_author_bibs, iter_output_dirs, iter_parsed_author_bibs
 from .id_utils import doi_bases_match, normalize_doi
 from .models import Record
 from .refresh.census import load_census
-from .text_utils import format_author_dirname, normalize_person_name, title_similarity
+from .text_utils import (
+    format_author_dirname,
+    name_signature,
+    normalize_person_name,
+    normalize_title,
+    parse_authors_any,
+    title_similarity,
+)
 
 _SUMMARY_CSV_FIELDNAMES = [
     "file_path",
@@ -58,6 +65,42 @@
 _SUMMARY_CSV_FLAG_FIELDS = [f for f in _SUMMARY_CSV_FIELDNAMES if f not in ("file_path", "trust_hits")]
 
 _CSV_LOCK = threading.Lock()
+_CitationCoherenceSignature = tuple[str, str, tuple[tuple[str, str], ...], str]
+
+
+def _citation_coherence_signature(entry: dict[str, Any]) -> _CitationCoherenceSignature:
+    """Return material citation fields that every copy of one DOI must agree on."""
+    fields = entry.get("fields") or {}
+    authors: list[tuple[str, str]] = []
+    for author in parse_authors_any(fields.get("author")):
+        signature = name_signature(author)
+        if signature and signature.get("last"):
+            authors.append((str(signature["last"]), str(signature.get("initials") or "")[:1]))
+    return (
+        str(entry.get("type") or "").casefold(),
+        normalize_title(str(fields.get("title") or "")),
+        tuple(authors),
+        str(fields.get("year") or "").strip(),
+    )
+
+
+def find_incoherent_doi_author_dirs(out_dir: str) -> frozenset[str]:
+    """Return author directories containing materially conflicting copies of one DOI."""
+    grouped: dict[str, list[tuple[str, _CitationCoherenceSignature]]] = {}
+    for dirname in iter_output_dirs(out_dir):
+        if dirname == A2I2_OUTPUT_DIR:
+            continue
+        author_dir = os.path.join(out_dir, dirname)
+        for _filename, _path, entry in iter_parsed_author_bibs(author_dir):
+            doi = normalize_doi((entry.get("fields") or {}).get("doi"))
+            if doi:
+                grouped.setdefault(doi, []).append((dirname, _citation_coherence_signature(entry)))
+
+    affected: set[str] = set()
+    for copies in grouped.values():
+        if len(copies) > 1 and len({signature for _, signature in copies}) > 1:
+            affected.update(dirname for dirname, _ in copies)
+    return frozenset(affected)
 
 
 def _project_root() -> str:
@@ -571,6 +614,10 @@ def _pick_richer(
 
         if matched_key is not None:
             ki = doi_to_idx[matched_key]
+            if _citation_coherence_signature(kept[ki][0]) != _citation_coherence_signature(entry):
+                raise ValueError(
+                    f"conflicting citation metadata for DOI {matched_key}: {kept[ki][1]} and {fpath}"
+                )
             kept[ki] = _pick_richer(kept[ki], (entry, fpath))
         else:
             doi_to_idx[doi] = len(kept)

citeforge/merge_utils.py

@@ -611,6 +611,29 @@ def value_ok(val: str | None) -> bool:
     merged = normalize_arxiv_metadata(merged, log=logger)
     _pop_fields(merged, {"keywords", "copyright"}, "unwanted_removed", logger)
 
+    # A Scholar baseline can combine one arXiv work with container metadata
+    # from a different publication by overlapping authors. When the baseline
+    # had no DOI and a validated secondary DOI identifies the record as a
+    # preprint, an unconfirmed Scholar-only journal is not publication
+    # evidence. Keep the validated preprint and discard the foreign container.
+    resolved_doi = _norm_doi(merged.get("doi"))
+    if (
+        resolved_doi
+        and _is_preprint_doi(resolved_doi)
+        and not primary_doi
+        and etype == "article"
+        and merged.get("journal")
+        and field_sources.get("journal") == "scholar_min"
+    ):
+        stale_journal = merged.get("journal")
+        _pop_fields(merged, {"journal", "publisher", "volume", "number", "pages"}, "unconfirmed_container", logger)
+        etype = "misc"
+        logger.debug(
+            f"preprint_container_removed | journal={stale_journal} | doi={resolved_doi} "
+            "| reason=scholar_container_unconfirmed",
+            category=LogCategory.CLEANUP,
+        )
+
     # Strip trailing digit suffixes leaked from Scholar/DBLP author
     # disambiguation markers (e.g., "Das1" becomes "Das").
     author_val = merged.get("author", "")

citeforge/pipeline/article.py

@@ -117,6 +117,9 @@ def _entry_is_complete(entry: dict[str, Any]) -> bool:
 
     has_essentials = all(fields.get(k) and not has_placeholder(str(fields.get(k))) for k in ("title", "author", "year"))
     has_doi = bool(doi) and not has_placeholder(str(doi))
+    normalized_doi = idu.normalize_doi(str(doi)) if has_doi else None
+    url_doi = idu.find_doi_in_text(str(fields.get("url") or ""))
+    url_doi_mismatch = bool(normalized_doi and url_doi and normalized_doi != url_doi)
 
     if has_essentials and has_venue and has_doi:
         doi_is_preprint = idu.is_secondary_doi(str(doi))
@@ -143,14 +146,15 @@ def _entry_is_complete(entry: dict[str, Any]) -> bool:
         and not journal_is_preprint
         and not venue_is_generic
         and not author_is_abbreviated
+        and not url_doi_mismatch
     )
 
     logger.debug(
         f"COMPLETE_CHECK | title={title[:50]} | has_title={bool(title)} "
         f"| has_author={bool(author)} | has_year={bool(year)} "
         f"| has_venue={has_venue} | has_doi={has_doi} "
         f"| doi_is_preprint={doi_is_preprint} | journal_is_preprint={journal_is_preprint} "
-        f"| author_is_abbreviated={author_is_abbreviated} "
+        f"| author_is_abbreviated={author_is_abbreviated} | url_doi_mismatch={url_doi_mismatch} "
         f"| result={result}",
         category=LogCategory.AUDIT,
     )
@@ -202,10 +206,8 @@ def _try_multiple_candidates(
 ) -> tuple[bool, Any | None]:
     """Try candidates from an API source in relevance order until one matches the baseline.
 
-    When *seen_dois* is provided, every DOI encountered across all candidates
-    (matched or not) is collected. This enables downstream duplicate detection
-    against files already on disk even when the candidate was rejected by the
-    matching gate.
+    When *seen_dois* is provided, the DOI from an identity-validated candidate
+    is collected for downstream duplicate detection against files on disk.
 
     Returns a (matched, matched_candidate) tuple.
     """
@@ -224,19 +226,17 @@ def _try_multiple_candidates(
             if not candidate_dict:
                 continue
 
-            # Collect DOI from every parsed candidate for dedup
-            if seen_dois is not None:
-                cand_doi = idu.normalize_doi((candidate_dict.get("fields") or {}).get("doi", ""))
-                if cand_doi:
-                    seen_dois.add(cand_doi)
-
             evidence = evaluate_identity(baseline_entry, candidate_dict, context=IdentityContext.ENRICHMENT)
             match = evidence.verdict
             logger.debug(
                 f"ENTRY_IDENTITY | reason={evidence.reason.value} | result={match}",
                 category=LogCategory.DEDUP,
             )
             if match:
+                if seen_dois is not None:
+                    cand_doi = idu.normalize_doi((candidate_dict.get("fields") or {}).get("doi", ""))
+                    if cand_doi:
+                        seen_dois.add(cand_doi)
                 enr_list.append((flag_key, candidate_dict))
                 flags[flag_key] = True
                 logger.success(

citeforge/pipeline/postrun.py

@@ -37,6 +37,7 @@
 from citeforge.io_utils import (
     build_a2i2_folder,
     collect_orphan_files,
+    find_incoherent_doi_author_dirs,
     flush_summary_csv,
     reconcile_summary_csv,
     retarget_summary_csv_paths,
@@ -359,7 +360,15 @@ def finalize_run(
                 category=LogCategory.CLEANUP,
             )
 
-    a2i2_count = build_a2i2_folder(DEFAULT_A2I2_INPUT, records, out_dir)
+    unresolved_doi_conflicts = find_incoherent_doi_author_dirs(out_dir)
+    if unresolved_doi_conflicts:
+        affected = ", ".join(sorted(unresolved_doi_conflicts))
+        raise FinalizationError(f"conflicting copies of the same DOI remain in author directories: {affected}")
+
+    try:
+        a2i2_count = build_a2i2_folder(DEFAULT_A2I2_INPUT, records, out_dir)
+    except ValueError as exc:
+        raise FinalizationError("a2i2 rebuild rejected incoherent citation metadata") from exc
     if a2i2_count:
         logger.info(
             f"Built a2i2 folder: {a2i2_count} deduplicated files",

citeforge/pipeline/scheduler.py

@@ -40,6 +40,7 @@
     FULL_OPERATION_ERRORS,
 )
 from citeforge.fsscan import iter_author_bibs, iter_parsed_author_bibs
+from citeforge.io_utils import find_incoherent_doi_author_dirs
 from citeforge.log_utils import LogCategory, LogSource, logger
 from citeforge.models import Record
 from citeforge.pipeline.article import process_article
@@ -350,6 +351,13 @@ def run_all(
     total_saved = 0
     processed = 0
     accounted: set[Future[int]] = set()
+    incoherent_author_dirs = find_incoherent_doi_author_dirs(out_dir)
+    if incoherent_author_dirs:
+        logger.warn(
+            f"Forcing enrichment for {len(incoherent_author_dirs)} author output directories "
+            "with conflicting copies of the same DOI",
+            category=LogCategory.PLAN,
+        )
 
     def _account_result(future: Future[int], rec: Record) -> None:
         nonlocal processed, total_saved
@@ -420,7 +428,7 @@ def _thread_excepthook(args: Any) -> None:
                     or_creds=or_creds,
                     gemini_api_key=gemini_api_key,
                     summary_csv_path=summary_csv_path,
-                    force_enrich=force_enrich,
+                    force_enrich=force_enrich or _author_dirname(rec) in incoherent_author_dirs,
                 )
                 future_to_author[future] = rec
 

citeforge/text_utils.py

@@ -257,6 +257,7 @@ def normalize_person_name(n: Any | None) -> str:
         return ""
     n_str = to_text(n)
     n2 = strip_accents(n_str).lower()
+    n2 = n2.replace("'", "").replace("\u2019", "").replace("\u02bc", "")
     n2 = _PERSON_PUNCT_RE.sub(" ", n2)
     return " ".join(n2.split())
 
@@ -555,19 +556,8 @@ def authors_overlap(authors_a: str | None, authors_b: str | None) -> bool:
     return False
 
 
-def _author_sig_key(sig: dict[str, Any]) -> str:
-    """Build a set key from a name signature, including initials when available."""
-    last = sig.get("last", "")
-    initials = sig.get("initials", "")
-    return f"{last}_{initials}" if initials else last
-
-
 def author_overlap_ratio(authors_a: str | None, authors_b: str | None) -> float:
-    """Jaccard coefficient on normalized author signatures between two author lists.
-
-    Uses last_name + initials when both sides have initials, falling back to
-    last-name-only matching otherwise.
-    """
+    """Jaccard coefficient on compatible normalized author signatures."""
     names_a = parse_authors_any(authors_a or "")
     names_b = parse_authors_any(authors_b or "")
     if not names_a or not names_b:
@@ -576,18 +566,42 @@ def author_overlap_ratio(authors_a: str | None, authors_b: str | None) -> float:
     sigs_b_raw = [sig for nm in names_b if (sig := name_signature(nm)) and sig.get("last")]
     if not sigs_a_raw or not sigs_b_raw:
         return 0.0
-    a_has_initials = all(s.get("initials") for s in sigs_a_raw)
-    b_has_initials = all(s.get("initials") for s in sigs_b_raw)
-    if a_has_initials and b_has_initials:
-        sigs_a = {_author_sig_key(s) for s in sigs_a_raw}
-        sigs_b = {_author_sig_key(s) for s in sigs_b_raw}
-    else:
-        # Fall back to last-name-only when one side lacks initials
-        sigs_a = {s["last"] for s in sigs_a_raw}
-        sigs_b = {s["last"] for s in sigs_b_raw}
-    intersection = len(sigs_a & sigs_b)
-    union = len(sigs_a | sigs_b)
-    return intersection / union if union > 0 else 0.0
+    sigs_a = {(str(sig["last"]), str(sig.get("initials") or "")) for sig in sigs_a_raw}
+    unmatched_b = {(str(sig["last"]), str(sig.get("initials") or "")) for sig in sigs_b_raw}
+    b_count = len(unmatched_b)
+    matches = 0
+    # Match the most specific signatures first and prefer exact initials. This
+    # produces a maximal one-to-one match when the same surname occurs more
+    # than once and one source abbreviates a given name to a prefix.
+    ordered_a = sorted(sigs_a, key=lambda item: (bool(item[1]), len(item[1]), item), reverse=True)
+    for last_a, initials_a in ordered_a:
+        candidates = [
+            candidate
+            for candidate in unmatched_b
+            if candidate[0] == last_a
+            and (
+                not initials_a
+                or not candidate[1]
+                or initials_a == candidate[1]
+                or initials_a.startswith(candidate[1])
+                or candidate[1].startswith(initials_a)
+            )
+        ]
+        compatible = min(
+            candidates,
+            key=lambda candidate: (
+                candidate[1] != initials_a,
+                not candidate[1],
+                -len(candidate[1]),
+                candidate,
+            ),
+            default=None,
+        )
+        if compatible is not None:
+            unmatched_b.remove(compatible)
+            matches += 1
+    union = len(sigs_a) + b_count - matches
+    return matches / union if union > 0 else 0.0
 
 
 def venue_similarity(fields_a: dict[str, Any], fields_b: dict[str, Any]) -> float:

tests/test_bibtex_build.py

@@ -230,6 +230,7 @@ def test_determine_entry_type_publication_types(pub_types: list[str], expected:
     [
         ("journal-article", "article"),
         ("proceedings-article", "inproceedings"),
+        ("proceedings", "book"),
         ("book-chapter", "incollection"),
         ("book", "book"),
         ("something-weird", "misc"),

tests/test_finalize_run.py

@@ -751,6 +751,8 @@ def test_unreadable_year_window_candidate_raises(tmp_path: Path, no_a2i2: None)
     with pytest.raises(FinalizationError, match="year-window check"):
         finalize_run(str(out_dir), _records(), total_saved=1, processed=1, summary_csv_path=str(csv_path))
 
+    assert broken.exists(), "aborting must not have deleted the file it could not read"
+
 
 def test_unreadable_candidate_raises(tmp_path: Path, no_a2i2: None) -> None:
     """A tracked file that cannot be read aborts the run rather than being skipped.
@@ -773,6 +775,39 @@ def test_unreadable_candidate_raises(tmp_path: Path, no_a2i2: None) -> None:
     assert broken.exists(), "aborting must not have deleted the file it could not read"
 
 
+def test_incoherent_a2i2_source_raises_finalization_error(
+    monkeypatch: pytest.MonkeyPatch, tmp_path: Path, no_a2i2: None
+) -> None:
+    """Conflicting member citations abort through the public finalization contract."""
+    out_dir = tmp_path / "out"
+    _author_dir(out_dir)
+
+    def reject_conflict(*_args: object, **_kwargs: object) -> int:
+        raise ValueError("conflicting citation metadata for DOI 10.5555/example")
+
+    monkeypatch.setattr(postrun, "build_a2i2_folder", reject_conflict)
+
+    with pytest.raises(FinalizationError, match="a2i2 rebuild"):
+        finalize_run(str(out_dir), _records(), total_saved=0, processed=1, summary_csv_path=None)
+
+
+def test_unresolved_same_doi_conflict_aborts_finalization(tmp_path: Path, no_a2i2: None) -> None:
+    out_dir = tmp_path / "out"
+    first_dir = _author_dir(out_dir, "Doe (abc123)")
+    second_dir = _author_dir(out_dir, "Roe (def456)")
+    common = {
+        "title": "A Shared Network Study",
+        "year": str(_IN_WINDOW_YEAR),
+        "journal": "IEEE Transactions on Machine Learning in Communications and Networking",
+        "doi": "10.1109/example",
+    }
+    write_bib(first_dir, article(key="DoeStudy", author="Jane Doe and Richard Roe", **common), "Doe-Study.bib")
+    write_bib(second_dir, article(key="DoeStudy", author="Jane Doe and Alice Poe", **common), "Doe-Study.bib")
+
+    with pytest.raises(FinalizationError, match="conflicting copies of the same DOI"):
+        finalize_run(str(out_dir), _records(), total_saved=2, processed=1, summary_csv_path=None)
+
+
 def test_failed_baseline_write_raises(tmp_path: Path, no_a2i2: None, monkeypatch: pytest.MonkeyPatch) -> None:
     """A baseline.json write that does not land aborts the run. The baseline is
     the run's own record of what it produced, so a discarded failure there makes