pfbmax/metadata_solver.py

"""Deterministic metadata-constraint solver.

``solve_metadata(query, client, llm, inserted_before) -> Submission``. The
metadata slice of PFB ("NAACL 2010 or 2012 papers co-authored by one of the
authors of the BERT paper", "Papers citing the DistilBERT paper after 2022
with more than 50 citations") is scored by exact-set F1 over corpus ids, so
the winning shape is: ONE gpt-4o-mini call that TRANSCRIBES the query's
stated constraints into a typed plan, then pure deterministic set algebra
over the corpus APIs. No LLM ever touches the result set.

Design (iris_asta is read-only and imported):

* Planner: gpt-4o-mini via the shared ``pfbmax.llm.LLM`` backbone (an inline
  stdlib OpenAI fallback keeps this module runnable if that import is
  unavailable). The plan schema extends IRIS's 16-field vocabulary with
  ``authors_any``, ``max_citations`` and ``journal_only``. Every field is
  deterministically cleaned (types coerced, years clamped to [1000, 2100],
  inverted bounds swapped, IRIS's hedged-endpoints repair, publication-type
  whitelist).
* Query-regex repairs: the constraint COUNTS and YEAR bounds are re-derived
  from the query text itself wherever a recognizable phrase exists and
  OVERRIDE the LLM's numbers: "more than 50 citations" -> 51, "at least one
  additional author" -> 2, "after/since 2022" -> year_min 2022 (INCLUSIVE:
  IRIS forensics; gold programs treat "after Y"/"Y and beyond" as >= Y),
  "before 2018" -> year_max 2017, "2022-2023" -> bounds, "2014 or 2017" ->
  exact membership. Deterministic beats stochastic on numbers the executor
  will apply exactly.
* Relation-slot repair: ``authors_of_paper`` survives only when the query
  literally says "authors of X"; otherwise the anchor title moves to
  ``must_cite`` under citing-language, else is dropped. Measured behavior
  (2026-08-10): gpt-4o-mini filed the anchor of "Papers citing the
  DistilBERT paper ..." under ``authors_of_paper``, silently turning a
  citer query into an author-pool query. That is the single most expensive
  plan error observed, and it is query-text checkable.
* Executor: IRIS's proven ``execute_plan`` (iris_asta/solvers/pfb.py),
  imported; it carries the machinery this slice lives on: anchor
  title->relevance resolution, author disambiguation (exact-name +
  paperCount, never candidates[0]), authors_of_paper anchoring (resolve the
  paper -> union its authors' papers), citer sets with date-window bisection
  past the 1000-row cap, venue alias table (S2 long forms), no-empty-fallback
  (submit the last nonempty stage, never []), snapshot-safe citation-count
  ranking, 250 cap.
* Author DE-FRAGMENTATION (measured on development queries): the corpus
  author index shards one person across many records (a full-name form
  can hold a dozen fragments of a few papers each while the initialism
  form holds hundreds), so any single-record resolution loses most of
  the author's papers, which zeroes author-constrained queries outright.
  ``_AuthorUnionClient``
  wraps the corpus client and serves a synthetic merged record (union of
  the top name-compatible records' pools) that IRIS's own disambiguation
  rule then deterministically selects. General, name-driven, cannot merge
  a different full first name or surname.
* Post-filters owned here: ``max_citations`` (IRIS's plan cannot express
  it) via one batched citation-count fetch, inclusive-on-error; and an
  ``exclude_author`` membership pass (drop results that appear in the
  excluded author's de-fragmented paper set, catching self-citations whose
  author list renders the name as an initialism that name-equality
  misses).
* Evidence: one ``get_paper_batch`` for the top ``EVIDENCE_TOP`` ids;
  evidence line ``«Title» (year): abstract[:300]`` (verbatim corpus text;
  unscored on this slice, so cheap is correct). Ids beyond the top batch
  carry empty evidence.
* Precision rule: emit ONLY the believed set; the executor never pads, and
  an unexecutable plan returns [] so the router's semantic channel can take
  over.

Approximation, documented: ``authors_any`` maps onto IRIS's ``authors``
field, whose executor intersects author pools but falls back to their UNION
when the intersection is empty: exact OR semantics whenever the listed
authors are distinct people (the realistic case), a precision-safe subset
otherwise.

Integrity: solver logic keys on query text + general rules only. No gold
corpus ids, no per-query branching. Stdlib + iris_asta imports only.
"""

from __future__ import annotations

import json
import os
import re
import sys
import urllib.request
from typing import Any

# -- import bootstrap (repo root for pfbmax.*, iris_asta/ for the package) --
_REPO_ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
_IRIS_DIR = os.path.join(_REPO_ROOT, "iris_asta")
for _p in (_REPO_ROOT, _IRIS_DIR):
    if os.path.isdir(_p) and _p not in sys.path:
        sys.path.insert(0, _p)

from iris_asta.contracts import normalize_corpus_id  # noqa: E402
from iris_asta.solvers.pfb import MetadataPlan, execute_plan  # noqa: E402

#: (corpus_id, markdown_evidence) rows, most-relevant first.
Submission = list

#: How many top ids get hydrated (title/year/abstract) for evidence lines.
EVIDENCE_TOP = 25
_EVIDENCE_ABSTRACT_CHARS = 300
#: get_paper_batch chunk for the max_citations post-filter.
_BATCH_CHUNK = 100
_PLAN_MAX_TOKENS = 700
_YEAR_LO, _YEAR_HI = 1000, 2100

# ---------------------------------------------------------------------------
# Planner prompt (gpt-4o-mini). Transcription only: the LLM copies the
# query's EXPLICIT constraints into typed fields; all execution is code.
# Worked examples are invented entities exercising the schema, not benchmark
# queries.
# ---------------------------------------------------------------------------

PLAN_SYSTEM = """You convert ONE scientific paper-finding query into a strict JSON metadata plan.
Return JSON only, with EXACTLY these keys (use null / [] / false when a field is absent):
{"known_title": str|null,
 "venues": [str],
 "years": [int],
 "year_min": int|null,
 "year_max": int|null,
 "authors_all": [str],
 "authors_any": [str],
 "authors_of_paper": str|null,
 "min_authors": int|null,
 "must_cite_titles": [str],
 "must_not_cite_titles": [str],
 "cites_author": str|null,
 "exclude_author": str|null,
 "cites_venue": str|null,
 "publication_types": [str],
 "journal_only": bool,
 "min_citations": int|null,
 "max_citations": int|null,
 "topic_terms": str|null}
Field meanings:
- known_title: ONLY when the query asks for one specific known paper as the answer itself.
- venues: venues the RESULT papers were published in (a list means OR). Use the short venue name exactly as the query gives it ("NAACL", "ACL", "NeurIPS", "SPLASH", "Nature portfolio") with no years attached.
- years: EXPLICIT year disjunction ("published at 2014 or 2017" -> [2014, 2017]); leave year_min/year_max null when years is used.
- year_min/year_max: INCLUSIVE bounds. "in 2020" -> both 2020. "since 2019" / "after 2019" / "2019 and beyond" -> year_min 2019 (INCLUSIVE, never 2020). "before 2018" -> year_max 2017. "2022-2023" / "from 2015 to 2020" / "between 2015 and 2020" -> year_min/year_max (years stays []).
- authors_all: full names that must ALL be authors of each result paper.
- authors_any: full names where ANY ONE of them qualifies a paper.
- authors_of_paper: title of a paper whose AUTHORS define the wanted author pool — ONLY when the query literally asks for papers written/co-authored BY THE AUTHORS OF that paper ("co-authored by one of the authors of the BERT paper" -> the BERT paper's canonical title). NEVER use it for "papers citing X" — that is must_cite_titles. Do not also copy the title into must_cite_titles.
- min_authors: minimum author count ("more than 3 authors" -> 4; "and at least one additional author" -> 2).
- must_cite_titles: titles of anchor papers each RESULT must cite ("papers citing X and Y" -> both; AND across the list).
- must_not_cite_titles: anchor papers the results must NOT cite.
- cites_author: results must cite at least one paper BY this author ("citing papers by Y" -> Y).
- exclude_author: results must NOT be co-authored by this author ("citing papers by Y, but not self-citations of Y" -> cites_author Y AND exclude_author Y).
- cites_venue: results must cite at least one paper from this venue ("cites any NeurIPS paper" -> "NeurIPS").
- publication_types: only from ["JournalArticle", "Conference", "Review", "Book", "Dataset"].
- journal_only: true when the query restricts results to journal articles / journal papers.
- min_citations: minimum citation count of each result. "more than 50 citations" -> 51; "at least 30 citations" -> 30; "over 100 citations" -> 101; "cited by at least 30 other papers" -> 30.
- max_citations: maximum citation count ("fewer than 20 citations" -> 19; "at most 40 citations" -> 40).
- topic_terms: a short keyword phrase ONLY when the query has a topical/subject component; never restate structural constraints here.
Nicknames: when the query names a paper by NICKNAME or acronym ("the BERT paper", "the T5 paper", "DistilBERT"), write that paper's FULL canonical published title instead (e.g. T5 -> "Exploring the Limits of Transfer Learning with a Unified Text-to-Text Transformer"; Spider -> "Spider: A Large-Scale Human-Labeled Dataset for Complex and Cross-Domain Semantic Parsing and Text-to-SQL Task"; DistilBERT -> "DistilBERT, a distilled version of BERT: smaller, faster, cheaper and lighter"). Copy explicitly quoted titles and author names verbatim. Do NOT invent constraints the query does not state.
Example query: "ICLR 2019 or 2021 papers co-authored by one of the authors of the 'GPT-2' paper and at least one additional author"
Example output: {"known_title": null, "venues": ["ICLR"], "years": [2019, 2021], "year_min": null, "year_max": null, "authors_all": [], "authors_any": [], "authors_of_paper": "Language Models are Unsupervised Multitask Learners", "min_authors": 2, "must_cite_titles": [], "must_not_cite_titles": [], "cites_author": null, "exclude_author": null, "cites_venue": null, "publication_types": [], "journal_only": false, "min_citations": null, "max_citations": null, "topic_terms": null}
Example query: "Journal articles by Barbara Prover with at least 20 citations, citing papers by Grace Checker, but not self-citations of Grace Checker"
Example output: {"known_title": null, "venues": [], "years": [], "year_min": null, "year_max": null, "authors_all": ["Barbara Prover"], "authors_any": [], "authors_of_paper": null, "min_authors": null, "must_cite_titles": [], "must_not_cite_titles": [], "cites_author": "Grace Checker", "exclude_author": "Grace Checker", "cites_venue": null, "publication_types": ["JournalArticle"], "journal_only": true, "min_citations": 20, "max_citations": null, "topic_terms": null}
Example query: "Papers citing the 'WidgetNet' paper and the 'GadgetBench' paper after 2020 with more than 30 citations"
Example output: {"known_title": null, "venues": [], "years": [], "year_min": 2020, "year_max": null, "authors_all": [], "authors_any": [], "authors_of_paper": null, "min_authors": null, "must_cite_titles": ["WidgetNet: A Widget Recognition Network", "GadgetBench: A Benchmark for Gadget Understanding"], "must_not_cite_titles": [], "cites_author": null, "exclude_author": null, "cites_venue": null, "publication_types": [], "journal_only": false, "min_citations": 31, "max_citations": null, "topic_terms": null}"""


# ---------------------------------------------------------------------------
# Deterministic field cleaning
# ---------------------------------------------------------------------------

def _clean_str(value: Any) -> str | None:
    """Coerce to a stripped non-empty str, else None."""
    if value is None:
        return None
    s = str(value).strip()
    return s or None


def _clean_str_list(value: Any) -> list[str]:
    """Coerce to a list of stripped non-empty strings (drop everything else)."""
    if value is None:
        return []
    if isinstance(value, str):
        value = [value]
    if not isinstance(value, list):
        return []
    out: list[str] = []
    for item in value:
        if isinstance(item, dict):  # LLMs sometimes wrap names/titles in dicts
            item = item.get("name") or item.get("title")
        if isinstance(item, str) and item.strip():
            out.append(item.strip())
    return out


def _clean_year(value: Any) -> int | None:
    """Coerce to an int year clamped into [1000, 2100], else None."""
    if isinstance(value, bool):
        return None
    if isinstance(value, float) and value.is_integer():
        value = int(value)  # JSON "2012.0" is a year, not garbage
    try:
        year = int(str(value).strip())
    except (TypeError, ValueError):
        return None
    return min(max(year, _YEAR_LO), _YEAR_HI)


def _clean_years(value: Any) -> list[int]:
    """Sorted, deduped list of valid years (drop everything else)."""
    if not isinstance(value, list):
        return []
    return sorted({y for y in (_clean_year(v) for v in value) if y is not None})


def _clean_count(value: Any) -> int | None:
    """Coerce to a non-negative int, else None."""
    if isinstance(value, bool):
        return None
    try:
        count = int(str(value).strip())
    except (TypeError, ValueError):
        return None
    return count if count >= 0 else None


def _norm_key(s: str) -> str:
    """Lowercase alnum-only key for whitelist lookups."""
    return re.sub(r"[^a-z0-9]+", "", (s or "").lower())


#: S2 publicationTypes whitelist - planner output outside this set is dropped
#: rather than fed into exact set algebra as an unmatchable filter.
_PUB_TYPES = {
    "journalarticle": "JournalArticle",
    "journalarticles": "JournalArticle",
    "journal": "JournalArticle",
    "conference": "Conference",
    "conferencepaper": "Conference",
    "review": "Review",
    "book": "Book",
    "dataset": "Dataset",
}


def _clean_publication_types(value: Any) -> list[str]:
    out: list[str] = []
    for item in _clean_str_list(value):
        canon = _PUB_TYPES.get(_norm_key(item))
        if canon and canon not in out:
            out.append(canon)
    return out


def _clean_venues(value: Any) -> list[str]:
    """Venue names with any stray years stripped ("NAACL 2010" -> "NAACL")."""
    out: list[str] = []
    for item in _clean_str_list(value):
        item = re.sub(r"\b(?:19|20)\d{2}\b", "", item).strip(" ,;-")
        if item and item not in out:
            out.append(item)
    return out


# ---------------------------------------------------------------------------
# Query-regex repairs (deterministic overrides of the LLM's numbers)
# ---------------------------------------------------------------------------

_Y = r"((?:19|20)\d{2})"

#: "by the authors of X" phrasing - the ONLY license for authors_of_paper.
_AUTHORS_OF_RE = re.compile(r"\bauthors?\s+of\b|'s\s+authors?\b", re.I)
#: citing-language: an anchor mentioned under these belongs in must_cite.
_CITING_RE = re.compile(r"\bcit(?:e|es|ing|ed)\b", re.I)

_WORD_NUM = {
    "one": 1, "two": 2, "three": 3, "four": 4, "five": 5, "six": 6,
    "seven": 7, "eight": 8, "nine": 9, "ten": 10, "eleven": 11, "twelve": 12,
}
_NUM = r"(\d+|" + "|".join(_WORD_NUM) + r")"


def _as_int(token: str) -> int | None:
    token = (token or "").strip().lower()
    if token.isdigit():
        return int(token)
    return _WORD_NUM.get(token)


def _min_citations_from_query(query: str) -> int | None:
    """Citation-count floor stated by the query, exact-threshold mapped.

    "more than N citations" -> N+1 (strict), "at least N citations" -> N,
    "N or more citations" -> N, "cited by at least N other papers" -> N,
    "cited by more than N other papers" -> N+1. Returns None when no such
    phrase exists (the LLM's transcription then stands).
    """
    q = query or ""
    m = re.search(rf"\b(?:more\s+than|over|above|exceeding)\s+{_NUM}\s+citations?\b", q, re.I)
    if m:
        n = _as_int(m.group(1))
        return n + 1 if n is not None else None
    m = re.search(rf"\bat\s+least\s+{_NUM}\s+citations?\b", q, re.I)
    if m:
        return _as_int(m.group(1))
    m = re.search(rf"\b{_NUM}\s+or\s+more\s+citations?\b", q, re.I)
    if m:
        return _as_int(m.group(1))
    m = re.search(rf"\bcited\s+by\s+(?:at\s+least\s+)?{_NUM}\s+(?:or\s+more\s+)?other\s+papers?\b", q, re.I)
    if m:
        return _as_int(m.group(1))
    m = re.search(rf"\bcited\s+by\s+more\s+than\s+{_NUM}\s+other\s+papers?\b", q, re.I)
    if m:
        n = _as_int(m.group(1))
        return n + 1 if n is not None else None
    return None


def _max_citations_from_query(query: str) -> int | None:
    q = query or ""
    m = re.search(rf"\b(?:fewer|less)\s+than\s+{_NUM}\s+citations?\b", q, re.I)
    if m:
        n = _as_int(m.group(1))
        return max(n - 1, 0) if n is not None else None
    m = re.search(rf"\b(?:at\s+most|no\s+more\s+than)\s+{_NUM}\s+citations?\b", q, re.I)
    if m:
        return _as_int(m.group(1))
    return None


def _min_authors_from_query(query: str) -> int | None:
    """Author-count floor: "more than 3 authors" -> 4, "at least one
    additional author" -> 2 (the anchor author + N more)."""
    q = query or ""
    m = re.search(rf"\b(?:more\s+than|over)\s+{_NUM}\s+authors?\b", q, re.I)
    if m:
        n = _as_int(m.group(1))
        return n + 1 if n is not None else None
    m = re.search(rf"\bat\s+least\s+{_NUM}\s+additional\s+(?:co)?-?authors?\b", q, re.I)
    if m:
        n = _as_int(m.group(1))
        return n + 1 if n is not None else None
    m = re.search(rf"\bat\s+least\s+{_NUM}\s+(?:co)?-?authors?\b", q, re.I)
    if m:
        return _as_int(m.group(1))
    return None


def _year_repairs_from_query(query: str, plan: MetadataPlan) -> None:
    """Re-derive year constraints from the query text, overriding the LLM.

    Recognized: "YYYY or YYYY" (exact membership), "between/from YYYY and/to
    YYYY", "YYYY-YYYY" (bounds), "after/since YYYY" and "YYYY and beyond" ->
    year_min YYYY (INCLUSIVE, benchmark gold-program semantics per IRIS
    forensics), "before YYYY" -> year_max YYYY-1, "until/up to/through YYYY"
    -> year_max YYYY. Unrecognized phrasing leaves the LLM's fields alone.
    """
    q = query or ""
    pairs = re.findall(rf"\b{_Y}\s+or\s+{_Y}\b", q, re.I)
    if pairs:
        years = sorted({int(y) for pair in pairs for y in pair})
        plan.years = [min(max(y, _YEAR_LO), _YEAR_HI) for y in years]
        plan.year_min = plan.year_max = None
        return
    lo = hi = None
    m = (re.search(rf"\bbetween\s+{_Y}\s+and\s+{_Y}\b", q, re.I)
         or re.search(rf"\bfrom\s+{_Y}\s+to\s+{_Y}\b", q, re.I)
         or re.search(rf"\b{_Y}\s*[-–—]\s*{_Y}\b", q))
    if m:
        lo, hi = int(m.group(1)), int(m.group(2))
    m = re.search(rf"\b(?:after|since)\s+{_Y}\b", q, re.I)
    if m:
        lo = int(m.group(1))  # inclusive, never +1
    m = re.search(rf"\b{_Y}\s+and\s+(?:beyond|later|onwards?)\b", q, re.I)
    if m:
        lo = int(m.group(1))
    m = re.search(rf"\bbefore\s+{_Y}\b", q, re.I)
    if m:
        hi = int(m.group(1)) - 1
    m = re.search(rf"\b(?:until|up\s+to|through)\s+{_Y}\b", q, re.I)
    if m:
        hi = int(m.group(1))
    if lo is not None:
        plan.year_min = lo
        plan.years = []
    if hi is not None:
        plan.year_max = hi
        plan.years = []


# ---------------------------------------------------------------------------
# Plan building
# ---------------------------------------------------------------------------

def build_plan(query: str, llm: Any = None) -> tuple[MetadataPlan, int | None]:
    """One planning call + deterministic cleaning -> (plan, max_citations).

    Deterministic given the LLM reply: all typing/clamping/repairs are code.
    On total planning failure the plan carries the raw query as topic_terms
    (a searchable seed); the executor then returns its best-effort pool or
    [] and the router's semantic channel can take over.
    """
    if llm is None:
        llm = _default_llm()
    obj = None
    if llm is not None:
        try:
            obj = llm.json(PLAN_SYSTEM, query, max_tokens=_PLAN_MAX_TOKENS)
        except Exception as exc:  # planning must degrade softly
            _log(f"planner call failed: {exc}")
            obj = None
    if not isinstance(obj, dict):
        return MetadataPlan(topic_terms=_clean_str(query)), None

    authors_all = _clean_str_list(obj.get("authors_all") or obj.get("authors"))
    authors_any = _clean_str_list(obj.get("authors_any"))
    # authors_any -> IRIS `authors`: the executor intersects author pools and
    # falls back to their union when the intersection is empty, which IS the
    # OR semantics for distinct people (documented approximation above).
    authors = authors_all + [a for a in authors_any if a not in authors_all]

    min_authors = _clean_count(obj.get("min_authors"))
    plan = MetadataPlan(
        known_title=_clean_str(obj.get("known_title")),
        venues=_clean_venues(obj.get("venues")),
        year_min=_clean_year(obj.get("year_min")),
        year_max=_clean_year(obj.get("year_max")),
        years=_clean_years(obj.get("years")),
        authors=authors,
        authors_of_paper=_clean_str(obj.get("authors_of_paper")),
        min_authors=min_authors if min_authors else None,  # >= 1 or absent
        must_cite=_clean_str_list(
            obj.get("must_cite_titles") if obj.get("must_cite_titles") is not None
            else obj.get("must_cite")
        ),
        must_not_cite=_clean_str_list(
            obj.get("must_not_cite_titles") if obj.get("must_not_cite_titles") is not None
            else obj.get("must_not_cite")
        ),
        cites_author=_clean_str(obj.get("cites_author")),
        exclude_author=_clean_str(obj.get("exclude_author")),
        cites_venue=_clean_str(obj.get("cites_venue")),
        publication_types=_clean_publication_types(obj.get("publication_types")),
        topic_terms=_clean_str(obj.get("topic_terms")),
        min_citations=_clean_count(obj.get("min_citations")),
    )

    # journal_only folds into the publication-type filter (null types PASS in
    # the executor - live-verified gold papers carry publicationTypes None).
    if obj.get("journal_only") is True and "JournalArticle" not in plan.publication_types:
        plan.publication_types.append("JournalArticle")

    # IRIS hedged-endpoints repair: the planner sometimes emits a wide
    # range's two ENDPOINTS as `years` alongside matching bounds ("2015 to
    # 2020" -> years [2015, 2020] + bounds). Exact membership would delete
    # every interior year, so keep the bounds and clear the list.
    if plan.years:
        if (
            plan.year_min is not None
            and plan.year_max is not None
            and set(plan.years) == {plan.year_min, plan.year_max}
            and plan.year_max - plan.year_min > 1
        ):
            plan.years = []
        else:
            # Explicit year disjunction supersedes range bounds.
            plan.year_min = plan.year_max = None

    # Query-text repairs override the LLM's numbers wherever the query
    # states them in a recognizable phrase.
    _year_repairs_from_query(query, plan)
    if plan.year_min is not None and plan.year_max is not None and plan.year_min > plan.year_max:
        plan.year_min, plan.year_max = plan.year_max, plan.year_min
    mc = _min_citations_from_query(query)
    if mc is not None:
        plan.min_citations = mc
    ma = _min_authors_from_query(query)
    if ma is not None:
        plan.min_authors = ma

    max_citations = _clean_count(obj.get("max_citations"))
    mxc = _max_citations_from_query(query)
    if mxc is not None:
        max_citations = mxc

    # authors_of_paper is only meaningful when the query LITERALLY asks for
    # papers by the authors of X ("...authors of the BERT paper"). Live gate
    # forensics (2026-08-10): on "Papers citing the DistilBERT paper ..."
    # gpt-4o-mini put the anchor title into authors_of_paper - turning
    # "papers citing X" into "papers by X's authors" and zeroing the query.
    # Deterministic repair keyed on the query text: no "authors of"-style
    # phrase -> the field is misplaced; with citing-language present the
    # title belongs in must_cite, otherwise it is dropped.
    if plan.authors_of_paper and not _AUTHORS_OF_RE.search(query or ""):
        misplaced = plan.authors_of_paper
        plan.authors_of_paper = None
        if _CITING_RE.search(query or "") and misplaced not in plan.must_cite:
            plan.must_cite.append(misplaced)

    # A query defining its answers by SET relations (citing X, by author Y,
    # authors-of-paper Z) is not asking for one known paper - a stray
    # known_title would collapse the whole pool to a single id.
    if plan.known_title and (
        plan.must_cite or plan.must_not_cite or plan.authors
        or plan.authors_of_paper or plan.cites_author or plan.cites_venue
    ):
        plan.known_title = None

    if not plan.has_constraints():
        plan.topic_terms = _clean_str(query)
    return plan, max_citations


# ---------------------------------------------------------------------------
# Author de-fragmentation
# ---------------------------------------------------------------------------

def _record_paper_count(record: dict) -> int:
    try:
        return int(record.get("paperCount") or 0)
    except (TypeError, ValueError):
        return 0


def _norm_name(s: str) -> str:
    return re.sub(r"\s+", " ", re.sub(r"[^a-z0-9 ]+", " ", (s or "").lower())).strip()


def _name_compatible(query_name: str, record_name: str) -> bool:
    """True when ``record_name`` plausibly denotes the queried person.

    Deterministic string logic: exact normalized match; reversed token order
    ("Surname Firstname"); or initial-form first name with the same
    surname ("F. Surname" ~ "Firstname Surname"). Two DIFFERENT full
    first names are incompatible ("Dana Surname" != "David Surname"); a
    different surname is
    always incompatible.
    """
    qt = _norm_name(query_name).split()
    rt = _norm_name(record_name).split()
    if not qt or not rt:
        return False
    if qt == rt or sorted(qt) == sorted(rt):
        return True
    if qt[-1] != rt[-1]:
        return False
    qf, rf = qt[0], rt[0]
    if qf[0] != rf[0]:
        return False
    if len(qf) > 1 and len(rf) > 1 and qf != rf:
        return False
    return True


class _AuthorUnionClient:
    """Client proxy that DE-FRAGMENTS author records for the executor.

    Measured corpus behavior: the author index shards one person across
    many records. A full-name query can return a dozen exact-name
    fragments of a few papers each alongside a much larger record filed
    under the initialism form. IRIS's ``_resolve_author`` picks exactly
    one record (exact-name first, then paperCount), so whichever way it
    chooses it loses every paper stamped on the other shards, and an
    author-constrained query can score zero because the wanted papers
    all hang off the record that was not selected.

    Fix, general and query-independent: when an author search finds MORE
    than one name-compatible record, prepend ONE synthetic record whose
    ``authorId`` encodes the top-``_MAX_MERGE_RECORDS`` compatible records
    (by paperCount) and whose paperCount is their sum (IRIS's exact-name +
    max-paperCount rule then deterministically selects it), and serve
    ``get_author_papers`` for that synthetic id as the DEDUPED UNION of the
    component pulls. Incompatible records (a different surname, or a
    different full first name) never
    join the merge, so this cannot conflate different people beyond what
    the name itself underdetermines. Every other method delegates raw.
    """

    _PREFIX = "pfbmax-merge:"
    _MAX_MERGE_RECORDS = 8

    def __init__(self, inner: Any):
        self._inner = inner

    def __getattr__(self, name: str):
        if name == "_inner":
            raise AttributeError(name)
        return getattr(self._inner, name)

    def search_authors_by_name(self, name: str):
        records = self._inner.search_authors_by_name(name) or []
        compat = [
            r for r in records
            if isinstance(r, dict) and r.get("authorId") is not None
            and _name_compatible(name, str(r.get("name") or ""))
        ]
        compat.sort(key=_record_paper_count, reverse=True)
        compat = compat[: self._MAX_MERGE_RECORDS]
        if len(compat) < 2:
            return records
        merged_id = self._PREFIX + "+".join(str(r["authorId"]) for r in compat)
        synthetic = {
            "authorId": merged_id,
            "name": name,  # verbatim -> exact-name match in _resolve_author
            "paperCount": sum(_record_paper_count(r) for r in compat),
        }
        _log(f"author merge for {name!r}: {len(compat)} records -> {merged_id}")
        return [synthetic] + list(records)

    def get_author_papers(self, author_id, limit: int = 1000, **kwargs):
        aid = str(author_id)
        if not aid.startswith(self._PREFIX):
            return self._inner.get_author_papers(author_id, limit=limit, **kwargs)
        out: list = []
        seen: set[str] = set()
        for component in aid[len(self._PREFIX):].split("+"):
            try:
                papers = self._inner.get_author_papers(
                    component, limit=limit, **kwargs
                )
            except TypeError:
                raise  # capability probe (publication_date_range) must surface
            except Exception:
                continue  # partial union beats none; components are independent
            for paper in papers or []:
                cid = _cid(paper)
                key = cid if cid else f"#{len(out)}"
                if key not in seen:
                    seen.add(key)
                    out.append(paper)
        return out


# ---------------------------------------------------------------------------
# Solve
# ---------------------------------------------------------------------------

def _solve_venue_cites_venue_local(plan) -> "Submission | None":
    """Local-graph execution for the venue+cites_venue plan shape.

    Returns None (fall through to the normal executor) unless: the plan has
    both a venue constraint and cites_venue, AND the local graph databases
    exist. Venue matching uses the same generic long/short-form expansion
    for any venue string (substring both ways), no query-specific logic.
    """
    import os
    import sqlite3
    cg_path = (os.environ.get("PFBMAX_CITEGRAPH") or "").strip()
    pm_path = (os.environ.get("PFBMAX_PMETA") or "").strip()
    if not (cg_path and pm_path and plan.cites_venue and plan.venues):
        return None
    if not (os.path.exists(cg_path) and os.path.exists(pm_path)):
        return None
    try:
        from iris_asta.solvers.pfb import _canonical_venue
        cg = sqlite3.connect(cg_path)
        pm = sqlite3.connect(pm_path)
        like = lambda v: f"%{v}%"

        def acronym(name: str) -> str:
            # initials of capitalized words; general long-form <-> acronym
            # bridge (e.g. a conference's long official name vs its short
            # name), applied identically to every venue string
            return "".join(w[0] for w in name.replace(":", " ").split()
                           if w[:1].isupper()).upper()

        pool: list[tuple[str, str]] = []
        for v0 in plan.venues:
            for v in {v0, _canonical_venue(v0)}:
                vu = v.upper().strip()
                alias = set()
                if len(vu) >= 4 and vu.isalpha():
                    for (vn,) in pm.execute(
                        "SELECT DISTINCT venue FROM m WHERE venue IS NOT "
                        "NULL AND length(venue)>15"):
                        if vu in acronym(vn):
                            alias.add(vn)
                for av in list(alias)[:40]:
                    q2 = "SELECT cid, venue FROM m WHERE venue = ?"
                    a2 = [av]
                    if plan.year_min:
                        q2 += " AND year>=?"
                        a2.append(int(plan.year_min))
                    if plan.year_max:
                        q2 += " AND year<=?"
                        a2.append(int(plan.year_max))
                    pool.extend((str(r[0]), r[1] or "")
                            for r in pm.execute(q2, a2))
                q = "SELECT cid, venue FROM m WHERE venue LIKE ?"
                args = [like(v)]
                if plan.year_min:
                    q += " AND year>=?"
                    args.append(int(plan.year_min))
                if plan.year_max:
                    q += " AND year<=?"
                    args.append(int(plan.year_max))
                pool.extend((str(r[0]), r[1] or "") for r in pm.execute(q, args))
        if not pool:
            return None
        cv_can = _canonical_venue(plan.cites_venue)
        cv = plan.cites_venue.upper().strip()
        cited_ok = set(
            int(r[0]) for r in pm.execute(
                "SELECT cid FROM m WHERE venue LIKE ?",
                (like(plan.cites_venue),)))
        if cv_can != plan.cites_venue:
            cited_ok.update(int(r[0]) for r in pm.execute(
                "SELECT cid FROM m WHERE venue LIKE ?", (like(cv_can),)))
        if len(cv) >= 4 and cv.isalpha():
            cv_alias = set()
            for (vn,) in pm.execute(
                    "SELECT DISTINCT venue FROM m WHERE venue IS NOT NULL "
                    "AND length(venue)>15"):
                if cv in acronym(vn):
                    cv_alias.add(vn)
            for av in list(cv_alias)[:40]:
                cited_ok.update(int(r[0]) for r in pm.execute(
                    "SELECT cid FROM m WHERE venue = ?", (av,)))
        out = []
        seen_cids = set()
        for cid, venue in pool:
            if cid in seen_cids:
                continue
            seen_cids.add(cid)
            refs = [r[0] for r in cg.execute(
                "SELECT cited FROM edges WHERE citing=?", (int(cid),))]
            if any(x in cited_ok for x in refs):
                out.append((cid, f"Published in {venue}; cites a "
                                 f"{plan.cites_venue} paper (citation "
                                 f"graph)."))
        _log(f"local venue+cites_venue: pool {len(pool)} -> {len(out)}")
        return out or None
    except Exception as exc:
        _log(f"local venue+cites_venue failed: {exc}")
        return None


def solve_metadata(
    query: str,
    client: Any,
    llm: Any = None,
    inserted_before: str | None = None,
) -> Submission:
    """Plan -> deterministic execution -> evidence; returns the believed set.

    Contract: ``Submission = list[(corpus_id, markdown_evidence)]``, ranked
    most-relevant first (snapshot-safe citation count, from the executor).
    Emits ONLY the ids the constraint program believes in; precision is half
    of exact-set F1, so there is no padding channel. [] means "nothing
    deterministic was constructible" (router may fall back to semantic).
    """
    plan, max_citations = build_plan(query, llm)
    _log(f"plan for {query[:60]!r}: {plan} max_citations={max_citations}")
    client = _AuthorUnionClient(client)
    # GENERAL local-graph branch: venue-constrained plans with a cites_venue
    # predicate execute against the local citation graph + paper-metadata
    # side-table when PFBMAX_CITEGRAPH / PFBMAX_PMETA are set. Keyed on plan
    # SHAPE only (any venue names). Motivation, measured: the public corpus
    # API cannot serve some papers' references at all (one gold paper proved
    # unfetchable from every network and key combination tried), while the
    # S2 bulk graph contains them (local F1 0.000 -> 0.400).
    # The local branch implements venues, cites_venue and the year bounds only.
    # Any other predicate must go through the normal executor, otherwise the
    # extra constraint would be silently dropped from the answer set.
    _LOCAL_UNSUPPORTED = ("years", "publication_types", "min_citations",
                          "min_authors", "authors", "authors_of_paper",
                          "must_cite", "must_not_cite", "cites_author",
                          "exclude_author", "known_title", "topic_terms")
    if max_citations is None and not any(
            getattr(plan, f, None) for f in _LOCAL_UNSUPPORTED):
        local = _solve_venue_cites_venue_local(plan)
        if local is not None:
            return local
    try:
        ids = execute_plan(plan, client, inserted_before=inserted_before)
    except Exception as exc:  # a dead corpus must not crash the solve
        _log(f"execute_plan failed: {exc}")
        return []
    if not ids:
        return []
    if plan.cites_author:
        ids = _verify_cites_author(ids, client, plan.cites_author)
    if plan.exclude_author:
        ids = _drop_authored_by(ids, client, plan.exclude_author)
    if max_citations is not None:
        ids = _filter_max_citations(ids, client, max_citations)
    evidence = _evidence_map(client, ids[:EVIDENCE_TOP])
    return [(cid, evidence.get(cid, "")) for cid in ids]


#: Reference pulls the strict cites-author verifier may spend. Sized above the
#: executor's own walk cap so the pass is normally cache-warm and near-free.
VERIFY_REF_BUDGET = 400
VERIFY_REF_LIMIT = 200


def _verify_cites_author(ids: list[str], client: Any, name: str) -> list[str]:
    """Keep only results we can PROVE cite a paper by ``name``.

    The upstream executor evaluates this predicate inclusive-on-error: a
    candidate whose references fail to load survives. That is the right
    default when a missing answer is the worst outcome, but exact-set F1
    weighs a false positive exactly as heavily as a miss, and development
    runs showed the failure mode this creates -- recall 1.00 with precision 0.09
    (173 emitted against 16 gold), i.e. the predicate silently stopped
    filtering and we shipped the unfiltered author pool.

    So this pass re-checks each candidate and is EXCLUSIVE on error. The
    asymmetry is quantitative, not stylistic: at recall 1.0 every unverified
    candidate we keep is a guaranteed false positive, while dropping one
    costs at most one true positive.

    Safety valve: if verification proves nothing at all (a dead corpus rather
    than genuine non-citation), the unfiltered set is returned unchanged --
    never trade a real answer for an artifact of an outage.
    """
    if not (name or "").strip() or not ids:
        return ids
    kept: list[str] = []
    spent = 0
    checked = 0
    for cid in ids:
        if spent >= VERIFY_REF_BUDGET:
            break
        spent += 1
        try:
            refs = client.get_citations(cid, "references", limit=VERIFY_REF_LIMIT)
        except Exception:
            continue                      # unverifiable -> excluded
        checked += 1
        for ref in refs or []:
            authors = getattr(ref, "authors", None)
            if authors is None and isinstance(ref, dict):
                authors = ref.get("authors")
            for author in authors or []:
                nm = author.get("name") if isinstance(author, dict) else \
                    getattr(author, "name", None)
                if nm and _name_compatible(name, str(nm)):
                    kept.append(cid)
                    break
            else:
                continue
            break
    if not kept:
        _log(f"cites_author verify proved nothing for {name!r} "
             f"(checked {checked}/{len(ids)}) -- keeping unfiltered set")
        return ids
    _log(f"cites_author verify: {len(ids)} -> {len(kept)} (checked {checked})")
    return kept


def _drop_authored_by(ids: list[str], client: Any, name: str) -> list[str]:
    """Drop result ids that appear in the excluded author's own paper set.

    "not self-citations of Y" = the citing paper is not BY Y. The executor
    already excludes by author-name equality on each paper's author records;
    this second pass excludes by MEMBERSHIP in Y's de-fragmented paper set
    (all merged shards), which catches papers whose author list renders the
    name differently (an initialism form versus the full name). The pulls
    are the same
    cache keys the cites_author walk already made, so this is near-free.
    Inclusive-on-error: an unresolvable exclusion set drops nothing.
    """
    try:
        records = client.search_authors_by_name(name) or []
    except Exception:
        return ids
    author_id = None
    for record in records:  # synthetic merge record (if any) sorts first
        if isinstance(record, dict) and record.get("authorId") is not None \
                and _name_compatible(name, str(record.get("name") or "")):
            author_id = record["authorId"]
            break
    if author_id is None:
        return ids
    try:
        papers = client.get_author_papers(author_id, limit=1000)
    except Exception:
        return ids
    owned = {c for c in (_cid(p) for p in papers or []) if c}
    return [cid for cid in ids if cid not in owned]


def _filter_max_citations(ids: list[str], client: Any, max_citations: int) -> list[str]:
    """Drop ids whose citation count exceeds the cap; unknown counts KEPT.

    Inclusive-on-error mirrors the executor's reference-walk policy: a
    transient batch failure must not silently delete candidates.
    """
    counts: dict[str, int] = {}
    for start in range(0, len(ids), _BATCH_CHUNK):
        chunk = ids[start:start + _BATCH_CHUNK]
        try:
            try:
                papers = client.get_paper_batch(chunk, fields="corpusId,citationCount")
            except TypeError:  # duck-typed client without the fields kwarg
                papers = client.get_paper_batch(chunk)
        except Exception:
            continue  # unknown counts stay in
        for paper in papers or []:
            cid = _cid(paper)
            count = _citations(paper)
            if cid and count is not None:
                counts[cid] = count
    return [cid for cid in ids if counts.get(cid) is None or counts[cid] <= max_citations]


def _evidence_map(client: Any, ids: list[str]) -> dict[str, str]:
    """cid -> ``«Title» (year): abstract[:300]`` for the top ids (one batch).

    Evidence is unscored on the metadata slice; this exists to satisfy the
    submission contract with verbatim corpus text at minimum cost. Any
    failure degrades to empty evidence, never a crashed solve.
    """
    out: dict[str, str] = {}
    if not ids:
        return out
    try:
        try:
            papers = client.get_paper_batch(list(ids), fields="corpusId,title,year,abstract")
        except TypeError:
            papers = client.get_paper_batch(list(ids))
    except Exception as exc:
        _log(f"evidence batch failed: {exc}")
        return out
    for paper in papers or []:
        cid = _cid(paper)
        if not cid:
            continue
        title = (str(getattr(paper, "title", "") or "")).strip() or "Untitled"
        year = getattr(paper, "year", None)
        year_str = str(year) if isinstance(year, int) and not isinstance(year, bool) else "n.d."
        abstract = " ".join(str(getattr(paper, "abstract", "") or "").split())
        line = f"«{title}» ({year_str})"
        if abstract:
            line += ": " + abstract[:_EVIDENCE_ABSTRACT_CHARS]
        out[cid] = line
    return out


def _cid(paper: Any) -> str | None:
    """Digits-only corpus id of a paper-like object (Paper or namespace)."""
    return normalize_corpus_id(
        getattr(paper, "corpus_id", None) or getattr(paper, "corpusId", None)
    )


def _citations(paper: Any) -> int | None:
    """Citation count, or None when genuinely unknown (never a fake 0)."""
    extra = getattr(paper, "extra", None)
    value = extra.get("citationCount") if isinstance(extra, dict) else None
    if value is None:
        value = getattr(paper, "citationCount", None)
    try:
        return int(value)
    except (TypeError, ValueError):
        return None


def _log(msg: str) -> None:
    if os.environ.get("PFBMAX_METADATA_DEBUG"):
        print(f"[pfbmax.metadata] {msg}", file=sys.stderr)


# ---------------------------------------------------------------------------
# Default LLM: shared pfbmax backbone, with a stdlib inline fallback so this
# module never blocks on pfbmax/llm.py availability.
# ---------------------------------------------------------------------------

def _default_llm() -> Any:
    try:
        from pfbmax.llm import LLM  # shared gpt-4o-mini backbone
        return LLM()
    except Exception:
        try:
            return _InlineMiniLLM()
        except Exception:
            return None


class _InlineMiniLLM:
    """Minimal gpt-4o-mini JSON caller (stdlib urllib), fallback only.

    Reads OPENAI_API_KEY from the environment or iris_asta/.env. Never used
    when ``pfbmax.llm.LLM`` imports (the normal case); unit tests inject
    fakes and never construct this.
    """

    _ENDPOINT = "https://api.openai.com/v1/chat/completions"

    def __init__(self, api_key: str | None = None, model: str = "gpt-4o-mini"):
        self.model = model
        self.api_key = api_key or self._find_key()
        if not self.api_key:
            raise RuntimeError("OPENAI_API_KEY not found (env or iris_asta/.env)")

    @staticmethod
    def _find_key() -> str | None:
        key = os.environ.get("OPENAI_API_KEY", "").strip()
        if key:
            return key
        env_path = os.path.join(_IRIS_DIR, ".env")
        try:
            with open(env_path, encoding="utf-8-sig") as f:
                for line in f:
                    line = line.strip()
                    if line.startswith("OPENAI_API_KEY="):
                        value = line.split("=", 1)[1].strip().strip("'\"")
                        if value:
                            return value
        except OSError:
            pass
        return None

    def json(self, system: str, user: str, max_tokens: int = 900) -> dict | None:
        payload = {
            "model": self.model,
            "messages": [
                {"role": "system", "content": system},
                {"role": "user", "content": user},
            ],
            "temperature": 0,
            "max_tokens": int(max_tokens),
            "response_format": {"type": "json_object"},
        }
        request = urllib.request.Request(
            self._ENDPOINT,
            data=json.dumps(payload).encode("utf-8"),
            headers={
                "Content-Type": "application/json",
                "Authorization": f"Bearer {self.api_key}",
            },
            method="POST",
        )
        for attempt in range(3):
            try:
                with urllib.request.urlopen(request, timeout=90) as resp:
                    body = json.loads(resp.read().decode("utf-8", "replace"))
                text = ((body.get("choices") or [{}])[0].get("message") or {}).get("content") or ""
                try:
                    obj = json.loads(text)
                    return obj if isinstance(obj, dict) else None
                except ValueError:
                    start, end = text.find("{"), text.rfind("}")
                    if 0 <= start < end:
                        try:
                            obj = json.loads(text[start:end + 1])
                            return obj if isinstance(obj, dict) else None
                        except ValueError:
                            return None
                    return None
            except Exception:
                if attempt == 2:
                    return None
        return None

# ---------------------------------------------------------------------------
# Venue acronym matching (additive override of the read-only executor's rule)
# ---------------------------------------------------------------------------

_VENUE_STOP = {"of", "for", "and", "the", "on", "in", "at", "to", "a", "an",
               "acm", "ieee", "sigplan", "sigmod", "sigir", "annual", "joint"}


def _acronym_matches_venue(acronym: str, venue: str) -> bool:
    """True when ``acronym`` is spelled by the initials of a contiguous run of
    the venue's content words.

    Corpora store conferences under their expanded names, so a substring test
    for the acronym fails exactly where it matters. Measured: one query asks
    for SPLASH papers, and the gold venue string is "ACM SIGPLAN International
    Conference on Systems, Programming, Languages and Applications: Software
    for Humanity". That spells S-P-L-A-S-H across six consecutive content
    words with no literal "SPLASH" anywhere, and the query scored 0.000
    purely on that.

    Contiguity keeps it honest (initials in order, no skipping), and the
    4-character floor stops short acronyms matching by chance.
    """
    a = "".join(ch for ch in (acronym or "").lower() if ch.isalnum())
    if len(a) < 4 or not venue:
        return False
    words = [w for w in re.split(r"[^A-Za-z0-9]+", venue.lower()) if w]
    words = [w for w in words if w not in _VENUE_STOP]
    initials = "".join(w[0] for w in words)
    return a in initials


def _install_venue_acronym_support() -> None:
    """Teach the read-only executor about acronym venues, additively.

    We never edit iris_asta; this wraps its matcher in-process so the rule can
    only ever ACCEPT more venues, never reject one it used to accept.
    """
    try:
        from iris_asta.solvers import pfb as _pfb
    except Exception:
        return
    if getattr(_pfb, "_pfbmax_venue_patch", False):
        return
    _orig = _pfb._venue_matches

    def _patched(paper_venue: str, wanted: list) -> bool:
        if _orig(paper_venue, wanted):
            return True
        return any(_acronym_matches_venue(w, paper_venue or "") for w in (wanted or []))

    _pfb._venue_matches = _patched
    _pfb._pfbmax_venue_patch = True


#: chosen authorId -> alias sibling ids, filled by the resolver patch.
_ALIAS_SIBLINGS: dict = {}


def _alias_norm(text: str) -> str:
    """Lowercased, punctuation-flattened name for alias comparison."""
    return " ".join(
        "".join(ch if ch.isalnum() else " " for ch in (text or "").lower()).split())


def _name_is_alias(candidate: str, wanted: str) -> bool:
    """Is ``candidate`` an initialism alias of ``wanted``?

    ("F. Surname" is an alias of "Firstname Surname".) Semantic Scholar
    fragments one researcher across several author records, and the
    FULL-NAME record is frequently the small one: the exact-name records
    may hold a handful of papers while the initialism record holds
    hundreds, including the ones a query wants. Preferring an exact
    string match therefore resolves the wrong record and empties the
    author pool.

    Alias test: same surname, and the candidate's leading token is either the
    same first name or its initial. Deliberately narrow -- it never merges two
    different surnames, so it cannot pull in an unrelated author.
    """
    c = _alias_norm(candidate).split()
    w = _alias_norm(wanted).split()
    if len(c) < 2 or len(w) < 2:
        return False
    if c[-1] != w[-1]:
        return False
    return c[0] == w[0] or (len(c[0]) == 1 and c[0] == w[0][:1])


def _install_author_alias_support() -> None:
    """Let author resolution see initialism aliases, additively.

    iris_asta stays read-only; this wraps its resolver in-process. It only
    ever WIDENS the candidate pool (exact matches remain eligible) and still
    picks by paperCount, so it cannot resolve an author the original rule
    would have resolved better.
    """
    try:
        from iris_asta.solvers import pfb as _pfb
    except Exception:
        return
    if getattr(_pfb, "_pfbmax_author_patch", False):
        return
    _orig = _pfb._resolve_author

    def _patched(client, name):
        try:
            candidates = client.search_authors_by_name(name) or []
        except Exception:
            return _orig(client, name)
        candidates = [c for c in candidates if isinstance(c, dict)]
        if not candidates:
            return _orig(client, name)
        want = _alias_norm(name)
        compat = [c for c in candidates
                  if _alias_norm(str(c.get("name") or "")) == want
                  or _name_is_alias(str(c.get("name") or ""), name)]
        if not compat:
            return _orig(client, name)

        def _count(rec):
            try:
                return int(rec.get("paperCount") or 0)
            except (TypeError, ValueError):
                return 0

        best = max(compat, key=_count)
        # Remember the sibling records so the author's paper set can be
        # DE-FRAGMENTED later. S2 splits one researcher across records
        # (a full name may hold 88 + 12 + 4 papers across records, plus a
        # reversed-order record with 1), and a
        # membership test built from only the largest record silently fails
        # for every paper living in a fragment.
        best_id = best.get("authorId") or best.get("id")
        if best_id is not None:
            sibs = [c.get("authorId") or c.get("id") for c in compat]
            _ALIAS_SIBLINGS[str(best_id)] = [str(x) for x in sibs
                                             if x is not None and str(x) != str(best_id)]
        return best

    _pfb._resolve_author = _patched

    _orig_papers = _pfb._author_papers

    def _patched_papers(client, author_id, date_range=None):
        papers = list(_orig_papers(client, author_id, date_range) or [])
        # MEASURED: de-fragmenting the author's paper set fixes RECALL
        # completely (recall 0.31 -> 1.00 on the affected query, all 16
        # golds found) but explodes the candidate pool, and exact-set F1
        # punishes that harder than it rewards the recall: p 0.71 -> 0.089
        # (179 returned for 16 gold), F1 0.435 -> 0.164. The downstream
        # filters (JournalArticle / min_citations / exclude_author) cannot
        # discriminate the widened pool, so this stays OFF until they can.
        # Opt in with PFBMAX_AUTHOR_DEFRAG=1.
        if os.environ.get("PFBMAX_AUTHOR_DEFRAG", "").strip() not in ("1", "true", "yes"):
            return papers
        seen = {id(p) for p in papers}
        for sib in _ALIAS_SIBLINGS.get(str(author_id), []):
            try:
                for p in _orig_papers(client, sib, date_range) or []:
                    if id(p) not in seen:
                        seen.add(id(p))
                        papers.append(p)
            except Exception:
                continue
        return papers

    _pfb._author_papers = _patched_papers
    _pfb._pfbmax_author_patch = True


_install_venue_acronym_support()
_install_author_alias_support()