src/endopath/crosswalk.py

"""Draft crosswalk edges from each source field to a canonical concept (#93, the A3 pass).

The A3 pass reads the A1 induced field list (`src/endopath/induction.py`) together with the A2
canonical concept list (`src/endopath/mapping.py`) and drafts, for each source field the report or
dictionary states, one edge to the concept that answers it:

```json
{
  "source_std": "cap_uterus/5.1.0.0",
  "source_field": "Myometrial Invasion",
  "concept_id": "myometrial_invasion_percent",
  "relation": "exact | narrower | broader | derived_from | unresolved",
  "confidence": 0.88,
  "note": "CAP reports depth and thickness; the concept derives the percent"
}
```

The honesty rule is the point. When no clean concept answers a source field, the edge is `unresolved`
with a one-sentence reason, and it names no concept. An edge is never coerced to `exact` to clear the
queue. The `unresolved` and low-confidence edges are the reviewer work queue that the crosswalk surface
(#95) renders as the dashed-red connector and the maieutic loop resolves.

The pass is LLM-native: a language model reads the two lists and drafts the edges. The model call goes
through a `CrosswalkDrafter`, a provider-agnostic seam, so the fast test suite runs against an injected
fake with no API key. The mapping model is named explicitly here rather than inherited from another pass.

Two rules the contract enforces structurally, not by prompt alone:

1. Unresolved is not forced. A resolved edge (`exact`, `narrower`, `broader`, `derived_from`) names a
   real canonical concept. An edge that names no concept is `unresolved` and carries its reason, never a
   fabricated `exact`.
2. Every edge carries its rationale. `note` is never blank, and `confidence` stays in [0, 1], so an edge
   is never a bare link with no reason and no strength.

`relation` projects onto #35's `crosswalk.csv` `predicate` column, the SKOS controlled vocabulary
(dec_015). `RELATION_TO_PREDICATE` binds the five A3 relations to five of that column's six values;
`conflict`, the sixth, marks two standards' rules disagreeing and is a concept-to-concept judgement, not
a source-field-to-concept edge, so this pass does not produce it.

Where the confirmation provenance of a drafted edge lives, once a reviewer (#95) confirms it, is an open
decision recorded on #93: it grows #35's `crosswalk.csv` schema or gives the mapping stage its own
table. This pass drafts the candidate edges and does not decide that, so it writes no confirmation
provenance and adds no confirmed/confirmer/timestamp field.
"""

from __future__ import annotations

import json
from enum import Enum
from typing import Optional, Protocol, runtime_checkable

import anthropic
from pydantic import BaseModel, ConfigDict, ValidationError, model_validator

from endopath import fields, mapping
from endopath.dictionary import BUILTIN_DICTIONARY_ID
from endopath.induction import InducedReport
from endopath.llm_extraction import get_client

# The mapping model is named explicitly here rather than inherited from the extraction or induction
# pass. Drafting the crosswalk reads the whole concept list and reasons over an unfamiliar report's
# field set, so it runs on the most capable model by default; a caller overrides it per call.
CROSSWALK_MODEL = "claude-sonnet-5"


class CrosswalkError(ValueError):
    """The crosswalk draft did not yield a valid edge set: a resolved edge naming no concept, an
    unresolved edge that forces a link, an edge for a field the report did not state, a source field
    with no edge or more than one, a confidence outside [0, 1], a blank note, or a concept_id outside
    the canonical concept list."""


class Relation(str, Enum):
    """The closed vocabulary an edge's relation takes. Describes how the source field relates to the
    canonical concept it maps to.

    `exact`, `narrower`, and `broader` are hierarchical matches. `derived_from` marks a concept the
    report's primitives feed rather than state directly (CAP reports depth and thickness; the concept
    derives the percent). `unresolved` marks a source field no clean concept answers: it names no
    concept and routes to the reviewer queue."""

    EXACT = "exact"
    NARROWER = "narrower"
    BROADER = "broader"
    DERIVED_FROM = "derived_from"
    UNRESOLVED = "unresolved"


# The A3 relation projects onto #35's crosswalk.csv `predicate` column, the SKOS controlled vocabulary
# plus `conflict` (dec_015, a six-value column). Five relations bind to five predicates; `conflict` is
# not producible here, because it marks two standards' rules disagreeing rather than a source-field-to-
# concept edge. `derived_from` maps to `relatedMatch`: a derivation is a related, non-hierarchical link,
# and #35 carries the derivation itself in derivations.csv.
RELATION_TO_PREDICATE: dict[Relation, str] = {
    Relation.EXACT: "exactMatch",
    Relation.NARROWER: "narrowMatch",
    Relation.BROADER: "broadMatch",
    Relation.DERIVED_FROM: "relatedMatch",
    Relation.UNRESOLVED: "noMatch",
}


class CrosswalkEdge(BaseModel):
    """One drafted edge: the source standard and field, the canonical concept it maps to (or none when
    unresolved), the relation, a confidence, and a one-sentence rationale. The #93 contract row."""

    model_config = ConfigDict(frozen=True)

    source_std: str
    source_field: str
    concept_id: Optional[str] = None
    relation: Relation
    confidence: float
    note: str

    @model_validator(mode="after")
    def _edge_is_honest_and_carries_its_reason(self) -> "CrosswalkEdge":
        if not self.source_std.strip():
            raise ValueError("an edge must name a source_std")
        if not self.source_field.strip():
            raise ValueError("an edge must name a source_field")
        if not 0.0 <= self.confidence <= 1.0:
            raise ValueError(
                f"{self.source_field}: confidence must be in [0, 1], got {self.confidence}"
            )
        # Every edge carries its rationale, resolved or not: silence is never the response to a miss.
        if not self.note.strip():
            raise ValueError(f"{self.source_field}: an edge must carry a one-sentence note, not a blank")
        resolved = self.relation is not Relation.UNRESOLVED
        has_concept = self.concept_id is not None and self.concept_id.strip() != ""
        if resolved and not has_concept:
            # The honesty rule: a resolved relation names a real concept. An edge with no clean concept
            # is `unresolved` with a reason, never a fabricated `exact`.
            raise ValueError(
                f"{self.source_field}: relation {self.relation.value!r} names no concept_id; a field "
                "with no clean concept is 'unresolved' with a note, never a fabricated link"
            )
        if not resolved and has_concept:
            # The other half of the honesty rule: an unresolved edge does not force a link.
            raise ValueError(
                f"{self.source_field}: an 'unresolved' edge names no concept_id (it routes to the "
                f"reviewer queue), got {self.concept_id!r}"
            )
        return self

    @property
    def predicate(self) -> str:
        """This edge's relation as #35's SKOS `predicate` value (dec_015)."""
        return RELATION_TO_PREDICATE[self.relation]

    @property
    def is_unresolved(self) -> bool:
        """Whether this edge routes to the reviewer queue: an unresolved relation, which by the honesty
        rule names no concept. The dashed-red connector #95 renders."""
        return self.relation is Relation.UNRESOLVED


class Crosswalk(BaseModel):
    """The A3 draft for one source standard: the standard and the drafted edges, one per source field."""

    model_config = ConfigDict(frozen=True)

    source_std: str
    edges: tuple[CrosswalkEdge, ...]

    @property
    def unresolved(self) -> tuple[CrosswalkEdge, ...]:
        """The reviewer queue: the edges no clean concept answered (#95)."""
        return tuple(edge for edge in self.edges if edge.is_unresolved)


@runtime_checkable
class CrosswalkDrafter(Protocol):
    """Turns the source field list plus the canonical concept list into the raw edge set. The seam a
    local, on-premise model or a test fake plugs into: it never surfaces its provider to the call site."""

    def draft(self, *, source_std: str, source_fields: list[dict], concepts: list[dict]) -> dict: ...


# --- The schema-constrained tool the hosted model is forced to call --------------------------------

CROSSWALK_TOOL = {
    "name": "draft_crosswalk_edges",
    "description": (
        "Record the crosswalk draft: for each source field the report or dictionary states, one edge to "
        "the canonical concept that answers it, with a relation and a confidence, or an 'unresolved' edge "
        "with a reason when no clean concept answers it."
    ),
    "input_schema": {
        "type": "object",
        "properties": {
            "edges": {
                "type": "array",
                "items": {
                    "type": "object",
                    "properties": {
                        "source_std": {
                            "type": "string",
                            "description": "The standard the source field belongs to (the A1 detected_standard).",
                        },
                        "source_field": {
                            "type": "string",
                            "description": "The source field's own name, exactly as A1 induced it.",
                        },
                        "concept_id": {
                            "anyOf": [{"type": "string"}, {"type": "null"}],
                            "description": (
                                "The canonical concept this field maps to, taken from the concept list. "
                                "Null when relation is 'unresolved': an unresolved edge names no concept."
                            ),
                        },
                        "relation": {
                            "type": "string",
                            "enum": [r.value for r in Relation],
                            "description": (
                                "exact (the field is the concept), narrower or broader (a hierarchical "
                                "match), derived_from (the concept is computed from this field and others, "
                                "for example a percent from a depth and a thickness), or unresolved (no "
                                "clean concept answers it). Never coerce to exact to clear the queue."
                            ),
                        },
                        "confidence": {"type": "number", "minimum": 0, "maximum": 1},
                        "note": {
                            "type": "string",
                            "description": (
                                "A one-sentence rationale for the edge. For an unresolved edge, the reason "
                                "no concept answers the field. Never blank."
                            ),
                        },
                    },
                    "required": ["source_std", "source_field", "concept_id", "relation", "confidence", "note"],
                },
            }
        },
        "required": ["edges"],
    },
}

_INSTRUCTIONS = """\
You draft a crosswalk from one report's induced field set to the canonical concept list, and call \
draft_crosswalk_edges with the result. For each source field, emit exactly one edge to the concept that \
answers it, with a relation and a confidence.

The honesty rule is absolute:
- When a clean concept answers the field, name it and give the relation: exact, narrower, broader, or \
derived_from. Use derived_from when the concept is computed from this field together with others (a \
percent derived from a depth and a thickness, a stage derived from several primitives).
- When no concept in the list cleanly answers the field, emit relation 'unresolved' with concept_id null \
and a one-sentence reason. Do not force a link. Never coerce an edge to 'exact' to clear the queue.
- A pre-screening immunostain field whose value arrives from another source rather than the report \
(mismatch-repair or p53 immunohistochemistry read from a molecular join) is 'derived_from' or \
'unresolved', never 'exact'.

Every edge carries a one-sentence note, resolved or not. Confidence is your calibrated strength in [0, 1]; \
a weak or uncertain edge takes a low confidence and joins the reviewer queue alongside the unresolved edges.
"""


class AnthropicCrosswalkDrafter:
    """The hosted mapping-model drafter: it reads the source field list and the canonical concept list
    and returns the raw edge set through a forced tool call. Names no provider at the call site."""

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

    def draft(self, *, source_std: str, source_fields: list[dict], concepts: list[dict]) -> dict:
        client = self._client or get_client()
        system = [
            {"type": "text", "text": _INSTRUCTIONS},
            {
                "type": "text",
                "text": "CANONICAL CONCEPT LIST (the A2 targets):\n\n" + json.dumps(concepts, indent=2),
                # The concept list is identical across every report for one dictionary, so cache it once.
                "cache_control": {"type": "ephemeral"},
            },
        ]
        content = [
            {
                "type": "text",
                "text": (
                    f"SOURCE STANDARD source_std={source_std}\n\n"
                    "SOURCE FIELDS (the A1 induced fields to map):\n\n"
                    + json.dumps(source_fields, indent=2)
                    + "\n\nCall draft_crosswalk_edges now, one edge per source field."
                ),
            }
        ]
        response = client.messages.create(
            model=self._model,
            max_tokens=self._max_tokens,
            system=system,
            tools=[CROSSWALK_TOOL],
            tool_choice={"type": "tool", "name": CROSSWALK_TOOL["name"]},
            messages=[{"role": "user", "content": content}],
        )
        # Fail loudly on truncation rather than returning a partial tool call that would be persisted as
        # a truncated crosswalk with source fields silently dropped.
        if response.stop_reason == "max_tokens":
            raise CrosswalkError(
                "crosswalk draft hit max_tokens before completing the tool call; raise max_tokens"
            )
        tool_call = next((b for b in response.content if b.type == "tool_use"), None)
        if tool_call is None:
            raise CrosswalkError("crosswalk draft returned no tool call")
        return tool_call.input


def canonical_concept_ids() -> frozenset[str]:
    """The canonical concept ids an edge may map to, from the A2 pass (`mapping.compile_concepts`)."""
    return frozenset(concept.concept_id for concept in mapping.compile_concepts())


def _build_edge(item: object, *, source_std: str) -> CrosswalkEdge:
    if not isinstance(item, dict):
        raise CrosswalkError(f"each edge must be a JSON object, got {type(item).__name__}")
    payload = {key: item.get(key) for key in ("concept_id", "relation", "confidence", "note")}
    # source_std and source_field identify the edge; source_std is set from the trusted argument rather
    # than trusted from the model, the way A1 sets report_id from the request.
    payload["source_std"] = source_std
    payload["source_field"] = item.get("source_field")
    try:
        return CrosswalkEdge(**payload)
    except ValidationError as error:
        raise CrosswalkError(f"edge {item.get('source_field')!r}: {error}") from error


def build_crosswalk(
    raw: object,
    *,
    source_std: str,
    source_fields: list[str],
    concept_ids: Optional[frozenset[str]] = None,
) -> Crosswalk:
    """Validate the raw edge set into a `Crosswalk`, enforcing the honesty rule and full coverage.

    Raises `CrosswalkError` when the output is not a JSON object with an `edges` list, when a resolved
    edge names no concept or an unresolved edge forces a link, when an edge names a concept outside the
    canonical list, when an edge names a source field the report did not state, or when a source field
    draws no edge or more than one. `source_std` is set from the caller rather than trusted from the
    model, and `source_fields` is the A1 field set every edge must cover exactly once.
    """
    if not isinstance(raw, dict):
        raise CrosswalkError(f"crosswalk output must be a JSON object, got {type(raw).__name__}")
    raw_edges = raw.get("edges")
    if not isinstance(raw_edges, list):
        raise CrosswalkError("crosswalk output must carry an 'edges' list")

    allowed = concept_ids if concept_ids is not None else canonical_concept_ids()
    expected = list(source_fields)
    expected_set = set(expected)

    edges = tuple(_build_edge(item, source_std=source_std) for item in raw_edges)

    for edge in edges:
        if edge.concept_id is not None and edge.concept_id not in allowed:
            raise CrosswalkError(
                f"edge {edge.source_field!r} maps to {edge.concept_id!r}, which is not a canonical "
                "concept in the A2 list"
            )
        if edge.source_field not in expected_set:
            raise CrosswalkError(
                f"edge names source_field {edge.source_field!r}, which the report did not state"
            )

    # Each source field drafts exactly one edge: coverage (no field silently dropped) and uniqueness
    # (no field drawn twice). Silence is never the response to a miss, so an unmapped field is an error,
    # not an omission the reviewer would never see.
    covered = [edge.source_field for edge in edges]
    seen: set[str] = set()
    duplicates = {name for name in covered if name in seen or seen.add(name)}
    if duplicates:
        raise CrosswalkError(f"these source fields drew more than one edge: {sorted(duplicates)}")
    missing = [name for name in expected if name not in seen]
    if missing:
        raise CrosswalkError(f"these source fields drew no edge: {missing}")

    return Crosswalk(source_std=source_std, edges=edges)


def draft_crosswalk(
    report: InducedReport,
    *,
    concepts: Optional[list[dict]] = None,
    drafter: Optional[CrosswalkDrafter] = None,
) -> Crosswalk:
    """The A3 pass for one report: draft an edge from every induced source field to a canonical concept.

    Reads the A1 `InducedReport` for the source standard and its fields, and the A2 concept list for the
    targets. `drafter` is the model seam: the default calls the hosted mapping model, a fake substitutes
    for it in a test. Raises `CrosswalkError` when the drafted edges violate the contract.
    """
    concepts = concepts if concepts is not None else mapping.concept_list()
    concept_ids = frozenset(concept["concept_id"] for concept in concepts)
    source_fields = [field.name for field in report.fields]
    field_summaries = [
        {"name": field.name, "value_or_absent": field.value_or_absent} for field in report.fields
    ]
    drafter = drafter or AnthropicCrosswalkDrafter()
    raw = drafter.draft(
        source_std=report.detected_standard, source_fields=field_summaries, concepts=concepts
    )
    return build_crosswalk(
        raw,
        source_std=report.detected_standard,
        source_fields=source_fields,
        concept_ids=concept_ids,
    )


def crosswalk_json(crosswalk: Crosswalk) -> str:
    """The crosswalk draft as strict JSON, the #93 artifact."""
    return json.dumps(crosswalk.model_dump(mode="json"), indent=2) + "\n"


# --- The committed reference instance (#59, #95) --------------------------------------------------
#
# The A3 pass above drafts a report's crosswalk with a live model. The built-in CAP endometrium
# checklist has a committed, validated mapping to the canonical concepts already: `cap_checklist_coverage`
# (`src/endopath/mapping.py`) binds each CAP field to the concept it answers, and `UNMODELED_CAP_FIELDS`
# names any CAP field the concept list does not model (empty in the MVP). `reference_crosswalk()` turns
# that committed pair into the same `CrosswalkEdge` contract the model pass emits, so the mapping-review
# surface (#95, design-guideline section 9.1) renders a real crosswalk, and the maieutic loop resolves
# it, with no model call and no hand-authored edge list.
#
# The relation follows the target concept's own shape: a CAP field feeding a derived concept (a percent
# from a depth, a stage from an algorithm) is `derived_from`; a CAP field that is the concept is `exact`.
# An unmodeled CAP field is `unresolved`: it names no concept and routes to the reviewer queue.

# A drafted edge at or below this confidence joins the reviewer queue alongside the unresolved edges
# (#95): the queue is the unresolved-or-low-confidence work the maieutic loop resolves. It is never
# coerced to a high confidence to clear the queue.
REVIEW_CONFIDENCE_THRESHOLD = 0.75

_RELATION_CONFIDENCE: dict[Relation, float] = {
    Relation.EXACT: 0.96,
    Relation.NARROWER: 0.9,
    Relation.BROADER: 0.9,
    Relation.DERIVED_FROM: 0.82,
    Relation.UNRESOLVED: 0.2,
}

# CAP fields the coverage table maps but whose fit a reviewer should adjudicate. CAP's single
# "molecular classification" field is broader than the specific ProMisE four-class concept it derives,
# so the edge is a real derivation carried at a low confidence that routes it to the queue rather than a
# fabricated high-confidence exact match.
_UNCERTAIN_SOURCE_FIELDS: dict[str, float] = {"molecular_classification": 0.66}


def concept_labels() -> dict[str, str]:
    """Concept id to human label, from the dependency graph nodes the A2 concept ids key on."""
    data = json.loads(mapping.DEPENDENCY_GRAPH_PATH.read_text(encoding="utf-8"))
    return {node["id"]: node["label"] for node in data["nodes"]}


def _reference_note(source_label: str, concept_label: str, relation: Relation) -> str:
    if relation is Relation.DERIVED_FROM:
        return f"CAP states the primitives; the {concept_label} concept is derived from them."
    return f"The CAP '{source_label}' field states the {concept_label} concept directly."


def reference_crosswalk() -> Crosswalk:
    """The committed CAP-checklist crosswalk as `CrosswalkEdge` rows, the deterministic analog of an A3
    model draft.

    Reads `mapping.cap_checklist_coverage()` for the resolved edges and the remaining `fields.FIELDS`
    (the `UNMODELED_CAP_FIELDS`) for any unresolved edges, so the mapping-review surface renders the same
    contract the model pass emits without a live call. Edge order follows the CAP checklist.
    """
    coverage = mapping.cap_checklist_coverage()
    concepts = {c.concept_id: c for c in mapping.compile_concepts()}
    labels = concept_labels()
    edges: list[CrosswalkEdge] = []
    for spec in fields.FIELDS:
        name = spec.name
        if name in coverage:
            concept_id = coverage[name]
            concept = concepts[concept_id]
            relation = (
                Relation.DERIVED_FROM
                if concept.assertion_type is mapping.AssertionType.DERIVED
                else Relation.EXACT
            )
            confidence = _UNCERTAIN_SOURCE_FIELDS.get(name, _RELATION_CONFIDENCE[relation])
            edges.append(
                CrosswalkEdge(
                    source_std=BUILTIN_DICTIONARY_ID,
                    source_field=name,
                    concept_id=concept_id,
                    relation=relation,
                    confidence=confidence,
                    note=_reference_note(spec.label, labels.get(concept_id, concept_id), relation),
                )
            )
        else:
            # A CAP field the concept list does not model. It names no concept and routes to the
            # reviewer queue rather than a forced link (the honesty rule).
            edges.append(
                CrosswalkEdge(
                    source_std=BUILTIN_DICTIONARY_ID,
                    source_field=name,
                    concept_id=None,
                    relation=Relation.UNRESOLVED,
                    confidence=_RELATION_CONFIDENCE[Relation.UNRESOLVED],
                    note=(
                        f"No canonical concept models the CAP '{spec.label}' field; it is captured on "
                        "the report but projected into no concept."
                    ),
                )
            )
    return Crosswalk(source_std=BUILTIN_DICTIONARY_ID, edges=tuple(edges))


def is_queued(edge: CrosswalkEdge) -> bool:
    """Whether an edge is reviewer work: unresolved, or resolved at or below the review threshold. The
    unresolved-or-low-confidence queue the maieutic loop (#95) resolves."""
    return edge.is_unresolved or edge.confidence <= REVIEW_CONFIDENCE_THRESHOLD