src/endopath/mapping_review.py

"""Assemble the mapping-review surface and its maieutic reviewer loop (#95).

The crosswalk surface (docs/prd.md sections 8.1, 3; design-guideline section 9.1) shows one registry's
fields on the left, the canonical concept categories on the right, and every drafted edge between them,
colored by category. A reviewer confirms the mapping once per corpus, resolving each unresolved or
low-confidence edge through a two-candidate maieutic loop.

This module is the read side that assembles that view. The edges are `crosswalk.reference_crosswalk()`
(the committed CAP-checklist instance in the `CrosswalkEdge` contract, #93), never authored here. It
joins concept labels and categories from the registry, marks the reviewer queue (`crosswalk.is_queued`),
drafts the maieutic question and two candidate resolutions per queued edge, and joins the confirmations
a reviewer has recorded.

The maieutic draft is deterministic here: it reads the drafted edge and the concept list to state the
question and the two candidates with their tradeoffs, so the surface and the fast test suite run with no
model call. The hosted-model draft plugs in at the same shape when it lands.

Where a confirmed edge's provenance lives is an open decision recorded on #93 (grow #35's crosswalk.csv
schema, or a dedicated table). This surface stores confirmations in the app database's
`mapping_confirmations` table (see `storage.py`) as a provisional home and does not resolve that
decision.
"""

from __future__ import annotations

import json
from datetime import datetime, timezone
from typing import Optional, Protocol, runtime_checkable

import anthropic
from sqlalchemy import Connection

from endopath import crosswalk, dictionary, fields, mapping, storage
from endopath.crosswalk import CrosswalkEdge, Relation
from endopath.llm_extraction import get_client

# The three dispositions a reviewer's decision on an edge takes (#103). A `confirmed` decision resolves
# the edge to a concept and relation. A `flagged` or `deferred` decision parks it: the concern is
# recorded in the ledger, but the edge stays unresolved and the mapping stays unconfirmed until a later
# decision resolves it.
DISPOSITIONS = frozenset({"confirmed", "flagged", "deferred"})

# How the confirmed answer was reached (#103), for the decision ledger. `candidate`: an offered maieutic
# resolution accepted as-is. `written`: a resolution the reviewer targeted and wrote themselves.
# `chat`: a resolution that came out of the edge chat. `flag` and `defer` accompany the parked
# dispositions. None on a decision recorded before this field existed.
DECIDED_VIA = frozenset({"candidate", "written", "chat", "flag", "defer"})


class MappingReviewError(ValueError):
    """A confirmation names an edge the crosswalk does not carry, a relation outside the vocabulary, a
    disposition or decided-via outside its vocabulary, or a resolution that violates the honesty rule (a
    resolved relation with no concept, an unresolved one that forces a link, or a concept outside the
    canonical list)."""


def _mapping_id(edge: CrosswalkEdge) -> str:
    """A stable id for one edge: its source standard and field. #35's `mapping_id`, keyed so a
    confirmation round-trips to the same edge."""
    return f"{edge.source_std}::{edge.source_field}"


def _source_labels() -> dict[str, str]:
    return {spec.name: spec.label for spec in fields.FIELDS}


def _licence_required() -> dict[str, bool]:
    """Per-variable licence-required flag (docs/prd.md section 4.4), from the built-in CAP dictionary."""
    return {v.name: v.licence_required for v in dictionary.builtin_cap_dictionary().variables}


def draft_review(
    edge: CrosswalkEdge,
    *,
    compiled: dict[str, mapping.CompiledConcept],
    labels: dict[str, str],
    source_labels: dict[str, str],
) -> dict:
    """The maieutic draft for one queued edge: the question it needs answered and two candidate
    resolutions with their tradeoffs, each a resolution the confirm endpoint accepts."""
    source_label = source_labels.get(edge.source_field, edge.source_field)

    if edge.is_unresolved:
        stage_id = next((cid for cid, c in compiled.items() if c.category == "stage"), None)
        candidates = [
            {
                "label": "Confirm out of scope",
                "concept_id": None,
                "relation": Relation.UNRESOLVED.value,
                "tradeoff": (
                    "The field stays captured on the report and cited, but projects into no concept, so "
                    "no export column or downstream rule reads it."
                ),
            }
        ]
        if stage_id is not None:
            stage_label = labels.get(stage_id, stage_id)
            candidates.append(
                {
                    "label": f"Map as an input to {stage_label}",
                    "concept_id": stage_id,
                    "relation": Relation.DERIVED_FROM.value,
                    "tradeoff": (
                        f"Treats the field as a {stage_label} input; correct only if this edition's "
                        "staging reads it, which FIGO endometrial staging does not."
                    ),
                }
            )
        question = (
            f"No canonical concept answers '{source_label}'. Confirm it out of scope, or map it as a "
            "staging input?"
        )
        return {"question": question, "candidates": candidates}

    concept_label = labels.get(edge.concept_id, edge.concept_id)
    category = compiled[edge.concept_id].category if edge.concept_id in compiled else None
    sibling_id = next(
        (cid for cid, c in compiled.items() if c.category == category and cid != edge.concept_id),
        None,
    )
    candidates = [
        {
            "label": "Accept the drafted mapping",
            "concept_id": edge.concept_id,
            "relation": edge.relation.value,
            "tradeoff": (
                f"Binds '{source_label}' to {concept_label} as {edge.relation.value}; correct if the "
                "report states what that concept derives from."
            ),
        }
    ]
    if sibling_id is not None:
        sibling_label = labels.get(sibling_id, sibling_id)
        candidates.append(
            {
                "label": f"Map to {sibling_label} instead",
                "concept_id": sibling_id,
                "relation": Relation.DERIVED_FROM.value,
                "tradeoff": (
                    f"Redirects the field to {sibling_label} in the same category; choose this if the "
                    f"source names that scheme rather than {concept_label}."
                ),
            }
        )
    else:
        candidates.append(
            {
                "label": "Confirm unresolved instead",
                "concept_id": None,
                "relation": Relation.UNRESOLVED.value,
                "tradeoff": (
                    "Routes the field to the queue as unmapped; choose this if no concept in the list "
                    "cleanly answers it."
                ),
            }
        )
    question = (
        f"'{source_label}' drafts a low-confidence {edge.relation.value} to {concept_label} at "
        f"{edge.confidence:.2f}. Accept it, or map it elsewhere?"
    )
    return {"question": question, "candidates": candidates}


def _enrich_components(
    rows: list[dict], compiled: dict[str, "mapping.CompiledConcept"], labels: dict[str, str]
) -> list[dict]:
    """Attach a label and category to each stored component edge of a split, for the review surface and
    the field set. The stored row carries only the concept_id, relation, and predicate."""
    return [
        {
            "concept_id": row["concept_id"],
            "concept_label": labels.get(row["concept_id"], row["concept_id"]),
            "relation": row["relation"],
            "predicate": row["predicate"],
            "category": compiled[row["concept_id"]].category
            if row["concept_id"] in compiled
            else None,
        }
        for row in rows
    ]


def _field_set_row(
    edge: CrosswalkEdge,
    confirmation: Optional[dict],
    source_labels: dict[str, str],
    licence_by_field: dict[str, bool],
    components: Optional[list[dict]] = None,
) -> dict:
    """One row of the confirmed field set (#96 reads this). A queued edge carries its reviewer and time
    from the confirmation; a high-confidence candidate is ratified by the whole-mapping confirmation. A
    field confirmed as a split carries the component concepts it fans out to alongside its headline."""
    if confirmation is not None:
        concept_id = confirmation["concept_id"]
        relation = confirmation["relation"]
        predicate = confirmation["predicate"]
        origin = "confirmed"
        confirmed_by = confirmation["confirmed_by"]
        confirmed_at = confirmation["confirmed_at"]
        licence_required = confirmation["licence_required"]
    else:
        concept_id = edge.concept_id
        relation = edge.relation.value
        predicate = edge.predicate
        origin = "candidate"
        confirmed_by = None
        confirmed_at = None
        licence_required = licence_by_field.get(edge.source_field, False)
    return {
        "source_field": edge.source_field,
        "source_label": source_labels.get(edge.source_field, edge.source_field),
        "concept_id": concept_id,
        "relation": relation,
        "predicate": predicate,
        "licence_required": licence_required,
        "confirmed_by": confirmed_by,
        "confirmed_at": confirmed_at,
        "origin": origin,
        "components": components or [],
    }


def _resolve_source(conn: Connection, dictionary_id: Optional[str]):
    """The crosswalk to review and its source-side labels. The built-in CAP checklist (the default)
    reviews the committed reference crosswalk with the checklist's own field labels; a submitted
    dictionary reviews the crosswalk its compilation job drafted (#108), with the dictionary's variable
    labels. Raises `MappingReviewError` when a named dictionary was never submitted or its crosswalk has
    not been drafted yet. Returns the crosswalk, the label and licence maps, the source-field rows, and
    the source title."""
    if dictionary_id is None or dictionary_id == dictionary.BUILTIN_DICTIONARY_ID:
        crosswalk_obj = crosswalk.reference_crosswalk()
        source_labels = _source_labels()
        licence_by_field = _licence_required()
        source_fields_out = [
            {"name": spec.name, "label": spec.label, "licence_required": licence_by_field.get(spec.name, False)}
            for spec in fields.FIELDS
        ]
        return crosswalk_obj, source_labels, licence_by_field, source_fields_out, dictionary.BUILTIN_DICTIONARY_TITLE

    data_dictionary = dictionary.get_registered(conn, dictionary_id)
    if data_dictionary is None:
        raise MappingReviewError(f"dictionary {dictionary_id!r} was not submitted")
    artifact = storage.get_submission_artifact(conn, dictionary_id, "crosswalk_draft")
    if artifact is None:
        raise MappingReviewError(
            f"the crosswalk for {dictionary_id!r} has not been drafted yet; its compilation job is still running"
        )
    crosswalk_obj = crosswalk.Crosswalk.model_validate_json(artifact["payload_json"])
    source_labels = {v.name: (v.label or v.name) for v in data_dictionary.variables}
    licence_by_field = {v.name: v.licence_required for v in data_dictionary.variables}
    source_fields_out = [
        {"name": v.name, "label": v.label or v.name, "licence_required": v.licence_required}
        for v in data_dictionary.variables
    ]
    return crosswalk_obj, source_labels, licence_by_field, source_fields_out, data_dictionary.title


def mapping_review(conn: Connection, dictionary_id: Optional[str] = None) -> dict:
    """The mapping-review payload (#95, #109): the crosswalk edges joined to concept labels and
    categories, the reviewer queue with its maieutic drafts, and the confirmations recorded so far.

    The default reviews the built-in CAP reference crosswalk. A submitted `dictionary_id` reviews the
    crosswalk its compilation job drafted (#108), with the dictionary's own variable labels.

    `confirmed` is true once every queued edge carries a confirmation, the point the whole mapping is
    confirmed. Only a confirmed mapping exposes `field_set`, the column spec the per-case pipeline and
    field-spec generation (#96) read.
    """
    crosswalk_obj, source_labels, licence_by_field, source_fields_out, source_title = _resolve_source(
        conn, dictionary_id
    )
    compiled = {c.concept_id: c for c in mapping.compile_concepts()}
    labels = crosswalk.concept_labels()
    confirmations = storage.list_mapping_confirmations(conn, crosswalk_obj.source_std)
    components_by_mid = storage.list_mapping_components(conn, crosswalk_obj.source_std)

    edges_out: list[dict] = []
    queued_total = 0
    queued_resolved = 0
    parked_total = 0
    for edge in crosswalk_obj.edges:
        mid = _mapping_id(edge)
        queued = crosswalk.is_queued(edge)
        confirmation = confirmations.get(mid)
        disposition = confirmation["disposition"] if confirmation else None
        resolved = disposition == "confirmed"
        parked = disposition in ("flagged", "deferred")
        category = compiled[edge.concept_id].category if edge.concept_id in compiled else None
        review = None
        if queued:
            queued_total += 1
            if resolved:
                queued_resolved += 1
            else:
                # A parked or still-open queued edge keeps its maieutic draft, so the reviewer can
                # resolve it from the offered candidates or write their own.
                review = draft_review(
                    edge, compiled=compiled, labels=labels, source_labels=source_labels
                )
        if parked:
            parked_total += 1
        components = _enrich_components(components_by_mid.get(mid, []), compiled, labels)
        edges_out.append(
            {
                "mapping_id": mid,
                "source_std": edge.source_std,
                "source_field": edge.source_field,
                "source_label": source_labels.get(edge.source_field, edge.source_field),
                "concept_id": edge.concept_id,
                "concept_label": labels.get(edge.concept_id) if edge.concept_id else None,
                "category": category,
                "relation": edge.relation.value,
                "predicate": edge.predicate,
                "confidence": edge.confidence,
                "note": edge.note,
                "is_unresolved": edge.is_unresolved,
                "queued": queued,
                "attention": _attention(edge, queued=queued, resolved=resolved, parked=parked),
                "review": review,
                "confirmation": confirmation,
                "components": components,
            }
        )

    remaining = queued_total - queued_resolved
    # The mapping is confirmed once every queued edge resolves and nothing sits flagged or deferred: a
    # parked edge is an open concern, so it blocks the whole-mapping confirmation until resolved.
    confirmed = remaining == 0 and parked_total == 0

    # Concept cards for the right column, in first-appearance category order.
    concepts_out: list[dict] = []
    category_order: list[str] = []
    for concept in mapping.compile_concepts():
        if concept.category not in category_order:
            category_order.append(concept.category)
        concepts_out.append(
            {
                "concept_id": concept.concept_id,
                "label": labels.get(concept.concept_id, concept.concept_id),
                "category": concept.category,
                "assertion_type": concept.assertion_type.value,
            }
        )
    # Only the categories an edge actually terminates in form the legend.
    edge_categories = {e["category"] for e in edges_out if e["category"]}
    legend = [cat for cat in category_order if cat in edge_categories]

    field_set = None
    if confirmed:
        field_set = [
            _field_set_row(
                edge,
                confirmations.get(_mapping_id(edge)),
                source_labels,
                licence_by_field,
                _enrich_components(components_by_mid.get(_mapping_id(edge), []), compiled, labels),
            )
            for edge in crosswalk_obj.edges
        ]

    ledger = _build_ledger(confirmations, labels, source_labels)

    return {
        "source_std": crosswalk_obj.source_std,
        "source": {
            "std": crosswalk_obj.source_std,
            "title": source_title,
            "version": crosswalk_obj.source_std.split("/")[-1],
        },
        "threshold": crosswalk.REVIEW_CONFIDENCE_THRESHOLD,
        "categories": legend,
        "concepts": concepts_out,
        "source_fields": source_fields_out,
        "edges": edges_out,
        "queue": {
            "total": queued_total,
            "resolved": queued_resolved,
            "remaining": remaining,
            "parked": parked_total,
        },
        "confirmed": confirmed,
        "field_set": field_set,
        "ledger": ledger,
    }


def _attention(edge: CrosswalkEdge, *, queued: bool, resolved: bool, parked: bool) -> str:
    """Whether an edge needs the reviewer's judgment, directed by field type and state rather than a
    confidence number (#103; EndoExtract keeps fine-grained confidence off the screen).

    A resolved edge reads `confident`. A parked one, an unresolved one, a queued low-confidence one, and
    a derivation all read `needs_judgment`: a derivation is an interpretive call the reviewer should
    sanity-check even at high confidence, and a parked edge carries an unresolved concern.
    """
    if resolved:
        return "confident"
    if parked or queued or edge.is_unresolved or edge.relation is Relation.DERIVED_FROM:
        return "needs_judgment"
    return "confident"


def _build_ledger(
    confirmations: dict[str, dict],
    labels: dict[str, str],
    source_labels: dict[str, str],
) -> list[dict]:
    """The decision ledger (#103): one inspectable, amendable record per decided edge, newest first.

    Each record names the question the decision answered, the options weighed, the chosen or edited
    answer, who confirmed it and in what role, when, and where it is filed, so a second reviewer
    reconstructs how the edge reached its state.
    """
    records = []
    for conf in confirmations.values():
        mid = conf["mapping_id"]
        options = json.loads(conf["options_json"]) if conf.get("options_json") else []
        records.append(
            {
                "mapping_id": mid,
                "source_field": conf["source_field"],
                "source_label": source_labels.get(conf["source_field"], conf["source_field"]),
                "disposition": conf["disposition"],
                "decided_via": conf["decided_via"],
                "question": conf["question"],
                "options": options,
                "concept_id": conf["concept_id"],
                "concept_label": labels.get(conf["concept_id"]) if conf["concept_id"] else None,
                "relation": conf["relation"],
                "predicate": conf["predicate"],
                "note": conf["note"],
                "confirmed_by": conf["confirmed_by"],
                "role": conf["role"],
                "confirmed_at": conf["confirmed_at"],
                # Where the answer is filed: the edge on the crosswalk, and the mapping_confirmations
                # table that carries the provenance (#93's provisional home).
                "filed_to": {"edge": mid, "table": "mapping_confirmations"},
            }
        )
    records.sort(key=lambda r: r["confirmed_at"], reverse=True)
    return records


def confirm_edge(
    conn: Connection,
    *,
    mapping_id: str,
    concept_id: Optional[str],
    relation: str,
    confirmer: str,
    role: str,
    note: Optional[str] = None,
    licence_required: bool = False,
    disposition: str = "confirmed",
    decided_via: Optional[str] = None,
    question: Optional[str] = None,
    options: Optional[list] = None,
) -> None:
    """Record a reviewer's decision on one crosswalk edge (#95, #103): the disposition, the chosen
    concept and relation, the confirmer, their role, the licence-required flag, and the time, plus the
    ledger context (how the answer was reached, the question it answered, the options weighed).

    Any edge is decidable, not only the queued ones: a reviewer re-targets a high-confidence edge the
    same way. A `confirmed` disposition enforces the honesty rule the draft does: a resolved relation
    names a canonical concept, an `unresolved` resolution names none. A `flagged` or `deferred`
    disposition parks the edge: it names no concept, carries a reason, and does not resolve the edge.

    Raises `MappingReviewError` on an unknown edge, an unknown relation, disposition, or decided-via, a
    concept outside the canonical list, a violated honesty rule, or a flag or deferral with no reason.
    """
    if not confirmer.strip():
        raise MappingReviewError("a confirmation must name a confirmer")
    if not role.strip():
        raise MappingReviewError("a confirmation must name the confirmer's role")
    if disposition not in DISPOSITIONS:
        raise MappingReviewError(f"disposition {disposition!r} is not one of {sorted(DISPOSITIONS)}")
    if decided_via is not None and decided_via not in DECIDED_VIA:
        raise MappingReviewError(f"decided_via {decided_via!r} is not one of {sorted(DECIDED_VIA)}")

    by_mid = {_mapping_id(edge): edge for edge in crosswalk.reference_crosswalk().edges}
    edge = by_mid.get(mapping_id)
    if edge is None:
        raise MappingReviewError(f"no crosswalk edge with mapping_id {mapping_id!r}")

    if disposition == "confirmed":
        try:
            rel = Relation(relation)
        except ValueError as exc:
            raise MappingReviewError(
                f"relation {relation!r} is not one of {[r.value for r in Relation]}"
            ) from exc
        resolved = rel is not Relation.UNRESOLVED
        has_concept = bool(concept_id and concept_id.strip())
        if resolved and not has_concept:
            raise MappingReviewError(
                f"relation {rel.value!r} names no concept_id; a resolution with no concept is "
                "'unresolved'"
            )
        if not resolved and has_concept:
            raise MappingReviewError("an 'unresolved' resolution names no concept_id")
        if has_concept and concept_id not in crosswalk.canonical_concept_ids():
            raise MappingReviewError(f"{concept_id!r} is not a canonical concept")
        stored_concept = concept_id if has_concept else None
    else:
        # A flag or a deferral parks the edge. It resolves to no concept and reads as unresolved, but
        # its disposition marks it a live concern rather than a confirmed out-of-scope decision. A
        # reason is required: silence is never the response to a parked edge.
        if not (note and note.strip()):
            raise MappingReviewError(f"a {disposition} decision must carry a reason")
        rel = Relation.UNRESOLVED
        stored_concept = None

    storage.record_mapping_confirmation(
        conn,
        mapping_id=mapping_id,
        source_std=edge.source_std,
        source_field=edge.source_field,
        concept_id=stored_concept,
        relation=rel.value,
        predicate=crosswalk.RELATION_TO_PREDICATE[rel],
        confirmed_by=confirmer,
        role=role,
        note=note,
        licence_required=licence_required,
        confirmed_at=datetime.now(timezone.utc).isoformat(),
        disposition=disposition,
        decided_via=decided_via,
        question=question,
        options_json=json.dumps(options) if options else None,
    )
    # A plain single-edge decision carries no split, so clear any components a prior split left on this
    # field: re-targeting away from a split must not orphan its component rows.
    storage.record_mapping_components(
        conn, mapping_id=mapping_id, source_std=edge.source_std, components=[]
    )


def confirm_split(
    conn: Connection,
    *,
    mapping_id: str,
    headline_concept_id: Optional[str],
    headline_relation: str,
    components: list[dict],
    confirmer: str,
    role: str,
    note: Optional[str] = None,
    licence_required: bool = False,
    decided_via: Optional[str] = None,
    question: Optional[str] = None,
    options: Optional[list] = None,
) -> None:
    """Record a reviewer's decision to split one source field across several canonical concepts (#103
    follow-on). The field keeps a headline decision (an `unresolved` lane, or a derived summary concept
    such as the ProMisE class) recorded in mapping_confirmations, and fans out to the component concepts
    it also resolves to (the POLE, MMR, and p53 primitives), each recorded in
    mapping_confirmation_components.

    A split is always a `confirmed` resolution. The headline obeys the same honesty rule confirm_edge
    enforces. Each component names a canonical concept and a resolving relation (never `unresolved`),
    the components are distinct, and none repeats the headline concept.

    Raises `MappingReviewError` on an unknown edge, an empty component set, a component outside the
    canonical list, a component with an `unresolved` or unknown relation, a duplicate component, a
    component that repeats the headline, or a violated headline honesty rule.
    """
    if not confirmer.strip():
        raise MappingReviewError("a confirmation must name a confirmer")
    if not role.strip():
        raise MappingReviewError("a confirmation must name the confirmer's role")

    by_mid = {_mapping_id(edge): edge for edge in crosswalk.reference_crosswalk().edges}
    edge = by_mid.get(mapping_id)
    if edge is None:
        raise MappingReviewError(f"no crosswalk edge with mapping_id {mapping_id!r}")

    # The headline: a resolved summary concept, or the unresolved lane. Same honesty rule as confirm_edge.
    try:
        head_rel = Relation(headline_relation)
    except ValueError as exc:
        raise MappingReviewError(
            f"relation {headline_relation!r} is not one of {[r.value for r in Relation]}"
        ) from exc
    head_resolved = head_rel is not Relation.UNRESOLVED
    head_has_concept = bool(headline_concept_id and headline_concept_id.strip())
    if head_resolved and not head_has_concept:
        raise MappingReviewError(
            f"relation {head_rel.value!r} names no concept_id; a resolution with no concept is "
            "'unresolved'"
        )
    if not head_resolved and head_has_concept:
        raise MappingReviewError("an 'unresolved' headline names no concept_id")
    if head_has_concept and headline_concept_id not in crosswalk.canonical_concept_ids():
        raise MappingReviewError(f"{headline_concept_id!r} is not a canonical concept")
    headline_concept = headline_concept_id if head_has_concept else None

    if not components:
        raise MappingReviewError("a split names at least one component concept")

    canonical = crosswalk.canonical_concept_ids()
    seen: set[str] = set()
    resolved_components: list[dict] = []
    for component in components:
        cid = component.get("concept_id")
        if not cid or cid not in canonical:
            raise MappingReviewError(f"component {cid!r} is not a canonical concept")
        if cid == headline_concept:
            raise MappingReviewError(
                f"component {cid!r} repeats the headline concept; a component is a distinct primitive"
            )
        if cid in seen:
            raise MappingReviewError(f"component {cid!r} is named twice")
        seen.add(cid)
        try:
            crel = Relation(component.get("relation"))
        except ValueError as exc:
            raise MappingReviewError(
                f"component relation {component.get('relation')!r} is not one of "
                f"{[r.value for r in Relation]}"
            ) from exc
        if crel is Relation.UNRESOLVED:
            raise MappingReviewError(
                f"component {cid!r} names an 'unresolved' relation; a component resolves to its concept"
            )
        resolved_components.append(
            {
                "concept_id": cid,
                "relation": crel.value,
                "predicate": crosswalk.RELATION_TO_PREDICATE[crel],
                "note": component.get("note"),
            }
        )

    now = datetime.now(timezone.utc).isoformat()
    storage.record_mapping_confirmation(
        conn,
        mapping_id=mapping_id,
        source_std=edge.source_std,
        source_field=edge.source_field,
        concept_id=headline_concept,
        relation=head_rel.value,
        predicate=crosswalk.RELATION_TO_PREDICATE[head_rel],
        confirmed_by=confirmer,
        role=role,
        note=note,
        licence_required=licence_required,
        confirmed_at=now,
        disposition="confirmed",
        decided_via=decided_via,
        question=question,
        options_json=json.dumps(options) if options else None,
    )
    storage.record_mapping_components(
        conn,
        mapping_id=mapping_id,
        source_std=edge.source_std,
        components=resolved_components,
    )


# --- The edge chat co-pilot (#103) ----------------------------------------------------------------
#
# Clicking an edge opens a conversation scoped to it: the reviewer writes a concern, and the engine
# replies concisely and, when the concern points at a better resolution, carries one concrete remap the
# reviewer applies in a click (the frontend files the proposed concept and relation through confirm_edge
# with decided_via='chat'). The engine call goes through a provider-agnostic seam, so the fast test suite
# runs against an injected fake with no API key. The chat model is named explicitly here rather than
# inherited from another pass.

MAPPING_CHAT_MODEL = "claude-sonnet-5"


@runtime_checkable
class MappingChatResponder(Protocol):
    """Replies to a reviewer's message about one crosswalk edge, concisely, and proposes a remap when
    one is warranted. The seam a local, on-premise model or a test fake plugs into: it never surfaces
    its provider to the call site."""

    def explain(
        self, *, edge_context: dict, question: str, history: list[dict]
    ) -> dict: ...


EDGE_CHAT_TOOL = {
    "name": "reply_to_reviewer",
    "description": (
        "Reply to the reviewer's message about one crosswalk edge from a source field to a canonical "
        "concept. Answer concisely and professionally. When the message points at a better resolution, "
        "propose one concrete remap the reviewer can apply in a click; otherwise propose nothing and say "
        "in a sentence why the drafted edge stands. A remap can re-target the edge to a single concept, "
        "or split the field across several component concepts (with a derived summary as the headline). "
        "Propose only concepts from the given canonical list."
    ),
    "input_schema": {
        "type": "object",
        "properties": {
            "reply": {
                "type": "string",
                "description": "The reply to the reviewer, at most three short sentences. State the "
                "assessment, and when you propose a remap, name what changes and the one reason for it. "
                "No preamble, and do not restate the reviewer's message.",
            },
            "proposal": {
                "anyOf": [
                    {
                        "type": "object",
                        "properties": {
                            "summary": {
                                "type": "string",
                                "description": "One short line naming the remap to apply, so the reviewer "
                                "reads it on the apply button.",
                            },
                            "concept_id": {
                                "anyOf": [{"type": "string"}, {"type": "null"}],
                                "description": "The headline concept. For a plain re-target, the concept "
                                "to bind to, or null for an unresolved resolution. For a split, the "
                                "derived summary concept the components roll up to (or null to leave the "
                                "field's own lane unresolved and hold only the components).",
                            },
                            "relation": {
                                "type": "string",
                                "enum": [r.value for r in Relation],
                                "description": "The headline relation. Use 'unresolved' with a null "
                                "concept_id to route the field's own lane aside; use 'derived_from' when "
                                "the headline concept is computed from the components.",
                            },
                            "components": {
                                "type": "array",
                                "description": "The component concepts the field also resolves to, for a "
                                "split. Empty for a plain re-target. Each is a distinct canonical concept "
                                "the field carries in its own right, such as the POLE, MMR, and p53 "
                                "primitives behind a ProMisE class.",
                                "items": {
                                    "type": "object",
                                    "properties": {
                                        "concept_id": {
                                            "type": "string",
                                            "description": "A canonical concept the field resolves to.",
                                        },
                                        "relation": {
                                            "type": "string",
                                            "enum": [
                                                r.value for r in Relation if r is not Relation.UNRESOLVED
                                            ],
                                            "description": "How the field carries this component, never "
                                            "'unresolved'.",
                                        },
                                    },
                                    "required": ["concept_id", "relation"],
                                },
                            },
                        },
                        "required": ["summary", "concept_id", "relation", "components"],
                    },
                    {"type": "null"},
                ],
                "description": "One concrete remap the reviewer can apply, or null to leave the edge as "
                "drafted.",
            },
        },
        "required": ["reply", "proposal"],
    },
}

_CHAT_INSTRUCTIONS = """\
You are the crosswalk engine's co-pilot. A reviewer is confirming a mapping from one registry's source \
field to a canonical concept, and writes to you about one edge. Reply to their message: whether the \
drafted edge is right, and if not, what it should be.

Write plainly and professionally. Keep the reply to at most three short sentences. Do not restate the \
reviewer's message, and do not open with a preamble.

Ground your reply in the edge context and the canonical concept list you are given. Do not invent a \
concept: every proposed concept is one from the list by its concept_id. A remap either re-targets the \
field to one concept, or splits it across several component concepts when a single edge would collapse \
distinct results the reviewer needs kept apart. For a split, set components to the primitives the field \
carries, and set the headline concept and relation to the derived summary they roll up to (or a null \
headline concept with an 'unresolved' relation to hold only the components). When the reviewer's message \
points at a better resolution, set proposal to that one remap; otherwise set proposal to null and say in \
one sentence why the drafted edge stands. Call reply_to_reviewer with your answer.\
"""

# The primitives a ProMisE molecular class rolls up from (docs/taxonomy.md#1.7). The offline stub
# proposes this split for the molecular_classification field so the demo runs with no API key; the hosted
# model reaches the same shape from the concept list and the reviewer's message.
_PROMISE_COMPONENTS = ("pole_exonuclease_mutation", "mmr_ihc", "p53_ihc")


class AnthropicMappingChatResponder:
    """The hosted chat co-pilot: it reads the edge context and the concept list and answers the
    reviewer's question through a forced tool call. Names no provider at the call site."""

    def __init__(
        self,
        client: Optional[anthropic.Anthropic] = None,
        model: str = MAPPING_CHAT_MODEL,
        max_tokens: int = 1024,
    ) -> None:
        self._client = client
        self._model = model
        self._max_tokens = max_tokens

    def explain(self, *, edge_context: dict, question: str, history: list[dict]) -> dict:
        client = self._client or get_client()
        system = [
            {"type": "text", "text": _CHAT_INSTRUCTIONS},
            {
                "type": "text",
                "text": "CANONICAL CONCEPT LIST:\n\n"
                + json.dumps(edge_context["concepts"], indent=2),
                # The concept list is identical across every edge, so cache it once.
                "cache_control": {"type": "ephemeral"},
            },
        ]
        messages: list[dict] = []
        for turn in history:
            role = "assistant" if turn.get("role") == "assistant" else "user"
            messages.append({"role": role, "content": turn.get("text", "")})
        messages.append(
            {
                "role": "user",
                "content": (
                    "EDGE UNDER REVIEW:\n\n"
                    + json.dumps(edge_context["edge"], indent=2)
                    + f"\n\nREVIEWER QUESTION:\n{question}\n\n"
                    "Call explain_mapping_edge now."
                ),
            }
        )
        response = client.messages.create(
            model=self._model,
            max_tokens=self._max_tokens,
            system=system,
            tools=[EDGE_CHAT_TOOL],
            tool_choice={"type": "tool", "name": EDGE_CHAT_TOOL["name"]},
            messages=messages,
        )
        tool_call = next((b for b in response.content if b.type == "tool_use"), None)
        if tool_call is None:
            raise MappingReviewError("edge chat returned no tool call")
        return tool_call.input


class StubMappingChatResponder:
    """A deterministic, offline chat responder for local demos and screenshot verification: it answers
    from the edge context alone, with no model call and no API key. Enabled by the
    ENDOPATH_MAPPING_CHAT_STUB environment variable, never the default."""

    def explain(self, *, edge_context: dict, question: str, history: list[dict]) -> dict:
        edge = edge_context["edge"]
        relation = edge["relation"]
        canonical = {c["concept_id"] for c in edge_context["concepts"]}
        components_available = all(cid in canonical for cid in _PROMISE_COMPONENTS)
        if edge["source_field"] == "molecular_classification" and components_available:
            reply = (
                "Mapping this one field only to the ProMisE class collapses the POLE, MMR, and p53 "
                "results, which can be discordant. Split it into those three components and keep the "
                "ProMisE class as the derived headline."
            )
            proposal = {
                "summary": "Split into POLE, MMR, and p53; derive the ProMisE class from them.",
                "concept_id": edge["concept_id"],
                "relation": "derived_from",
                "components": [
                    {"concept_id": cid, "relation": "exact"} for cid in _PROMISE_COMPONENTS
                ],
            }
        elif edge["is_unresolved"]:
            reply = (
                f"No canonical concept answers '{edge['source_label']}', so the edge stays unresolved "
                "and routes to its own lane."
            )
            proposal = {
                "summary": "Confirm unresolved.",
                "concept_id": None,
                "relation": "unresolved",
                "components": [],
            }
        else:
            reply = (
                f"'{edge['source_label']}' is drafted a {relation.replace('_', ' ')} link to "
                f"{edge['concept_label']}: {edge['note']}"
            )
            proposal = {
                "summary": f"Keep the {relation.replace('_', ' ')} link to {edge['concept_label']}.",
                "concept_id": edge["concept_id"],
                "relation": relation,
                "components": [],
            }
        return {"reply": reply, "proposal": proposal}


_CHAT_RESPONDER: Optional[MappingChatResponder] = None


def configure_chat_responder(responder: Optional[MappingChatResponder]) -> None:
    """Test/CLI hook to inject a chat responder (a fake in the fast suite), or reset to the default
    hosted one with None."""
    global _CHAT_RESPONDER
    _CHAT_RESPONDER = responder


def _chat_responder() -> MappingChatResponder:
    if _CHAT_RESPONDER is not None:
        return _CHAT_RESPONDER
    import os

    if os.environ.get("ENDOPATH_MAPPING_CHAT_STUB"):
        return StubMappingChatResponder()
    return AnthropicMappingChatResponder()


def _edge_context(edge: CrosswalkEdge, compiled: dict, labels: dict, source_labels: dict) -> dict:
    """The context the chat co-pilot reads: the edge as drafted, and the canonical concept list it may
    re-target to."""
    concepts = [
        {
            "concept_id": c.concept_id,
            "label": labels.get(c.concept_id, c.concept_id),
            "category": c.category,
            "assertion_type": c.assertion_type.value,
        }
        for c in compiled.values()
    ]
    category = compiled[edge.concept_id].category if edge.concept_id in compiled else None
    return {
        "edge": {
            "source_field": edge.source_field,
            "source_label": source_labels.get(edge.source_field, edge.source_field),
            "concept_id": edge.concept_id,
            "concept_label": labels.get(edge.concept_id) if edge.concept_id else None,
            "category": category,
            "relation": edge.relation.value,
            "predicate": edge.predicate,
            "confidence": edge.confidence,
            "note": edge.note,
            "is_unresolved": edge.is_unresolved,
        },
        "concepts": concepts,
    }


def explain_edge(
    conn: Connection,
    *,
    mapping_id: str,
    question: str,
    history: Optional[list[dict]] = None,
    responder: Optional[MappingChatResponder] = None,
) -> dict:
    """Reply to a reviewer's message about one crosswalk edge (#103), concisely, and carry a concrete
    remap the reviewer can apply in a click when one is warranted. `conn` is unused today (the crosswalk
    is committed, not per-connection) and kept for the endpoint's symmetry with the other mapping calls
    and for a future per-corpus crosswalk.

    Raises `MappingReviewError` on a blank message or an edge the crosswalk does not carry. A proposed
    concept outside the canonical list, or a relation outside the vocabulary, drops the proposal to null
    rather than raising: a chat proposal is advice a reviewer applies deliberately, not a write.
    """
    if not question.strip():
        raise MappingReviewError("an edge chat message must carry a question")

    by_mid = {_mapping_id(e): e for e in crosswalk.reference_crosswalk().edges}
    edge = by_mid.get(mapping_id)
    if edge is None:
        raise MappingReviewError(f"no crosswalk edge with mapping_id {mapping_id!r}")

    compiled = {c.concept_id: c for c in mapping.compile_concepts()}
    labels = crosswalk.concept_labels()
    source_labels = _source_labels()
    context = _edge_context(edge, compiled, labels, source_labels)

    responder = responder or _chat_responder()
    raw = responder.explain(edge_context=context, question=question, history=history or [])

    reply = str(raw.get("reply") or "").strip()
    proposal = _clean_proposal(raw.get("proposal"), labels)

    return {"mapping_id": mapping_id, "reply": reply, "proposal": proposal}


def _clean_proposal(raw: object, labels: dict[str, str]) -> Optional[dict]:
    """Validate a raw remap proposal into a fileable one, or None. The headline is held to the same
    honesty rule the draft is: an 'unresolved' relation names no concept, and a resolving relation with
    no concept and no components is incoherent and drops to None. Components are validated one by one and
    the invalid ones dropped: each must name a canonical concept and a resolving relation, be distinct,
    and not repeat the headline. A chat proposal is advice a reviewer applies deliberately, not a write,
    so a malformed part drops rather than raising."""
    if not isinstance(raw, dict):
        return None

    canonical = crosswalk.canonical_concept_ids()
    concept = raw.get("concept_id")
    if concept is not None and concept not in canonical:
        concept = None
    relation = raw.get("relation")
    if relation not in {r.value for r in Relation}:
        return None
    if relation == Relation.UNRESOLVED.value:
        concept = None

    components: list[dict] = []
    seen: set[str] = set()
    for item in raw.get("components") or []:
        if not isinstance(item, dict):
            continue
        cid = item.get("concept_id")
        crel = item.get("relation")
        if cid not in canonical or cid == concept or cid in seen:
            continue
        if crel not in {r.value for r in Relation} or crel == Relation.UNRESOLVED.value:
            continue
        seen.add(cid)
        components.append(
            {"concept_id": cid, "relation": crel, "concept_label": labels.get(cid, cid)}
        )

    # A resolving headline relation with no concept is only coherent when components carry the split.
    if relation != Relation.UNRESOLVED.value and concept is None and not components:
        return None

    if components:
        default_summary = "Split across " + ", ".join(c["concept_label"] for c in components)
        if concept is not None:
            default_summary += f", derived into {labels.get(concept, concept)}"
        default_summary += "."
    elif concept is not None:
        default_summary = f"Re-target to {labels.get(concept, concept)} as {relation.replace('_', ' ')}."
    else:
        default_summary = "Confirm unresolved."

    return {
        "summary": str(raw.get("summary") or "").strip() or default_summary,
        "concept_id": concept,
        "relation": relation,
        "concept_label": labels.get(concept) if concept else None,
        "components": components,
    }