src/endopath/induction.py

"""Induce a report's own reporting schema into a strict-JSON contract (#92, the A1 pass).

The A1 pass reads one report and drafts, per report, the strict-JSON contract:

```json
{
  "report_id": "TCGA-AJ-A3NH",
  "detected_standard": "cap_uterus/5.1.0.0",
  "fields": [
    {
      "name": "histologic_type",
      "value_or_absent": "endometrioid adenocarcinoma",
      "reason_if_absent": null,
      "span": {"page": 2, "bbox": [0.12, 0.34, 0.61, 0.38]},
      "confidence": 0.94
    }
  ]
}
```

This is the A1 half of the LLM-native Stage-1 mapping engine. A2 (`src/endopath/mapping.py`)
compiles the target dictionary into the canonical concept list; A1 here induces each report's own
surface schema, which ranges from synoptic to narrative across institutions and eras, into the same
provenance-carrying shape. #93 draws edges from the fields A1 emits to the concepts A2 compiles.

The pass is LLM-native: a language model reads the report together with the canonical registry
(`docs/taxonomy.md`) and drafts the contract. The report's text arrives through the pluggable text
source seam (`src/endopath/textsource.py`), never a hardcoded reader, so the vision-LLM MVP and a
local on-premise model plug in at the same seam. The model call itself goes through a `SchemaInducer`,
a provider-agnostic seam, so the fast test suite runs against an injected fake with no API key.

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

1. No field without its source span. A field that carries a value carries a `span` (a page plus a
   normalized bounding box). A value the pass cannot localize is not emitted as a bare value: the
   assembly raises rather than persisting a value with no evidence (CLAUDE.md invariant 2).
2. Absence is a coded reason. An absent field sets `value_or_absent` to null and `reason_if_absent`
   to a code from the closed `docs/prd.md` section 4.3 vocabulary, never a blank.

The two invariants this pass serves:

1. Numbers stay numbers. The induced `value_or_absent` is the surface value the report itself stated
   (for example "4 foci" or "5 mm in a 21 mm wall"), the most primitive quantity, so a later
   projection can still ask either convention's question. A1 records the surface form; it does not
   categorize it.
2. No value without its evidence. Every emitted value carries its span, and `AbsenceReason` codes the
   reason for every field that carries none.
"""

from __future__ import annotations

import base64
import csv
import json
from enum import Enum
from pathlib import Path
from typing import Optional, Protocol, runtime_checkable

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

from endopath import textsource
from endopath.llm_extraction import get_client
from endopath.textsource import TextRequest

REPO_ROOT = Path(__file__).resolve().parents[2]
REGISTRY_PATH = REPO_ROOT / "docs" / "taxonomy.md"
SOURCES_PATH = REPO_ROOT / "data" / "taxonomy" / "sources.csv"

# The induction model is named explicitly here rather than inherited from the extraction pass,
# which pins its own model (`src/endopath/llm_extraction.py`). Schema induction reads the whole
# registry and reasons over an unfamiliar report structure, so it runs on the most capable model by
# default; a caller overrides it per call. Both the model id and the reader stay swappable.
INDUCTION_MODEL = "claude-sonnet-5"

# The two structural forms `detected_standard` may take beyond a registry `source_id`: a report with
# no synoptic template (`narrative`), or one that mixes a template with narrative prose (`mixed`).
STRUCTURAL_STANDARDS: frozenset[str] = frozenset({"narrative", "mixed"})


class InductionError(ValueError):
    """The induction did not yield a valid contract: a value with no span, an absent field with no
    coded reason, a confidence outside [0, 1], a malformed span, or a detected_standard that names
    neither a registry standard nor a structural form."""


class AbsenceReason(str, Enum):
    """The closed vocabulary a field with no plain value takes (docs/prd.md section 4.3, from the
    HL7 `dataAbsentReason` code system and its two neighbours). Silence is never the response to a
    miss: an absent field always carries one of these."""

    NOT_APPLICABLE = "not_applicable"  # a gate removes the field's meaning here
    ASKED_UNKNOWN = "asked-unknown"  # the field applies, the report was searched, and it is silent
    INDETERMINATE = "indeterminate"  # the report states the feature cannot be assessed
    DETERMINED_BY_JOIN = "determined_by_join"  # the value arrives from another source, not the report


class Span(BaseModel):
    """Where a value came from: a 1-indexed page and a normalized bounding box `[x0, y0, x1, y1]`
    in `[0, 1]`, so a reviewer and a second centre look at the same region of the same page
    (docs/prd.md section 4.6)."""

    model_config = ConfigDict(frozen=True)

    page: int
    bbox: tuple[float, float, float, float]
    # Which specimen within the containing InducedReport this value describes (issue #110).
    # The report itself is already named by the enclosing InducedReport.report_id, so only the
    # specimen dimension is new here. Optional and unpopulated by today's induction pass.
    specimen_id: Optional[str] = None

    @model_validator(mode="after")
    def _page_and_bbox_are_well_formed(self) -> "Span":
        if self.page < 1:
            raise ValueError(f"span page must be 1-indexed, got {self.page}")
        x0, y0, x1, y1 = self.bbox
        if not all(0.0 <= c <= 1.0 for c in self.bbox):
            raise ValueError(f"span bbox must be normalized to [0, 1], got {self.bbox}")
        if x1 < x0 or y1 < y0:
            raise ValueError(f"span bbox must have x0<=x1 and y0<=y1, got {self.bbox}")
        return self


class InducedField(BaseModel):
    """One induced field: the report's own field name, its surface value or a coded absence, the
    span the value came from, and a confidence. The contract row #92 defines."""

    model_config = ConfigDict(frozen=True)

    name: str
    value_or_absent: Optional[str] = None
    reason_if_absent: Optional[AbsenceReason] = None
    span: Optional[Span] = None
    confidence: float

    @model_validator(mode="after")
    def _value_absence_and_span_are_consistent(self) -> "InducedField":
        if not 0.0 <= self.confidence <= 1.0:
            raise ValueError(f"{self.name}: confidence must be in [0, 1], got {self.confidence}")
        present = self.value_or_absent is not None
        if present:
            # No value without its span, and no value that is also coded absent.
            if self.reason_if_absent is not None:
                raise ValueError(
                    f"{self.name}: a field with a value must not also carry a reason_if_absent"
                )
            if self.span is None:
                raise ValueError(
                    f"{self.name}: a value carries no span and is not emitted as a bare value "
                    "(CLAUDE.md invariant 2)"
                )
        else:
            # Absence is a coded reason, never a blank.
            if self.reason_if_absent is None:
                raise ValueError(
                    f"{self.name}: an absent field must carry a coded reason_if_absent, not a blank"
                )
        return self


class InducedReport(BaseModel):
    """The A1 contract for one report: its id, the detected standard, and the induced fields."""

    model_config = ConfigDict(frozen=True)

    report_id: str
    detected_standard: str
    fields: tuple[InducedField, ...]


@runtime_checkable
class SchemaInducer(Protocol):
    """Turns a report plus the canonical registry into the raw contract dict. The seam a local,
    on-premise model or a test fake plugs into: it never surfaces its provider to the call site."""

    def induce(self, *, request: TextRequest, report_text: str, registry: str) -> dict: ...


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

_SPAN_SCHEMA = {
    "type": "object",
    "properties": {
        "page": {"type": "integer", "minimum": 1, "description": "1-indexed page the value appears on."},
        "bbox": {
            "type": "array",
            "items": {"type": "number", "minimum": 0, "maximum": 1},
            "minItems": 4,
            "maxItems": 4,
            "description": "Normalized [x0, y0, x1, y1] region of the page, each coordinate in [0, 1].",
        },
    },
    "required": ["page", "bbox"],
}

INDUCTION_TOOL = {
    "name": "induce_report_schema",
    "description": (
        "Record the induced reporting schema of one pathology report: the detected reporting "
        "standard and, per field, either the surface value with the page region it came from, or a "
        "coded reason it is absent."
    ),
    "input_schema": {
        "type": "object",
        "properties": {
            "detected_standard": {
                "type": "string",
                "description": (
                    "The reporting standard this report follows, inferred from its surface forms. "
                    "Use the registry source_id when the report follows a named synoptic template "
                    "(for example 'cap_uterus/5.1.0.0'), 'narrative' when it has no template, or "
                    "'mixed' when it interleaves a template with narrative prose."
                ),
            },
            "fields": {
                "type": "array",
                "items": {
                    "type": "object",
                    "properties": {
                        "name": {"type": "string", "description": "The report's own field name."},
                        "value_or_absent": {
                            "anyOf": [{"type": "string"}, {"type": "null"}],
                            "description": (
                                "The surface value exactly as the report states it (the most "
                                "primitive quantity, for example '4 foci' or '5 mm'), or null when "
                                "the field carries no plain value."
                            ),
                        },
                        "reason_if_absent": {
                            "anyOf": [
                                {"type": "string", "enum": [r.value for r in AbsenceReason]},
                                {"type": "null"},
                            ],
                            "description": (
                                "Null when value_or_absent is set. Otherwise a coded reason: "
                                "not_applicable (a gate removes its meaning), asked-unknown "
                                "(applies but the report is silent), indeterminate (the report says "
                                "it cannot be assessed), or determined_by_join (it comes from "
                                "another source)."
                            ),
                        },
                        "span": {
                            "anyOf": [_SPAN_SCHEMA, {"type": "null"}],
                            "description": (
                                "Required whenever value_or_absent is set: the page region the value "
                                "came from. A value that cannot be localized is not emitted."
                            ),
                        },
                        "confidence": {"type": "number", "minimum": 0, "maximum": 1},
                    },
                    "required": ["name", "value_or_absent", "reason_if_absent", "span", "confidence"],
                },
            },
        },
        "required": ["detected_standard", "fields"],
    },
}

_INSTRUCTIONS = """\
You induce the reporting schema of one endometrial carcinoma pathology report and call \
induce_report_schema with the result. You do not extract against a fixed checklist; you read the \
report's own structure, which ranges from a synoptic template to free narrative, and record it.

Use the canonical registry below to recognize what each report finding means and which value sets \
it draws from. Report the surface value verbatim, the most primitive quantity the report states, so \
a later step can project it into any convention. Do not categorize or normalize it.

Two rules are absolute:
- No value without its span. Emit a value only with the page and normalized region it came from. If \
you cannot localize a finding to a region, do not emit it as a value.
- Absence is a coded reason. For a field that applies but carries no plain value, set value_or_absent \
to null and reason_if_absent to one of the coded reasons, never a blank.
"""


class AnthropicSchemaInducer:
    """The hosted vision-LLM inducer: it reads the report text (and the rendered pages, when the
    request carries them, so it can localize spans) together with the registry, and returns the raw
    contract through a forced tool call. Names no provider at the call site."""

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

    def induce(self, *, request: TextRequest, report_text: str, registry: str) -> dict:
        client = self._client or get_client()
        system = [
            {"type": "text", "text": _INSTRUCTIONS},
            {
                "type": "text",
                "text": "CANONICAL REGISTRY (docs/taxonomy.md):\n\n" + registry,
                # The registry is identical across every report, so cache it once.
                "cache_control": {"type": "ephemeral"},
            },
        ]
        content: list[dict] = [
            {"type": "text", "text": f"REPORT report_id={request.case_barcode}\n\n{report_text}"}
        ]
        content.extend(_image_blocks(request.page_images))
        content.append({"type": "text", "text": "Call induce_report_schema now."})

        response = client.messages.create(
            model=self._model,
            max_tokens=self._max_tokens,
            system=system,
            tools=[INDUCTION_TOOL],
            tool_choice={"type": "tool", "name": INDUCTION_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 contract.
        if response.stop_reason == "max_tokens":
            raise InductionError(
                "induction 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 InductionError("induction returned no tool call")
        return tool_call.input


_IMAGE_MEDIA_TYPES = {".png": "image/png", ".jpg": "image/jpeg", ".jpeg": "image/jpeg"}


def _image_blocks(page_images: tuple[Path, ...]) -> list[dict]:
    """Base64 image blocks for the rendered pages, so the vision model localizes spans against the
    same pixels a reviewer will see. Skips a page whose bytes cannot be read rather than failing the
    whole induction."""
    blocks: list[dict] = []
    for path in page_images:
        media_type = _IMAGE_MEDIA_TYPES.get(Path(path).suffix.lower())
        if media_type is None:
            continue
        try:
            data = Path(path).read_bytes()
        except OSError:
            continue
        blocks.append(
            {
                "type": "image",
                "source": {
                    "type": "base64",
                    "media_type": media_type,
                    "data": base64.standard_b64encode(data).decode("ascii"),
                },
            }
        )
    return blocks


def load_registry() -> str:
    """The canonical registry the model reads, from docs/taxonomy.md."""
    return REGISTRY_PATH.read_text(encoding="utf-8")


def known_standards() -> frozenset[str]:
    """The registry source_ids a detected_standard may name, from data/taxonomy/sources.csv."""
    with SOURCES_PATH.open(newline="", encoding="utf-8") as handle:
        return frozenset(row["source_id"] for row in csv.DictReader(handle))


def _build_field(item: object) -> InducedField:
    if not isinstance(item, dict):
        raise InductionError(f"each induced field must be a JSON object, got {type(item).__name__}")
    payload = {key: item.get(key) for key in ("name", "value_or_absent", "reason_if_absent", "span", "confidence")}
    try:
        return InducedField(**payload)
    except ValidationError as error:
        raise InductionError(f"field {item.get('name')!r}: {error}") from error


def build_induced_report(
    raw: object,
    *,
    report_id: str,
    standards: Optional[frozenset[str]] = None,
) -> InducedReport:
    """Validate the raw contract dict into an `InducedReport`, enforcing the two hard rules.

    Raises `InductionError` when the output is not a JSON object, names no detected_standard or one
    outside the registry, emits a value with no span, or emits an absent field with no coded reason.
    `report_id` is the case identity, set from the request rather than trusted from the model.
    """
    if not isinstance(raw, dict):
        raise InductionError(f"induction output must be a JSON object, got {type(raw).__name__}")
    detected = raw.get("detected_standard")
    if not isinstance(detected, str) or not detected.strip():
        raise InductionError("induction output must name a detected_standard")
    allowed = standards if standards is not None else known_standards()
    if detected not in allowed and detected not in STRUCTURAL_STANDARDS:
        raise InductionError(
            f"detected_standard {detected!r} names neither a registry standard nor one of "
            f"{sorted(STRUCTURAL_STANDARDS)}"
        )
    raw_fields = raw.get("fields")
    if not isinstance(raw_fields, list):
        raise InductionError("induction output must carry a 'fields' list")
    fields = tuple(_build_field(item) for item in raw_fields)
    return InducedReport(report_id=report_id, detected_standard=detected, fields=fields)


def induce_report(
    request: TextRequest,
    *,
    source_id: str = textsource.DEFAULT_SOURCE_ID,
    page_reader: Optional[textsource.PageReader] = None,
    inducer: Optional[SchemaInducer] = None,
    registry: Optional[str] = None,
    standards: Optional[frozenset[str]] = None,
) -> InducedReport:
    """The A1 pass for one report: obtain its text through the pluggable source, induce the schema,
    and validate it into the contract.

    The report's content arrives through `textsource.select(source_id)` rather than a hardcoded
    reader, so the vision MVP and a local model plug in at that one seam. `inducer` is the model
    seam: the default calls the hosted vision model, a fake substitutes for it in a test. Raises
    `InductionError` when the source yields no text (a schema cannot be induced from nothing) or the
    induced contract violates a rule.
    """
    source = textsource.select(source_id, page_reader=page_reader)
    report_text = source.text_for(request)
    if not report_text:
        raise InductionError(
            f"text source {source_id!r} returned no text for {request.case_barcode!r}; "
            "a schema cannot be induced without report content"
        )
    registry = registry if registry is not None else load_registry()
    inducer = inducer or AnthropicSchemaInducer()
    raw = inducer.induce(request=request, report_text=report_text, registry=registry)
    return build_induced_report(raw, report_id=request.case_barcode, standards=standards)


def contract_json(report: InducedReport) -> str:
    """The induced contract as strict JSON, the #92 artifact."""
    return json.dumps(report.model_dump(mode="json"), indent=2) + "\n"