src/endopath/taxonomy.py

"""The taxonomy registry engine (T9, #35): storage is convention-neutral, a convention is a projection.

One extraction is stored once, against the primitive the report stated, and any convention is computed
on demand. Adding a convention is a matter of adding registry rows, never of re-extracting a report.

The registry is six normalized CSVs under `data/taxonomy/`:

- `coding_systems.csv`   the systems a concept can belong to (FIGO 1988/2009/2023, CAP, WHO, ...).
- `dimensions.csv`       one row per checklist question, tagged `value_kind` (coded, quantity, derived).
- `concepts.csv`         the coded values, keyed `(system_id, dimension, code)`, because
                          `figo_endo/1988.IB` and `figo_endo/2009.IB` are different concepts that
                          share a code and denote opposite disease extents.
- `crosswalk.csv`        concept-to-concept mappings with a SKOS predicate plus `conflict` (six values).
- `crosswalk_precondition.csv`  conditions that gate a mapping. Rows sharing `(mapping_id, dimension)`
                          are OR'd; across dimensions they are AND'd.
- `derivations.csv`      the `formula`, `threshold`, and `membership` rules. No free-text formula in a
                          cell: a formula or membership row names a `formula_id` this module implements,
                          and a threshold row is data (an operator and a threshold scoped by a system).

The public surface: `resolve`, `codes`, `crosswalk`, `derive`, `project`, `derive_stage`, plus
`infer_edition` and `determinability_matrix`. `derive_stage` is the per-edition FIGO stage algorithm; it
projects from canonical assertions and never from another edition's stage label. The stage rows in
`crosswalk.csv` are the algorithm's golden oracle and documentation, not its runtime engine.

The two invariants this module serves:

1. **Numbers stay numbers.** A percentage and a category derive from the millimetre quantities by a
   named rule. `ratio_percent` is registered once (one `formula_id`) and bound twice (two rows), to the
   myometrial pair and to the cervical pair, so the cervical binding cannot contaminate the myometrial
   one. A report stating "4 foci" is `substantial` LVSI under FIGO 2009 (>=4) and `focal` under FIGO
   2023 (>=5): storing the count keeps both answers reachable, storing the category discards one.
2. **No value without its evidence.** Every registry row cites an `extraction_id` in
   `data/taxonomy/extractions.csv`, or a `docs/taxonomy.md` section for a rule the registry currently
   evidences in prose. A citation that resolves to neither fails the loader.

## How B1 (`endopath.validation`, #97) plugs in

`endopath.validation` recomputes a derived concept to catch an extraction error. It owns no rule logic:
it imports `RECOMPUTE_BY_CONCEPT` from this module, keyed by concept id, positional in dependency-graph
order. Every rule that a per-case record can recompute from its graph inputs alone lives there:
`ratio_percent` (bound to the myometrial and cervical percents), the myometrial-invasion-category
threshold, `grade_binary`, and `histotype_aggressiveness`. Two rules need a projection context the
per-case record does not carry and so are not in that table: `lvsi_extent` needs a coding system (the
substantial cut-off is 3, 4, or 5 vessels by system), and `figo_stage` needs a FIGO edition (a bare
stage code is edition-keyed). A caller projects those through `derive(..., under=...)` and
`derive_stage(..., edition)` when the context is supplied.
"""

from __future__ import annotations

import csv
import re
from collections import defaultdict
from dataclasses import dataclass
from enum import Enum
from functools import lru_cache
from pathlib import Path
from typing import Any, Callable, Mapping, Optional, Sequence

from pydantic import BaseModel, ConfigDict

REPO_ROOT = Path(__file__).resolve().parents[2]
TAXONOMY_DIR = REPO_ROOT / "data" / "taxonomy"
TAXONOMY_DOC_PATH = REPO_ROOT / "docs" / "taxonomy.md"

CODING_SYSTEMS_PATH = TAXONOMY_DIR / "coding_systems.csv"
DIMENSIONS_PATH = TAXONOMY_DIR / "dimensions.csv"
CONCEPTS_PATH = TAXONOMY_DIR / "concepts.csv"
CROSSWALK_PATH = TAXONOMY_DIR / "crosswalk.csv"
PRECONDITION_PATH = TAXONOMY_DIR / "crosswalk_precondition.csv"
DERIVATIONS_PATH = TAXONOMY_DIR / "derivations.csv"
EXTRACTIONS_PATH = TAXONOMY_DIR / "extractions.csv"


class RegistryError(ValueError):
    """The registry did not load: an unknown system or dimension, a dangling concept reference, a
    predicate or rule_kind outside its closed vocabulary, a precondition on an unknown value, or a
    citation that resolves to neither an extraction nor a docs/taxonomy.md section."""


# --------------------------------------------------------------------------- closed vocabularies


class Predicate(str, Enum):
    """The crosswalk `predicate` column: the five SKOS values plus `conflict` (dec_015, #85). The
    five are the SKOS vocabulary, not the RDF stack. `conflict` marks a row where two standards each
    state a rule for the same question and the rules disagree, distinct from `noMatch` (no counterpart
    is named) and from `relatedMatch` (overlap without contradiction)."""

    EXACT_MATCH = "exactMatch"
    BROAD_MATCH = "broadMatch"
    NARROW_MATCH = "narrowMatch"
    RELATED_MATCH = "relatedMatch"
    NO_MATCH = "noMatch"
    CONFLICT = "conflict"


class RuleKind(str, Enum):
    """The derivation kinds, closed at three so no free-text formula enters a cell.

    `formula` computes a quantity from quantities via a named formula with typed args. `threshold`
    computes a coded value from a quantity, scoped by `under_system`. `membership` computes a coded or
    bool value from coded values, partial by design (a value the source does not determine returns
    `indeterminate`)."""

    FORMULA = "formula"
    THRESHOLD = "threshold"
    MEMBERSHIP = "membership"


class ValueKind(str, Enum):
    """A dimension's answer kind. `coded` picks from a value set (the concepts), `quantity` is a
    number, `derived` is computed by a rule."""

    CODED = "coded"
    QUANTITY = "quantity"
    DERIVED = "derived"


class Certainty(str, Enum):
    CERTAIN = "certain"
    UNCERTAIN = "uncertain"


# --------------------------------------------------------------------------- row models


class CodingSystem(BaseModel):
    model_config = ConfigDict(frozen=True)
    system_id: str
    title: str
    kind: str
    edition_of: str = ""
    note: str = ""


class Dimension(BaseModel):
    model_config = ConfigDict(frozen=True)
    dimension: str
    value_kind: ValueKind
    closure: str = ""
    note: str = ""


class Concept(BaseModel):
    """One coded value, keyed `(system_id, dimension, code)`."""

    model_config = ConfigDict(frozen=True)
    concept_id: str
    system_id: str
    dimension: str
    code: str
    label: str = ""
    extraction_id: str = ""
    note: str = ""

    @property
    def key(self) -> tuple[str, str, str]:
        return (self.system_id, self.dimension, self.code)


class CrosswalkMapping(BaseModel):
    model_config = ConfigDict(frozen=True)
    mapping_id: str
    source_concept_id: str
    target_concept_id: str
    predicate: Predicate
    certainty: Certainty = Certainty.CERTAIN
    rationale: str = ""
    extraction_id: str = ""


class Precondition(BaseModel):
    model_config = ConfigDict(frozen=True)
    precondition_id: str
    mapping_id: str
    dimension: str
    operator: str
    value_code: str


class Derivation(BaseModel):
    model_config = ConfigDict(frozen=True)
    derivation_id: str
    rule_kind: RuleKind
    target_dimension: str
    target_concept_id: str = ""
    source_dimension: str = ""
    operator: str = ""
    threshold: str = ""
    unit: str = ""
    under_system: str = ""
    formula_id: str = ""
    arg_numerator: str = ""
    arg_denominator: str = ""
    extraction_id: str = ""


# --------------------------------------------------------------------------- projection results


@dataclass(frozen=True)
class Stage:
    """A determined stage: a code within a coding system, optionally carrying a molecular modifier."""

    code: str
    system_id: str
    modifier: Optional[str] = None


@dataclass(frozen=True)
class Constrained:
    """The asserted values narrow the answer to a set and no further (docs/taxonomy.md section 6.2).
    The set carries the answer, so it is never collapsed to a single value or to a blank."""

    candidates: tuple[str, ...]
    reason: str
    dimension: str = "figo_stage"
    system_id: Optional[str] = None


@dataclass(frozen=True)
class Indeterminate:
    """A required assertion is missing, or the report states the value cannot be assessed."""

    reason: str
    dimension: str = "figo_stage"
    system_id: Optional[str] = None


# A sentinel for LVSI present but unquantified: a foci count the report never states.
UNQUANTIFIED = "__lvsi_present_unquantified__"

# The determinable value set of each derived (or gated coded) dimension. Used to validate a
# threshold row's target and a precondition's value_code, and to enumerate a `constrained` set.
DERIVED_VALUE_SETS: dict[str, frozenset[str]] = {
    "myometrial_invasion_category": frozenset({"none", "less_than_half", "half_or_more"}),
    "lvsi_extent": frozenset({"negative", "focal", "substantial"}),
    "histotype_aggressiveness": frozenset({"non_aggressive", "aggressive", "indeterminate"}),
    "grade_binary": frozenset({"low", "high"}),
    "cervical_stromal_invasion": frozenset({"present", "absent", "cannot_be_assessed"}),
}


# --------------------------------------------------------------------------- the named rules


def ratio_percent(
    numerator: Any = None,
    denominator: Any = None,
    *,
    depth: Any = None,
    thickness: Any = None,
) -> Optional[float]:
    """Depth over thickness as a percentage (docs/taxonomy.md section 2). Registered once and bound
    twice, to the myometrial and to the cervical quantities. Returns None on a zero or non-numeric
    denominator, which the caller reports as an unrecomputable derivation rather than a value."""
    n = depth if depth is not None else numerator
    d = thickness if thickness is not None else denominator
    try:
        d = float(d)
        if d == 0:
            return None
        return float(n) / d * 100.0
    except (TypeError, ValueError):
        return None


def grade_binary(figo_grade: Any) -> Optional[str]:
    """The membership rule {1,2} low, {3} high (docs/taxonomy.md section 2)."""
    if figo_grade in ("1", "2", 1, 2):
        return "low"
    if figo_grade in ("3", 3):
        return "high"
    return None


def histotype_aggressiveness(histologic_type: Any, figo_grade: Any = None) -> Optional[str]:
    """FIGO 2023's aggressiveness membership (ex_figo2023_aggressive_types). Non-aggressive is
    low-grade (G1/G2) endometrioid; aggressive is every other combination. The rule is partial:
    endometrioid carcinoma of unknown grade is `indeterminate` rather than aggressive."""
    if histologic_type in (None, ""):
        return None
    if histologic_type != "endometrioid":
        return "aggressive"
    if figo_grade in ("1", "2", 1, 2):
        return "non_aggressive"
    if figo_grade in ("3", 3):
        return "aggressive"
    return "indeterminate"


def cervical_membership(source: Any) -> Optional[str]:
    """Reduce a source cervical assessment to the exclusive `cervical_stromal_invasion` value set,
    partial by design. A source that could not assess the cervix (for example CAP's
    `cannot_assess_supracervical`) determines neither `present` nor `absent`, so the rule returns None
    and `derive` reports `indeterminate`."""
    if source in (None, ""):
        return None
    s = str(source).lower()
    if "cannot" in s or "assess" in s or "supracervical" in s:
        return None
    if "present" in s or "invasion" in s or s == "present":
        return "present"
    if "absent" in s or "no_" in s or "not" in s or s == "absent":
        return "absent"
    return None


# formula_id -> the callable a `formula` or `membership` derivation row names. A threshold row carries
# no formula_id: its logic is the data in the row (an operator and a threshold), evaluated generically.
NAMED_RULES: dict[str, Callable[..., Any]] = {
    "ratio_percent": ratio_percent,
    "grade_binary": grade_binary,
    "histotype_aggressiveness": histotype_aggressiveness,
    "cervical_stromal_invasion": cervical_membership,
}


# --------------------------------------------------------------------------- the registry


class Registry:
    """The loaded, cross-validated registry: the six tables plus the indices the engine reads."""

    def __init__(
        self,
        coding_systems: list[CodingSystem],
        dimensions: list[Dimension],
        concepts: list[Concept],
        crosswalk: list[CrosswalkMapping],
        preconditions: list[Precondition],
        derivations: list[Derivation],
    ) -> None:
        self.coding_systems = coding_systems
        self.dimensions = dimensions
        self.concepts = concepts
        self.crosswalk = crosswalk
        self.preconditions = preconditions
        self.derivations = derivations

        self.system_by_id = {s.system_id: s for s in coding_systems}
        self.dimension_by_id = {d.dimension: d for d in dimensions}
        self.concept_by_id = {c.concept_id: c for c in concepts}
        self.concept_by_key: dict[tuple[str, str, str], Concept] = {c.key: c for c in concepts}
        self.mapping_by_id = {m.mapping_id: m for m in crosswalk}
        self.preconditions_by_mapping: dict[str, list[Precondition]] = defaultdict(list)
        for pc in preconditions:
            self.preconditions_by_mapping[pc.mapping_id].append(pc)

        self._validate()

    def _validate(self) -> None:
        extraction_ids = _load_extraction_ids()
        doc_sections = _load_doc_sections()

        def cite_ok(cited: str) -> bool:
            if not cited:
                return False
            if cited in extraction_ids:
                return True
            m = re.match(r"^docs/taxonomy\.md#(\d+(?:\.\d+)*)$", cited)
            return bool(m) and m.group(1) in doc_sections

        # concepts: unique ids and keys, a known system, a resolvable citation.
        if len(self.concept_by_id) != len(self.concepts):
            raise RegistryError("concept_id is not unique across the concept table")
        if len(self.concept_by_key) != len(self.concepts):
            raise RegistryError("concept key (system_id, dimension, code) is not unique")
        for c in self.concepts:
            if c.system_id not in self.system_by_id:
                raise RegistryError(f"{c.concept_id} names system {c.system_id!r} not in coding_systems.csv")

        # crosswalk: known concepts, a resolvable citation. Predicate is enum-checked on load.
        for m in self.crosswalk:
            if m.source_concept_id not in self.concept_by_id:
                raise RegistryError(f"{m.mapping_id} source {m.source_concept_id!r} is not a concept")
            if m.target_concept_id not in self.concept_by_id:
                raise RegistryError(f"{m.mapping_id} target {m.target_concept_id!r} is not a concept")
            if not cite_ok(m.extraction_id):
                raise RegistryError(f"{m.mapping_id} cites {m.extraction_id!r}, which resolves to no evidence")

        # preconditions: a real mapping, a known dimension, a value in that dimension's set.
        for pc in self.preconditions:
            if pc.mapping_id not in self.mapping_by_id:
                raise RegistryError(f"{pc.precondition_id} gates unknown mapping {pc.mapping_id!r}")
            if pc.dimension not in self.dimension_by_id:
                raise RegistryError(f"{pc.precondition_id} tests unknown dimension {pc.dimension!r}")
            if pc.operator != "=":
                raise RegistryError(f"{pc.precondition_id} uses unsupported operator {pc.operator!r}")
            allowed = DERIVED_VALUE_SETS.get(pc.dimension)
            if allowed is not None and pc.value_code not in allowed:
                raise RegistryError(
                    f"{pc.precondition_id} tests {pc.dimension}={pc.value_code!r}, not in {sorted(allowed)}"
                )

        # derivations: known dimensions and systems, a named rule for formula/membership, a
        # threshold that names a valid band, a resolvable citation.
        for d in self.derivations:
            if d.target_dimension not in self.dimension_by_id:
                raise RegistryError(f"{d.derivation_id} targets unknown dimension {d.target_dimension!r}")
            if d.source_dimension and d.source_dimension not in self.dimension_by_id:
                raise RegistryError(f"{d.derivation_id} reads unknown dimension {d.source_dimension!r}")
            if d.under_system and d.under_system not in self.system_by_id:
                raise RegistryError(f"{d.derivation_id} is scoped by unknown system {d.under_system!r}")
            if d.rule_kind in (RuleKind.FORMULA, RuleKind.MEMBERSHIP):
                if d.formula_id not in NAMED_RULES:
                    raise RegistryError(
                        f"{d.derivation_id} is a {d.rule_kind.value} row naming formula_id "
                        f"{d.formula_id!r}, which this module does not implement"
                    )
            if d.rule_kind is RuleKind.FORMULA and not (d.arg_numerator and d.arg_denominator):
                raise RegistryError(f"{d.derivation_id} is a formula row missing its typed args")
            if d.rule_kind is RuleKind.THRESHOLD:
                if not d.operator or d.threshold == "":
                    raise RegistryError(f"{d.derivation_id} is a threshold row missing an operator or threshold")
                allowed = DERIVED_VALUE_SETS.get(d.target_dimension)
                if allowed is not None and d.target_concept_id not in allowed:
                    raise RegistryError(
                        f"{d.derivation_id} maps to band {d.target_concept_id!r}, not in {sorted(allowed)}"
                    )
            if not cite_ok(d.extraction_id):
                raise RegistryError(f"{d.derivation_id} cites {d.extraction_id!r}, which resolves to no evidence")


def _read_rows(path: Path) -> list[dict]:
    with path.open(newline="", encoding="utf-8") as fh:
        return list(csv.DictReader(fh))


def _load_extraction_ids() -> frozenset[str]:
    return frozenset(r["extraction_id"] for r in _read_rows(EXTRACTIONS_PATH))


def _load_doc_sections() -> frozenset[str]:
    heading = re.compile(r"^#{2,4}\s+(\d+(?:\.\d+)*)\.?\s")
    sections: set[str] = set()
    for line in TAXONOMY_DOC_PATH.read_text(encoding="utf-8").splitlines():
        m = heading.match(line)
        if m:
            sections.add(m.group(1))
    return frozenset(sections)


@lru_cache(maxsize=1)
def load_registry() -> Registry:
    """Load and cross-validate the six registry tables. Cached: the CSVs do not change at runtime."""
    return Registry(
        coding_systems=[CodingSystem(**r) for r in _read_rows(CODING_SYSTEMS_PATH)],
        dimensions=[Dimension(**r) for r in _read_rows(DIMENSIONS_PATH)],
        concepts=[Concept(**r) for r in _read_rows(CONCEPTS_PATH)],
        crosswalk=[CrosswalkMapping(**r) for r in _read_rows(CROSSWALK_PATH)],
        preconditions=[Precondition(**r) for r in _read_rows(PRECONDITION_PATH)],
        derivations=[Derivation(**r) for r in _read_rows(DERIVATIONS_PATH)],
    )


# --------------------------------------------------------------------------- resolve / codes


def resolve(code: str, system_id: str, dimension: Optional[str] = None) -> Concept:
    """The concept a code names within a coding system. `figo_endo/1988.IB` and `figo_endo/2009.IB`
    resolve to different concepts, because a stage code has no meaning without its edition (section 4).
    Pass `dimension` when a system reuses a code across dimensions."""
    reg = load_registry()
    system_id = _system_id(system_id)
    matches = [
        c
        for c in reg.concepts
        if c.system_id == system_id and c.code == code and (dimension is None or c.dimension == dimension)
    ]
    if not matches:
        raise LookupError(f"no concept for code {code!r} in {system_id!r}" + (f"/{dimension}" if dimension else ""))
    if len(matches) > 1:
        raise LookupError(f"code {code!r} in {system_id!r} is ambiguous across dimensions; pass a dimension")
    return matches[0]


def codes(system_id: str, dimension: Optional[str] = None) -> list[Concept]:
    """The concepts a coding system carries, optionally within one dimension."""
    reg = load_registry()
    system_id = _system_id(system_id)
    return [
        c
        for c in reg.concepts
        if c.system_id == system_id and (dimension is None or c.dimension == dimension)
    ]


# --------------------------------------------------------------------------- derive


def _to_float(value: Any) -> Optional[float]:
    try:
        return float(value)
    except (TypeError, ValueError):
        return None


def _cmp(value: float, operator: str, threshold: float) -> bool:
    if operator == ">=":
        return value >= threshold
    if operator == ">":
        return value > threshold
    if operator == "<=":
        return value <= threshold
    if operator == "<":
        return value < threshold
    if operator == "=":
        return value == threshold
    raise RegistryError(f"unsupported threshold operator {operator!r}")


def _threshold_category(dimension: str, value: Any, under: str) -> Optional[str]:
    """Evaluate the threshold bands for a dimension under a coding system. Each band is a lower bound
    (`>` or `>=` a threshold); the value falls into the band with the greatest satisfied bound. A tie
    at one threshold prefers the strict `>` band (the narrower one), which is how `none` (>= 0) and
    `less_than_half` (> 0) share the boundary 0 without overlapping."""
    fvalue = _to_float(value)
    if fvalue is None:
        return None
    reg = load_registry()
    under = _system_id(under) if under else ""
    best_rank: Optional[tuple[float, int]] = None
    best_band: Optional[str] = None
    for d in reg.derivations:
        if d.rule_kind is not RuleKind.THRESHOLD or d.target_dimension != dimension:
            continue
        if (d.under_system or "") != under:
            continue
        thr = float(d.threshold)
        if _cmp(fvalue, d.operator, thr):
            rank = (thr, 1 if d.operator == ">" else 0)
            if best_rank is None or rank > best_rank:
                best_rank, best_band = rank, d.target_concept_id
    return best_band


_INPUT_ALIASES: dict[str, str] = {
    "foci_count": "lvsi_foci_count",
    "percent": "myometrial_invasion_percent",
    "depth": "invasion_depth",
    "thickness": "myometrial_thickness",
    "source": "cervical_stromal_invasion",
}


def derive(dimension: str, *, under: Optional[str] = None, **inputs: Any):
    """Compute a derived dimension by its registered rule.

    - A `formula` dimension returns the number its `ratio_percent` binding produces.
    - A `threshold` dimension returns the band code, `Constrained` over its determinable set when the
      quantity is absent (a count the report never states), or `Indeterminate`.
    - A `membership` dimension returns the mapped value, or `Indeterminate` when the source does not
      determine it (partial by design).

    Inputs are passed by their source-dimension name; a short alias is also accepted (`foci_count` for
    `lvsi_foci_count`, `depth`/`thickness` for the ratio args, `source` for a membership input).
    """
    reg = load_registry()
    rows = [d for d in reg.derivations if d.target_dimension == dimension]
    if not rows:
        raise RegistryError(f"{dimension!r} is not a derived dimension in derivations.csv")

    def pull(name: str) -> Any:
        if name in inputs:
            return inputs[name]
        for alias, target in _INPUT_ALIASES.items():
            if target == name and alias in inputs:
                return inputs[alias]
        return None

    kind = rows[0].rule_kind
    if kind is RuleKind.FORMULA:
        row = rows[0]
        rule = NAMED_RULES[row.formula_id]
        return rule(pull(row.arg_numerator), pull(row.arg_denominator))

    if kind is RuleKind.THRESHOLD:
        source_dim = rows[0].source_dimension
        value = pull(source_dim)
        if value is None:
            value = inputs.get("value")
        if value is None:
            determinable = tuple(sorted(DERIVED_VALUE_SETS.get(dimension, frozenset()) - {"negative"}))
            return Constrained(determinable, f"no {source_dim} stated", dimension)
        band = _threshold_category(dimension, value, under or "")
        if band is None:
            return Indeterminate(f"{dimension} could not be evaluated from {source_dim}={value!r}", dimension)
        return band

    # membership
    row = rows[0]
    rule = NAMED_RULES[row.formula_id]
    if row.formula_id == "histotype_aggressiveness":
        result = rule(pull("histologic_type"), pull("figo_grade"))
    else:
        value = pull(row.source_dimension)
        if value is None:
            value = inputs.get("value")
        result = rule(value)
    if result is None:
        return Indeterminate(f"{dimension} is not determined by the source value", dimension)
    return result


# --------------------------------------------------------------------------- crosswalk + preconditions


def _match(value: Any, operator: str, value_code: str) -> bool:
    if operator == "=":
        return value == value_code
    return False


def preconditions_satisfied(mapping_id: str, assertions: Mapping[str, Any]) -> bool:
    """Rows sharing `(mapping_id, dimension)` are OR'd; across dimensions they are AND'd. A mapping
    with no preconditions is always satisfied. A dimension the assertions do not carry fails its
    precondition, so a mapping is never fired on a value the case never asserted."""
    reg = load_registry()
    rows = reg.preconditions_by_mapping.get(mapping_id, [])
    if not rows:
        return True
    by_dim: dict[str, list[Precondition]] = defaultdict(list)
    for pc in rows:
        by_dim[pc.dimension].append(pc)
    for dimension, group in by_dim.items():
        value = assertions.get(dimension)
        if not any(_match(value, pc.operator, pc.value_code) for pc in group):
            return False
    return True


def crosswalk(
    source_concept_id: str,
    target_system: Optional[str] = None,
    assertions: Optional[Mapping[str, Any]] = None,
) -> list[CrosswalkMapping]:
    """The mappings from a source concept: optionally restricted to a target system, and optionally to
    those whose preconditions the assertions satisfy. With no assertions, preconditions are ignored and
    every mapping from the source is returned."""
    reg = load_registry()
    if target_system is not None:
        target_system = _system_id(target_system)
    out: list[CrosswalkMapping] = []
    for m in reg.crosswalk:
        if m.source_concept_id != source_concept_id:
            continue
        if target_system is not None:
            target = reg.concept_by_id.get(m.target_concept_id)
            if target is None or target.system_id != target_system:
                continue
        if assertions is not None and not preconditions_satisfied(m.mapping_id, assertions):
            continue
        out.append(m)
    return out


# --------------------------------------------------------------------------- edition inference


_EDITION_SYSTEMS = {
    "1988": "figo_endo/1988",
    "1995": "figo_endo/1988",  # the 1995 revision shares the 1988 corpus sub-staging
    "2009": "figo_endo/2009",
    "2023": "figo_endo/2023",
    "figo_endo/1988": "figo_endo/1988",
    "figo_endo/2009": "figo_endo/2009",
    "figo_endo/2023": "figo_endo/2023",
}


def _system_id(edition: Any) -> str:
    key = str(edition).strip()
    if key in _EDITION_SYSTEMS:
        return _EDITION_SYSTEMS[key]
    if key in load_registry().system_by_id:
        return key
    raise RegistryError(f"unknown coding system or edition {edition!r}")


def _normalize_stage_code(code: Any) -> Optional[str]:
    if code in (None, ""):
        return None
    c = str(code).strip().upper().replace("P", "").replace("T1", "I")
    # "T1B" -> after replace P/T1: "1B" -> "IB"; already-figo "IB" untouched.
    c = c.replace("1A", "IA").replace("1B", "IB").replace("1C", "IC")
    return c


# --------------------------------------------------------------------------- reported-stage normalization
#
# A reported `pathologic_stage` value is a free-text string a pathologist wrote: "pT1a", "pT1a, pN0 pMX",
# "pT2N0", "1B", "IA", "T1a N1". Stored raw, its confirmed-value distribution counts formatting variants
# and combined TNM tokens as distinct categories (the dashboard's "Other (N values)" fold, #165). This
# parses the string deterministically into its TNM components and, when no FIGO code is written, the FIGO
# stage the local T plus any nodal or metastatic upstaging implies. It never guesses: a string that yields
# no component leaves every field None (`parsed` is False), so the caller routes it to `indeterminate`
# rather than a wrong category. The verbatim string is kept, so normalization derives, never overwrites
# (invariant 2). Reading the scanned page into this string is the model's job; parsing the formal codes out
# of it is deterministic work.


@dataclass(frozen=True)
class ReportedStage:
    """A reported pathologic-stage string parsed into normalized components, the verbatim string kept."""

    verbatim: str
    t_category: Optional[str]  # AJCC pT, lower-cased suffix: "T1a", "T1b", "T2", "T3a", "T3b", "T4"
    n_category: Optional[str]  # nodal: "N0", "N1", "N1mi", "N1a", "N2", "N2mi", "N2a", "NX"
    m_category: Optional[str]  # "M0", "M1", "MX"
    figo_stage: Optional[str]  # normalized FIGO code: "IA", "IB", "II", "IIIA", "IIIC1", "IIIC2", "IVB", ...

    @property
    def parsed(self) -> bool:
        """False when nothing was recognized, so the caller codes the value `indeterminate`, not a guess."""
        return any((self.t_category, self.n_category, self.m_category, self.figo_stage))


# TNM tokens. A leading "p" (pathologic) is optional; no trailing boundary, so a run-together "pT2N0"
# still yields T2 and N0. The patterns are specific enough (T1-4, N0-2, M0/1/X) that a stage string does
# not false-match. Case-insensitive.
_T_TOKEN = re.compile(r"P?T([1-4][ABC]?)", re.IGNORECASE)
_N_TOKEN = re.compile(r"P?N([0-2](?:MI|[AB])?|X)", re.IGNORECASE)
_M_TOKEN = re.compile(r"P?M([01X])", re.IGNORECASE)
# A written FIGO stage: Roman (IA..IVB, IIIC1/IIIC2), or arabic shorthand ("1B"). Boundaries keep a lone
# "I" inside a word from reading as stage I.
_FIGO_ROMAN = re.compile(r"(?<![A-Z0-9])(IV|III|II|I)([ABC])?([12])?(?![A-Z0-9])", re.IGNORECASE)
_FIGO_ARABIC = re.compile(r"(?<![A-Z0-9])([1-4])([ABC])(?![A-Z0-9])")
_ARABIC_TO_ROMAN = {"1": "I", "2": "II", "3": "III", "4": "IV"}
# The endometrial pT-to-FIGO local correspondence (node-negative, M0), applied only when the string carries
# no explicit FIGO code. Nodal and metastatic upstaging is layered on top. T1c is the FIGO-1988 IC code
# (deep myometrial invasion); an ambiguous legacy code without a clear single-edition FIGO stage is left
# out, so it stays verbatim rather than mapping to an invented category.
_T_LOCAL_FIGO = {
    "T1A": "IA",
    "T1B": "IB",
    "T1C": "IC",
    "T2": "II",
    "T3A": "IIIA",
    "T3B": "IIIB",
    "T4": "IVA",
}


def _canonical_tnm(token: Optional[str]) -> Optional[str]:
    """Lower-case a TNM suffix so a distribution groups on one form: "T1A" -> "T1a", "N1MI" -> "N1mi"."""
    if token is None:
        return None
    return token[:2] + token[2:].lower()


def _derive_figo_from_tnm(t: Optional[str], n: Optional[str], m: Optional[str]) -> Optional[str]:
    """The FIGO stage a reported TNM tuple implies, most-advanced component first: a distant metastasis is
    IVB, a positive node is IIIC (C1 pelvic, C2 para-aortic), otherwise the local pT correspondence."""
    if m == "M1":
        return "IVB"
    if n and n.upper().startswith("N1"):
        return "IIIC1"
    if n and n.upper().startswith("N2"):
        return "IIIC2"
    if t:
        return _T_LOCAL_FIGO.get(t.upper())
    return None


def parse_reported_stage(raw: Any) -> ReportedStage:
    """Parse a reported pathologic-stage string into normalized TNM components and a FIGO stage (#165).

    An explicit FIGO code in the string wins; otherwise the FIGO stage is derived from the TNM tuple by
    `_derive_figo_from_tnm`. A string with no recognizable component returns an all-None `ReportedStage`
    (`parsed` is False), never a guessed category.
    """
    verbatim = "" if raw is None else str(raw)
    s = verbatim.upper()

    def _first(rx: "re.Pattern[str]", prefix: str) -> Optional[str]:
        match = rx.search(s)
        return _canonical_tnm(f"{prefix}{match.group(1)}") if match else None

    t = _first(_T_TOKEN, "T")
    n = _first(_N_TOKEN, "N")
    m = _first(_M_TOKEN, "M")

    figo: Optional[str] = None
    roman = _FIGO_ROMAN.search(s)
    if roman:
        figo = roman.group(1) + (roman.group(2) or "") + (roman.group(3) or "")
    else:
        arabic = _FIGO_ARABIC.search(s)
        if arabic:
            figo = _ARABIC_TO_ROMAN[arabic.group(1)] + arabic.group(2)
    if figo is None:
        figo = _derive_figo_from_tnm(t, n, m)

    return ReportedStage(verbatim=verbatim, t_category=t, n_category=n, m_category=m, figo_stage=figo)


# A reported nodal string's positive/examined count, e.g. "0/17" or "0 of 16".
_NODE_COUNT = re.compile(r"(\d+)\s*(?:/|\bOF\b)\s*\d+")


def parse_reported_nodal(raw: Any) -> Optional[str]:
    """Normalize a reported regional-lymph-node value to a `pn_category` code (#165), or None.

    An explicit pN code in the string is taken as written. "pNX" or "cannot be assessed" is `pNX`. A
    zero positive count ("0/17 lymph nodes positive") or prose stating no nodal disease ("No regional
    lymph node metastasis") is `pN0`. A positive count returns None rather than a category, because
    `pN1` (pelvic) versus `pN2` (para-aortic) needs the anatomic location the count alone does not carry:
    the honest failure is `indeterminate`, not an invented category.
    """
    if raw is None:
        return None
    s = str(raw).upper()
    token = _N_TOKEN.search(s)
    if token:
        return "p" + str(_canonical_tnm(f"N{token.group(1)}"))
    if "CANNOT" in s or "NOT ASSESS" in s:
        return "pNX"
    count = _NODE_COUNT.search(s)
    if count:
        return "pN0" if int(count.group(1)) == 0 else None
    # Prose that states no nodal disease and no positive node: "All lymph nodes negative", "No regional
    # lymph node metastasis". "POSITIVE" absent guards against a mixed report ("... 1/3 positive").
    if "POSITIVE" not in s and ("NEGATIVE" in s or ("NODE" in s and "NO " in s)):
        return "pN0"
    return None


def _normalize_extent(extent: Any) -> Optional[str]:
    if extent in (None, ""):
        return None
    e = str(extent).strip().lower()
    if e in ("less_than_half", "none") or "less than" in e or "less-than" in e or "inner" in e:
        return "less_than_half"
    if e in ("half_or_more",) or "half or more" in e or "more than half" in e or "equal to or more" in e or "outer" in e:
        return "half_or_more"
    return None


def infer_edition(extent: Any = None, code: Any = None) -> Optional[str]:
    """Infer the FIGO edition of a stage code from the extent phrase beside it (docs section 4, 5.3).
    `T1b`/`IB` denote opposite extents across editions: a report saying "less than one-half" beside
    `T1b` is FIGO 1988, because only 1988 labels less-than-half invasion `IB`. `IC` and `IIIC1`/`IIIC2`
    pin the edition by themselves. Returns a system id, or None when the pair does not determine one."""
    c = _normalize_stage_code(code)
    if c == "IC":
        return "figo_endo/1988"
    if c in ("IIIC1", "IIIC2"):
        return "figo_endo/2009"
    e = _normalize_extent(extent)
    if e == "less_than_half":
        if c == "IB":
            return "figo_endo/1988"
        if c == "IA":
            return "figo_endo/2009"
    if e == "half_or_more":
        if c == "IC":
            return "figo_endo/1988"
        if c == "IB":
            return "figo_endo/2009"
    return None


# --------------------------------------------------------------------------- derive_stage


_STAGE_I_OR_II = frozenset({"IA", "IA1", "IA2", "IA3", "IB", "IC", "II", "IIA", "IIB", "IIC"})
_ANATOMIC_2023 = {"none": "IA1", "less_than_half": "IA2", "half_or_more": "IB"}


def _node_positive(pn: Any) -> bool:
    return pn in ("pN1mi", "pN1a", "pN2mi", "pN2a")


def _iiic(pn: Any) -> str:
    return "IIIC2" if pn in ("pN2mi", "pN2a") else "IIIC1"


def _myo_category(a: Mapping[str, Any]) -> Optional[str]:
    if a.get("myometrial_invasion_category") is not None:
        return a["myometrial_invasion_category"]
    pct = a.get("myometrial_invasion_percent")
    if pct is None and a.get("invasion_depth") is not None and a.get("myometrial_thickness") is not None:
        pct = ratio_percent(a["invasion_depth"], a["myometrial_thickness"])
    if pct is not None:
        return _threshold_category("myometrial_invasion_category", pct, "")
    return None


def _lvsi_extent(a: Mapping[str, Any], edition: str) -> Optional[str]:
    if a.get("lvsi_extent") is not None:
        return a["lvsi_extent"]
    foci = a.get("lvsi_foci_count")
    if foci is not None:
        return _threshold_category("lvsi_extent", foci, edition)
    status = a.get("lvsi_status")
    if status in ("not_identified", "negative", "absent"):
        return "negative"
    if status == "present":
        return UNQUANTIFIED
    return None  # unstated: treated as no/focal LVSI by the caller


def _aggressiveness(a: Mapping[str, Any]) -> Optional[str]:
    if a.get("histotype_aggressiveness") is not None:
        return a["histotype_aggressiveness"]
    if a.get("histologic_type") is not None:
        return histotype_aggressiveness(a["histologic_type"], a.get("figo_grade"))
    return None


def _molecular_modifier(a: Mapping[str, Any]) -> Optional[str]:
    """The FIGO 2023 molecular modifier, from a confirmed ProMisE class only (dec_019). A raw
    `tcga_class` of `copy_number_high` does not upstage on its own: the copy-number-high to p53abn
    surrogate carries a ~5% error, so the p53abn stage is offered as a candidate a reviewer accepts,
    never applied automatically."""
    pc = a.get("promise_class")
    if pc in ("pole_mutated", "polemut"):
        return "POLEmut"
    if pc in ("p53_abnormal", "p53abn"):
        return "p53abn"
    return None


def _stage_1988_2009(a: Mapping[str, Any], edition: str):
    pn = a.get("pn_category")
    if _node_positive(pn):
        return Stage("IIIC" if edition == "figo_endo/1988" else _iiic(pn), edition)
    cervical = a.get("cervical_stromal_invasion")
    if cervical == "cannot_be_assessed":
        return Indeterminate("cervical stromal invasion cannot be assessed", "figo_stage", edition)
    if cervical == "present":
        return Stage("II", edition)
    cat = _myo_category(a)
    if cat is None:
        return Indeterminate("myometrial invasion category is unknown", "figo_stage", edition)
    if edition == "figo_endo/1988":
        return Stage({"none": "IA", "less_than_half": "IB", "half_or_more": "IC"}[cat], edition)
    return Stage("IA" if cat in ("none", "less_than_half") else "IB", edition)


def _stage_2023(a: Mapping[str, Any], edition: str):
    pn = a.get("pn_category")
    if _node_positive(pn):
        # The i/ii micrometastasis/macrometastasis split is unmeasurable in this corpus; the base
        # IIIC1/IIIC2 is determined by which nodal group is involved.
        return Stage(_iiic(pn), edition)
    cervical = a.get("cervical_stromal_invasion")
    if cervical == "cannot_be_assessed":
        return Indeterminate("cervical stromal invasion cannot be assessed", "figo_stage", edition)
    aggr = _aggressiveness(a)
    if aggr is None or aggr == "indeterminate":
        return Indeterminate("histotype aggressiveness cannot be determined (grade unknown)", "figo_stage", edition)
    cat = _myo_category(a)

    if aggr == "aggressive":
        if cat is None and cervical != "present":
            return Indeterminate("myometrial invasion presence unknown for an aggressive histotype", "figo_stage", edition)
        any_mi = cat is not None and cat != "none"
        base = Stage("IIC" if (any_mi or cervical == "present") else "IC", edition)
    else:  # non-aggressive
        if cervical == "present":
            base = Stage("IIA", edition)
        else:
            extent = _lvsi_extent(a, edition)
            if extent == "substantial":
                base = Stage("IIB", edition)
            elif extent == UNQUANTIFIED:
                if cat is None:
                    return Indeterminate("myometrial invasion category is unknown", "figo_stage", edition)
                anatomic = _ANATOMIC_2023[cat]
                return Constrained(
                    (anatomic, "IIB"),
                    f"LVSI is present but unquantified: focal keeps {anatomic}, substantial raises to IIB",
                    "figo_stage",
                    edition,
                )
            else:  # negative, focal, or unstated (treated as no/focal)
                if cat is None:
                    return Indeterminate("myometrial invasion category is unknown", "figo_stage", edition)
                base = Stage(_ANATOMIC_2023[cat], edition)

    modifier = _molecular_modifier(a)
    if isinstance(base, Stage) and base.code in _STAGE_I_OR_II and modifier:
        if modifier == "POLEmut":
            return Stage("IAmPOLEmut", edition, modifier="POLEmut")
        if modifier == "p53abn":
            return Stage("IICmp53abn", edition, modifier="p53abn")
    return base


def derive_stage(assertions: Mapping[str, Any], edition: Any):
    """Project a FIGO stage from canonical assertions, never from another edition's stage label.

    Returns a `Stage`, a `Constrained(candidates, reason)` when the assertions narrow the answer to a
    set and no further (the IB-versus-IIB pair when LVSI is present but unquantified), or an
    `Indeterminate(reason)` when a required assertion is missing or the report states a value cannot be
    assessed. An unstated cervical or nodal finding is read as not involved, the standard reading that
    a pathology report states positive findings; a value the report explicitly marks unassessable is
    `Indeterminate`, never assumed negative.
    """
    edition = _system_id(edition)
    if edition == "figo_endo/2023":
        return _stage_2023(assertions, edition)
    return _stage_1988_2009(assertions, edition)


def project(assertions: Mapping[str, Any], dimension: str = "figo_stage", under: Any = None):
    """Project a derived dimension into a target convention. For `figo_stage` this is `derive_stage`
    under the target edition; for any other derived dimension it is `derive` under the target system.
    Returns a value, `Stage`, `Constrained`, or `Indeterminate`."""
    if dimension == "figo_stage":
        return derive_stage(assertions, under)
    return derive(dimension, under=(str(under) if under is not None else None), **dict(assertions))


# --------------------------------------------------------------------------- B1 interface


def _category_from_percent(percent: Any) -> Optional[str]:
    if _to_float(percent) is None:
        return None
    return _threshold_category("myometrial_invasion_category", percent, "")


# concept_id -> the recompute B1 (`endopath.validation`) calls, positional in dependency-graph order.
# Only the rules a per-case record can recompute from its graph inputs alone appear here. `lvsi_extent`
# (needs a coding system) and `figo_stage` (needs an edition) are projected through `derive`/
# `derive_stage` with that context, not recomputed in the editionless per-case pass.
RECOMPUTE_BY_CONCEPT: dict[str, Callable[..., Optional[Any]]] = {
    "myometrial_invasion_percent": lambda depth, thickness: ratio_percent(depth, thickness),
    "cervical_wall_percent": lambda depth, thickness: ratio_percent(depth, thickness),
    "myometrial_invasion_category": _category_from_percent,
    "grade_binary": grade_binary,
    "histotype_aggressiveness": histotype_aggressiveness,
}


def recompute(concept_id: str, inputs: Sequence[Any]) -> Optional[Any]:
    """B1's clean interface: recompute a derived concept from its dependency-graph inputs, positional.
    Returns the recomputed value, or None when this module does not recompute the concept in an
    editionless per-case pass (`lvsi_extent`, `figo_stage`)."""
    fn = RECOMPUTE_BY_CONCEPT.get(concept_id)
    if fn is None:
        return None
    return fn(*inputs)


# --------------------------------------------------------------------------- determinability (section 9)


@dataclass(frozen=True)
class DeterminabilityRow:
    code: str
    inputs_beyond_anatomy: str
    corpus_availability: str
    verdict: str


# The section-9 matrix. Emitted by scripts/report_determinability.py, not hand-written into a doc.
# The verdict is a function of what a FIGO 2023 code needs beyond anatomy and whether the corpus
# supplies it: `determined` (anatomy and histology suffice), `determined_by_join` (needs the TCGA
# molecular join), `constrained` (the projector enumerates candidates and states why), `unmeasurable`.
_DETERMINABILITY: tuple[DeterminabilityRow, ...] = (
    DeterminabilityRow("III*/IV* (except IIIC*i/ii)", "none", "high", "determined"),
    DeterminabilityRow("IIIC1i/ii, IIIC2i/ii", "micro vs macro deposit", "0.4% / 0%", "unmeasurable"),
    DeterminabilityRow("IA1", "non-aggressive, confined to endometrium/polyp", "grade 82.6%", "determined"),
    DeterminabilityRow("IA2", "non-aggressive, MI<50%, no/focal LVSI", "MI 14.8%; LVSI 85.5%", "determined_or_constrained"),
    DeterminabilityRow("IA3", "low-grade endometrioid, uterus and ovary", "ovary/adnexa 96.5%", "determined"),
    DeterminabilityRow("IB", "non-aggressive, MI>=50%, no/focal LVSI", "as above", "determined_or_constrained"),
    DeterminabilityRow("IC", "aggressive histotype, no MI", "grade + histotype", "determined"),
    DeterminabilityRow("IIA", "non-aggressive, cervical stromal invasion", "cervical stroma 11.5%", "determined"),
    DeterminabilityRow("IIB", "non-aggressive, substantial LVSI", "foci count 0", "constrained"),
    DeterminabilityRow("IIC", "aggressive, any MI", "grade + histotype", "determined"),
    DeterminabilityRow("IAmPOLEmut", "POLEmut, stage I/II", "49 cases via join", "determined_by_join"),
    DeterminabilityRow("IICmp53abn", "p53abn, stage I/II", "162 cases via join", "determined_by_join"),
)


def determinability_matrix() -> tuple[DeterminabilityRow, ...]:
    """The section-9 determinability matrix: one verdict per FIGO 2023 code group. `IB` against `IIB`
    is the only genuinely constrained pair, and only where LVSI is present but unquantified."""
    return _DETERMINABILITY