"""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, }