src/endopath/validation.py

"""B1: validate a per-case extraction against the confirmed A2 schema (#97).

The per-case extraction (ColPali + rules + LLM, #99) populates one value per concept.
This pass reads that extraction and the confirmed concept list (#91/#95) and, for each
concept, emits a validation block:

- a **type** check (the value is the concept's primitive type),
- an **enum** check (a categorical value is one of the concept's registered codes),
- a **cardinality** check (a single-valued concept does not carry a list of values), and
- for a derived concept, a **derivation recompute**: the concept is recomputed from its
  input concepts by its named rule (#35), and the result is compared to the extracted value.

A field whose block passes is promoted from needs-review to `confirmed_eligible`. A person
still confirms it: no value counts as data without human confirmation (docs/prd.md 8.3).
A field whose block fails stays `needs_review` with the reason named, never silently.

Two facts the compiled concept (`mapping.CompiledConcept`) does not carry are supplied here:

- the **value shape** (`_VALUE_SHAPES`): the primitive type and, for a categorical concept,
  the closed set of codes. It is hand-transcribed from `docs/taxonomy.md` section 1 (asserted
  values) and section 2 (derived values), the same way `mapping._ENRICHMENT` transcribes the
  three facts the dependency graph omits, and `value_shapes()` validates it covers exactly the
  compiled concept set so a concept added to the registry fails until its shape is supplied.
- the **recompute** of a derived concept (`_RECOMPUTE`): the concept-keyed map from
  `endopath.taxonomy.RECOMPUTE_BY_CONCEPT`. The rule logic lives in the taxonomy registry (#35),
  not here: this pass owns no formula. `ratio_percent` (the myometrial and cervical ratios), the
  myometrial-invasion-category threshold, `grade_binary`, and `histotype_aggressiveness` recompute
  from a record's dependency-graph inputs alone and so are in that map. `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); neither is carried by the editionless per-case
  record, so they are projected through `taxonomy.derive`/`taxonomy.derive_stage` with that context
  rather than recomputed here, and their block reports `rule_not_implemented` in this pass so a value
  that depends on them is not promoted on an unverified derivation.

The two invariants this pass serves:

1. Numbers stay numbers. A derived category (`myometrial_invasion_category`) is recomputed
   from the primitive (`myometrial_invasion_percent`, itself recomputed from the two millimetre
   quantities), never trusted as extracted. A category that disagrees with its recomputed value
   fails validation.
2. No value without its evidence. The extracted evidence span is carried through the block for
   the confirmation card (#98) to render and for #49 to resolve to a page region. A value that
   is absent must name a coded reason (docs/prd.md 4.3); a missing value with no reason fails,
   because silence is never the response to a miss.
"""

from __future__ import annotations

import math
from dataclasses import dataclass
from enum import Enum
from typing import Any, Callable, Iterable, Optional

from pydantic import BaseModel, ConfigDict, Field

from endopath import taxonomy
from endopath.mapping import (
    AssertionType,
    Cardinality,
    CompiledConcept,
    compile_concepts,
)
from endopath.schema import EvidenceSpan


class ValidationConfigError(ValueError):
    """The validator is misconfigured against the registry: the value-shape table and the
    compiled concept list have drifted, or a record names a concept the schema does not carry."""


# ------------------------------------------------------------------ value shapes


class ValueType(str, Enum):
    """The primitive type of a concept's value, mirroring the JSON-schema types
    `src/endopath/fields.py` already uses for the CAP checklist."""

    NUMBER = "number"
    INTEGER = "integer"
    STRING = "string"
    BOOLEAN = "boolean"


@dataclass(frozen=True)
class ValueShape:
    """The type and, for a categorical concept, the closed code set a value must satisfy.

    `enum` is `None` for a quantity (a number carries no value set) and for the concepts whose
    value set `docs/taxonomy.md` evidences in prose rather than as an explicit closed list
    (the raw POLE assay, and the edition-keyed FIGO stage code); enum membership is then
    reported `not_applicable` rather than checked against a fabricated set.
    """

    value_type: ValueType
    enum: Optional[frozenset[str]] = None
    minimum: Optional[float] = None
    maximum: Optional[float] = None


# concept_id -> value shape, transcribed from docs/taxonomy.md sections 1 and 2. The comment on
# each block cites the section. `value_shapes()` checks this covers exactly the compiled concept
# set, so a concept added to the dependency graph fails the pass until its shape lands here.
_VALUE_SHAPES: dict[str, ValueShape] = {
    # --- Histotype (section 1.1, 1.2; section 2) ---------------------------------------
    "histologic_type": ValueShape(
        ValueType.STRING,
        frozenset(
            {
                "endometrioid",
                "serous",
                "clear_cell",
                "undifferentiated",
                "dedifferentiated",
                "carcinosarcoma",
                "mesonephric_like",
                "squamous_cell",
                "gastric_type",
                "small_cell_neuroendocrine",
                "large_cell_neuroendocrine",
                "mixed",
                "other_nos",
            }
        ),
    ),
    "figo_grade": ValueShape(ValueType.STRING, frozenset({"1", "2", "3"})),
    "grade_binary": ValueShape(ValueType.STRING, frozenset({"low", "high"})),
    "histotype_aggressiveness": ValueShape(
        ValueType.STRING, frozenset({"non_aggressive", "aggressive", "indeterminate"})
    ),
    # --- Myometrial and cervical invasion (section 1.3, 1.4; section 2) -----------------
    # Numbers stay numbers: the millimetre quantities are numbers, the percentage and the
    # category derive from them.
    "invasion_depth": ValueShape(ValueType.NUMBER, minimum=0.0),
    "myometrial_thickness": ValueShape(ValueType.NUMBER, minimum=0.0),
    "myometrial_invasion_percent": ValueShape(ValueType.NUMBER, minimum=0.0, maximum=100.0),
    "myometrial_invasion_category": ValueShape(
        ValueType.STRING, frozenset({"none", "less_than_half", "half_or_more"})
    ),
    "cervical_stromal_invasion": ValueShape(
        ValueType.STRING, frozenset({"present", "absent", "cannot_be_assessed"})
    ),
    "cervical_invasion_depth": ValueShape(ValueType.NUMBER, minimum=0.0),
    "cervical_wall_thickness": ValueShape(ValueType.NUMBER, minimum=0.0),
    "cervical_wall_percent": ValueShape(ValueType.NUMBER, minimum=0.0, maximum=100.0),
    # --- Lymphovascular space invasion (section 1.5; section 2, 5) ----------------------
    "lvsi_status": ValueShape(
        ValueType.STRING,
        frozenset({"present", "not_identified", "suspicious", "cannot_be_determined"}),
    ),
    "lvsi_foci_count": ValueShape(ValueType.INTEGER, minimum=0),
    "lvsi_extent": ValueShape(ValueType.STRING, frozenset({"focal", "substantial"})),
    # --- Regional lymph nodes (section 1.6) --------------------------------------------
    "pn_category": ValueShape(
        ValueType.STRING,
        frozenset({"pN0", "pN0_i_plus", "pN1mi", "pN1a", "pN2mi", "pN2a"}),
    ),
    # --- Molecular (section 1.7) -------------------------------------------------------
    # The raw POLE assay has no explicit closed value set in section 1.7 (the section states the
    # ProMisE precedence in prose): enum is left open rather than fabricated. MMR and p53 read
    # into the explicit sets the interpretation notes give.
    "pole_exonuclease_mutation": ValueShape(ValueType.STRING),
    "mmr_ihc": ValueShape(ValueType.STRING, frozenset({"intact", "lost", "subclonal_loss"})),
    "p53_ihc": ValueShape(ValueType.STRING, frozenset({"wild_type", "abnormal"})),
    "promise_class": ValueShape(
        ValueType.STRING,
        frozenset({"pole_mutated", "mmr_deficient", "p53_abnormal", "nsmp", "double_classifier"}),
    ),
    "tcga_class": ValueShape(
        ValueType.STRING,
        frozenset(
            {"pole_ultramutated", "msi_hypermutated", "copy_number_low", "copy_number_high"}
        ),
    ),
    # --- Stage (section 4) -------------------------------------------------------------
    # A FIGO stage code is edition-keyed (`1988.IB` and `2009.IB` are different concepts,
    # section 4), so no single closed set covers it: enum is left open.
    "figo_stage": ValueShape(ValueType.STRING),
}


def value_shapes() -> dict[str, ValueShape]:
    """The value shape of every compiled concept, validated to cover exactly the concept set.

    Raises ValidationConfigError if the table and the compiled concept list have drifted, the
    same drift guard `mapping.compile_concepts` runs over its enrichment table.
    """
    concept_ids = {concept.concept_id for concept in compile_concepts()}
    missing = sorted(concept_ids - set(_VALUE_SHAPES))
    extra = sorted(set(_VALUE_SHAPES) - concept_ids)
    if missing or extra:
        raise ValidationConfigError(
            "value-shape table and the compiled concept list disagree on the concept set: "
            f"missing shape for {missing}; shape names unknown concepts {extra}"
        )
    return dict(_VALUE_SHAPES)


# ------------------------------------------------------------------ recompute of derived concepts

# concept_id -> the recompute of its named rule, owned by the taxonomy registry (#35) and imported
# rather than duplicated here. `ratio_percent` is bound twice (once per organ), so the map is keyed by
# concept. `histotype_aggressiveness` recomputes from a record's inputs alone and is verified here now.
# `lvsi_extent` (needs a coding system) and `figo_stage` (needs a FIGO edition) carry a context the
# per-case record does not, so they are absent from this map and report `rule_not_implemented`; a caller
# projects them through `taxonomy.derive`/`taxonomy.derive_stage`.
_RECOMPUTE: dict[str, Callable[..., Optional[Any]]] = dict(taxonomy.RECOMPUTE_BY_CONCEPT)


# ------------------------------------------------------------------ the extraction record contract


class FieldState(str, Enum):
    """A field is never simply present or blank (CLAUDE.md, the field-state corollary; docs/prd.md
    4.3). It carries a concrete value, or a coded reason it does not."""

    PRESENT = "present"
    NOT_APPLICABLE = "not_applicable"
    NOT_STATED = "not_stated"
    INDETERMINATE = "indeterminate"
    DETERMINED_BY_JOIN = "determined_by_join"
    CONSTRAINED = "constrained"


_ABSENCE_STATES = frozenset(state for state in FieldState if state is not FieldState.PRESENT)


class ExtractedValue(BaseModel):
    """One concept's extraction: the value the pipeline drafted (or a coded absence), and the
    evidence span that justified it. This is the per-case extraction contract B1 consumes (#99
    produces it) and the row the validation block is attached to."""

    model_config = ConfigDict(frozen=True)

    concept_id: str
    value: Optional[Any] = None
    state: FieldState = FieldState.PRESENT
    evidence: Optional[EvidenceSpan] = None
    # For a CONSTRAINED field: the set the evidence narrowed the answer to, no further.
    candidates: Optional[list[str]] = None


# ------------------------------------------------------------------ the validation block


class CheckName(str, Enum):
    TYPE = "type"
    ENUM = "enum"
    CARDINALITY = "cardinality"
    STATE = "state"


class CheckStatus(str, Enum):
    PASS = "pass"
    FAIL = "fail"
    NOT_APPLICABLE = "not_applicable"


class DerivationStatus(str, Enum):
    """The result of recomputing a derived concept and comparing it to the extracted value."""

    NOT_DERIVED = "not_derived"  # the concept is asserted; nothing to recompute
    MATCHED = "matched"  # recomputed value agrees with the extracted value
    MISMATCHED = "mismatched"  # recomputed value disagrees: the extracted value is not trusted
    INPUTS_ABSENT = "inputs_absent"  # a required input is missing, so the rule cannot recompute
    RULE_NOT_IMPLEMENTED = "rule_not_implemented"  # the named rule is not recomputed here (#35)
    NOT_CHECKED = "not_checked"  # the value is a coded absence; no derivation to recompute


class Promotion(str, Enum):
    """Where the validation block leaves the field. `confirmed_eligible` is promoted from
    needs-review toward confirmation; a person still confirms it before it counts as data."""

    CONFIRMED_ELIGIBLE = "confirmed_eligible"
    NEEDS_REVIEW = "needs_review"


class Check(BaseModel):
    model_config = ConfigDict(frozen=True)

    name: CheckName
    status: CheckStatus
    detail: str = ""


class Derivation(BaseModel):
    model_config = ConfigDict(frozen=True)

    status: DerivationStatus
    rule: Optional[str] = None
    inputs: tuple[str, ...] = ()
    recomputed: Optional[Any] = None
    extracted: Optional[Any] = None
    detail: str = ""


class FieldValidation(BaseModel):
    """The per-field validation block: the shape checks, the derivation recompute, and the
    promotion the block earns. This is the data the confirmation card (#98) renders beside the
    value."""

    model_config = ConfigDict(frozen=True)

    concept_id: str
    assertion_type: AssertionType
    cardinality: Cardinality
    state: FieldState
    value: Optional[Any] = None
    evidence: Optional[EvidenceSpan] = None
    checks: list[Check] = Field(default_factory=list)
    derivation: Derivation
    promotion: Promotion
    reasons: list[str] = Field(default_factory=list)

    @property
    def passed(self) -> bool:
        return self.promotion is Promotion.CONFIRMED_ELIGIBLE


class ValidationRecord(BaseModel):
    """The whole per-case extraction after validation: one block per validated concept."""

    model_config = ConfigDict(frozen=True)

    fields: dict[str, FieldValidation]

    @property
    def all_confirmed_eligible(self) -> bool:
        return bool(self.fields) and all(block.passed for block in self.fields.values())

    def promoted(self) -> list[str]:
        return [cid for cid, block in self.fields.items() if block.passed]


def _is_multivalued(value: Any) -> bool:
    return isinstance(value, (list, tuple, set))


def _type_ok(value: Any, value_type: ValueType) -> bool:
    # bool is a subclass of int in Python; keep the two kinds distinct.
    if value_type is ValueType.BOOLEAN:
        return isinstance(value, bool)
    if value_type is ValueType.INTEGER:
        return isinstance(value, int) and not isinstance(value, bool)
    if value_type is ValueType.NUMBER:
        return isinstance(value, (int, float)) and not isinstance(value, bool)
    return isinstance(value, str)


def _values_agree(recomputed: Any, extracted: Any, shape: ValueShape) -> bool:
    if shape.value_type in (ValueType.NUMBER, ValueType.INTEGER):
        try:
            return math.isclose(float(recomputed), float(extracted), rel_tol=1e-6, abs_tol=0.5)
        except (TypeError, ValueError):
            return False
    return recomputed == extracted


def _shape_checks(value: Any, concept: CompiledConcept, shape: ValueShape) -> list[Check]:
    """The type, enum, and cardinality checks for a present value."""
    checks: list[Check] = []
    multivalued = _is_multivalued(value)

    # Cardinality: a single-valued concept must not carry a list of values.
    if concept.cardinality is Cardinality.ZERO_OR_MORE:
        checks.append(Check(name=CheckName.CARDINALITY, status=CheckStatus.PASS))
    elif multivalued:
        checks.append(
            Check(
                name=CheckName.CARDINALITY,
                status=CheckStatus.FAIL,
                detail=(
                    f"cardinality {concept.cardinality.value} allows one value, "
                    f"but {len(list(value))} were extracted"
                ),
            )
        )
    else:
        checks.append(Check(name=CheckName.CARDINALITY, status=CheckStatus.PASS))

    # Type: checked on the scalar. When cardinality already failed on a list, type and enum are
    # not applicable rather than run against the collection.
    if multivalued and concept.cardinality is not Cardinality.ZERO_OR_MORE:
        checks.append(Check(name=CheckName.TYPE, status=CheckStatus.NOT_APPLICABLE))
        checks.append(Check(name=CheckName.ENUM, status=CheckStatus.NOT_APPLICABLE))
        return checks

    scalar = value
    type_ok = _type_ok(scalar, shape.value_type)
    if not type_ok:
        checks.append(
            Check(
                name=CheckName.TYPE,
                status=CheckStatus.FAIL,
                detail=f"expected {shape.value_type.value}, got {type(scalar).__name__}",
            )
        )
    else:
        detail = ""
        if shape.minimum is not None and scalar < shape.minimum:
            type_ok = False
            detail = f"below minimum {shape.minimum}"
        elif shape.maximum is not None and scalar > shape.maximum:
            type_ok = False
            detail = f"above maximum {shape.maximum}"
        checks.append(
            Check(
                name=CheckName.TYPE,
                status=CheckStatus.PASS if type_ok else CheckStatus.FAIL,
                detail=detail,
            )
        )

    # Enum: only when the concept registers a closed set and the type held.
    if shape.enum is None:
        checks.append(
            Check(
                name=CheckName.ENUM,
                status=CheckStatus.NOT_APPLICABLE,
                detail="concept registers no closed value set",
            )
        )
    elif not type_ok:
        checks.append(Check(name=CheckName.ENUM, status=CheckStatus.NOT_APPLICABLE))
    elif scalar in shape.enum:
        checks.append(Check(name=CheckName.ENUM, status=CheckStatus.PASS))
    else:
        checks.append(
            Check(
                name=CheckName.ENUM,
                status=CheckStatus.FAIL,
                detail=f"{scalar!r} is not one of {sorted(shape.enum)}",
            )
        )
    return checks


def _recompute_derivation(
    concept: CompiledConcept,
    extracted_value: Any,
    record: dict[str, ExtractedValue],
    shape: ValueShape,
) -> Derivation:
    """Recompute a derived concept from its inputs and compare to the extracted value."""
    rule = concept.derived.rule if concept.derived else None
    inputs = concept.derived.inputs if concept.derived else ()
    fn = _RECOMPUTE.get(concept.concept_id)
    if fn is None:
        return Derivation(
            status=DerivationStatus.RULE_NOT_IMPLEMENTED,
            rule=rule,
            inputs=inputs,
            extracted=extracted_value,
            detail=(
                f"rule {rule!r} is implemented in endopath.taxonomy but needs a projection context "
                "this per-case pass does not carry (a coding system for lvsi_extent, a FIGO edition "
                "for figo_stage); project it through taxonomy.derive/derive_stage instead"
            ),
        )

    input_values: list[Any] = []
    for input_id in inputs:
        entry = record.get(input_id)
        if entry is None or entry.state is not FieldState.PRESENT or entry.value is None:
            return Derivation(
                status=DerivationStatus.INPUTS_ABSENT,
                rule=rule,
                inputs=inputs,
                extracted=extracted_value,
                detail=f"input {input_id!r} is absent, so the derivation cannot be recomputed",
            )
        input_values.append(entry.value)

    recomputed = fn(*input_values)
    if recomputed is None:
        return Derivation(
            status=DerivationStatus.INPUTS_ABSENT,
            rule=rule,
            inputs=inputs,
            extracted=extracted_value,
            detail="inputs present but the rule could not compute a value (e.g. zero denominator)",
        )

    agrees = _values_agree(recomputed, extracted_value, shape)
    return Derivation(
        status=DerivationStatus.MATCHED if agrees else DerivationStatus.MISMATCHED,
        rule=rule,
        inputs=inputs,
        recomputed=recomputed,
        extracted=extracted_value,
        detail=(
            ""
            if agrees
            else f"extracted {extracted_value!r} disagrees with recomputed {recomputed!r}"
        ),
    )


def validate_field(
    extracted: ExtractedValue,
    concept: CompiledConcept,
    shape: ValueShape,
    record: dict[str, ExtractedValue],
) -> FieldValidation:
    """Validate one extracted concept against its confirmed shape and, if derived, its recompute."""
    checks: list[Check] = []
    reasons: list[str] = []
    is_derived = concept.assertion_type is AssertionType.DERIVED

    # Coded absence: a value is either present or names a reason it is not (silence is never a
    # miss). Shape and derivation checks do not apply to a coded absence.
    if extracted.state in _ABSENCE_STATES:
        if extracted.value is not None:
            checks.append(
                Check(
                    name=CheckName.STATE,
                    status=CheckStatus.FAIL,
                    detail=f"state {extracted.state.value} carries a value {extracted.value!r}",
                )
            )
            reasons.append("a coded absence must not carry a value")
        else:
            checks.append(Check(name=CheckName.STATE, status=CheckStatus.PASS))
        derivation = Derivation(
            status=DerivationStatus.NOT_CHECKED,
            rule=concept.derived.rule if concept.derived else None,
            inputs=concept.derived.inputs if concept.derived else (),
            detail="value is a coded absence; no value to recompute",
        )
        promotion = Promotion.CONFIRMED_ELIGIBLE if not reasons else Promotion.NEEDS_REVIEW
        return FieldValidation(
            concept_id=concept.concept_id,
            assertion_type=concept.assertion_type,
            cardinality=concept.cardinality,
            state=extracted.state,
            value=None,
            evidence=extracted.evidence,
            checks=checks,
            derivation=derivation,
            promotion=promotion,
            reasons=reasons,
        )

    # State PRESENT but no value: silence with no coded reason.
    if extracted.value is None:
        checks.append(
            Check(
                name=CheckName.STATE,
                status=CheckStatus.FAIL,
                detail="value is absent but names no coded reason (section 4.3)",
            )
        )
        reasons.append(
            "This value is blank and no reason is recorded. Enter a value, or mark it not stated, "
            "not applicable, or indeterminate."
        )
        derivation = Derivation(
            status=DerivationStatus.NOT_DERIVED if not is_derived else DerivationStatus.INPUTS_ABSENT,
            rule=concept.derived.rule if concept.derived else None,
            inputs=concept.derived.inputs if concept.derived else (),
        )
        return FieldValidation(
            concept_id=concept.concept_id,
            assertion_type=concept.assertion_type,
            cardinality=concept.cardinality,
            state=extracted.state,
            value=None,
            evidence=extracted.evidence,
            checks=checks,
            derivation=derivation,
            promotion=Promotion.NEEDS_REVIEW,
            reasons=reasons,
        )

    # A present, concrete value: run the shape checks.
    checks.extend(_shape_checks(extracted.value, concept, shape))
    shape_ok = all(check.status is not CheckStatus.FAIL for check in checks)
    for check in checks:
        if check.status is CheckStatus.FAIL:
            reasons.append(f"{check.name.value} check failed: {check.detail}")

    # The derivation recompute for a derived concept.
    if is_derived:
        derivation = _recompute_derivation(concept, extracted.value, record, shape)
        if derivation.status is DerivationStatus.MISMATCHED:
            reasons.append(f"derivation disagrees: {derivation.detail}")
        elif derivation.status is DerivationStatus.INPUTS_ABSENT:
            reasons.append(f"derivation unverified: {derivation.detail}")
        elif derivation.status is DerivationStatus.RULE_NOT_IMPLEMENTED:
            reasons.append(f"derivation unverified: {derivation.detail}")
        derivation_ok = derivation.status is DerivationStatus.MATCHED
    else:
        derivation = Derivation(status=DerivationStatus.NOT_DERIVED)
        derivation_ok = True

    promotion = (
        Promotion.CONFIRMED_ELIGIBLE if (shape_ok and derivation_ok) else Promotion.NEEDS_REVIEW
    )
    return FieldValidation(
        concept_id=concept.concept_id,
        assertion_type=concept.assertion_type,
        cardinality=concept.cardinality,
        state=extracted.state,
        value=extracted.value,
        evidence=extracted.evidence,
        checks=checks,
        derivation=derivation,
        promotion=promotion,
        reasons=reasons,
    )


def _as_record(record: Iterable[ExtractedValue] | dict[str, ExtractedValue]) -> dict[str, ExtractedValue]:
    if isinstance(record, dict):
        return dict(record)
    by_id: dict[str, ExtractedValue] = {}
    for extracted in record:
        by_id[extracted.concept_id] = extracted
    return by_id


def validate_extraction(
    record: Iterable[ExtractedValue] | dict[str, ExtractedValue],
    *,
    concepts: Optional[list[CompiledConcept]] = None,
    shapes: Optional[dict[str, ValueShape]] = None,
) -> ValidationRecord:
    """Validate a per-case extraction against the confirmed A2 schema.

    `concepts` defaults to the whole compiled concept list (`mapping.compile_concepts`); pass the
    confirmed in-scope subset to restrict validation to it (see `confirmed_concept_ids`). Every
    concept the record names must be in the schema, or the record is invalid.
    """
    by_id = _as_record(record)
    concepts = concepts if concepts is not None else compile_concepts()
    shapes = shapes if shapes is not None else value_shapes()
    concept_by_id = {concept.concept_id: concept for concept in concepts}

    unknown = sorted(set(by_id) - set(concept_by_id))
    if unknown:
        raise ValidationConfigError(
            f"extraction names concepts not in the confirmed schema: {unknown}"
        )

    fields: dict[str, FieldValidation] = {}
    for concept_id, extracted in by_id.items():
        concept = concept_by_id[concept_id]
        shape = shapes.get(concept_id)
        if shape is None:
            raise ValidationConfigError(f"no value shape registered for concept {concept_id!r}")
        fields[concept_id] = validate_field(extracted, concept, shape, by_id)
    return ValidationRecord(fields=fields)


def confirmed_concept_ids(field_set: Iterable[dict]) -> frozenset[str]:
    """The distinct concepts a confirmed mapping (`mapping_review.mapping_review(...)['field_set']`,
    #95) resolves its source fields to. A caller passes the compiled concepts filtered to these,
    plus their derivation inputs, as the confirmed in-scope schema for `validate_extraction`."""
    return frozenset(
        row["concept_id"] for row in field_set if row.get("concept_id") is not None
    )