src/workflows/variant_status.py

"""Curation: source variant/CNV calls -> the gene×sample STATUS matrix.

The data model the whole agent is built on (CLAUDE.md, ADR-0019 §2). Milestone 1.

The status matrix is the single object every tool consumes:
  - rows = genes (the PDAC panel), columns = samples, values = a categorical status.
  - mutation status  ∈ {WT, missense, truncating, hotspot}
  - cnv status       ∈ {deep_del, loss, neutral, gain, amp}
  - the specific variant (e.g. 'KRAS G12D') is retained as PROVENANCE, not the value.

All the messy MAF/VCF/GISTIC parsing lives HERE (at curation), never in the tools.
"""
from __future__ import annotations

import re
from dataclasses import dataclass
from pathlib import Path

import pandas as pd

from . import cbioportal_io, curated_store

# --- classification rules -------------------------------------------------------------

# Hotspot classification is per-gene and must be curated, not guessed. Two complementary
# rules, both traceable: an explicit variant set, plus recurrent-codon patterns so any
# amino-acid substitution at a known hotspot codon (KRAS G12x/G13x/Q61x etc.) is caught.
HOTSPOTS: dict[str, set[str]] = {
    "KRAS": {"G12D", "G12R", "G12V", "G12C", "G12A", "G12S", "G13D", "Q61H", "Q61R", "Q61L"},
    "GNAS": {"R201C", "R201H"},
    "TP53": {"R175H", "R248Q", "R248W", "R273H", "R273C", "R282W"},
}

# Recurrent hotspot CODONS per gene: a substitution at one of these positions is a hotspot
# regardless of the substituted residue (the "G12x / G13x / Q61x" rule from TODO.md).
HOTSPOT_CODONS: dict[str, set[int]] = {
    "KRAS": {12, 13, 61},
    "GNAS": {201},
}

# cBioPortal `mutationType` values that ablate the protein -> "truncating".
TRUNCATING_TYPES = {
    "Nonsense_Mutation",
    "Frame_Shift_Del",
    "Frame_Shift_Ins",
    "Splice_Site",
    "Splice_Region",
    "Translation_Start_Site",
    "Nonstop_Mutation",
}

# Silent / non-coding calls carry no status; treated as WT (never emitted as an alteration).
SILENT_TYPES = {"Silent", "3'UTR", "5'UTR", "Intron", "IGR", "RNA", "5'Flank", "3'Flank"}

# mutation-status precedence: a sample with several calls in one gene takes the strongest.
_MUT_RANK = {"WT": 0, "missense": 1, "truncating": 2, "hotspot": 3}

# discrete GISTIC (-2..2) -> cnv status.
CNV_MAP = {-2: "deep_del", -1: "loss", 0: "neutral", 1: "gain", 2: "amp"}

_PROTEIN_POS = re.compile(r"[A-Za-z]\*?(\d+)")


def is_hotspot(gene: str, protein_change: str) -> bool:
    """True if `protein_change` (e.g. 'G12D') is a curated hotspot for `gene`."""
    if not protein_change:
        return False
    if protein_change in HOTSPOTS.get(gene, ()):  # explicit curated variant
        return True
    m = _PROTEIN_POS.match(protein_change)
    if m and int(m.group(1)) in HOTSPOT_CODONS.get(gene, ()):  # recurrent codon
        return True
    return False


def classify_mutation(gene: str, protein_change: str, mutation_type: str) -> str:
    """One MAF row -> {WT, missense, truncating, hotspot}. Silent -> WT."""
    if mutation_type in SILENT_TYPES:
        return "WT"
    if is_hotspot(gene, protein_change):
        return "hotspot"
    if mutation_type in TRUNCATING_TYPES:
        return "truncating"
    return "missense"


def classify_cnv(gistic_value: int) -> str | None:
    """Discrete GISTIC score (-2..2) -> cnv status, or None if out of range."""
    return CNV_MAP.get(int(gistic_value))


# --- structural variants ---------------------------------------------------------------
#
# The third modality. It exists because the KRAS-wild-type actionable genes — NRG1, NTRK1/2/3,
# ALK, ROS1, RET — are driven in PDAC by FUSIONS, not point mutations or copy number. Carrying
# them on a mutation+CNV panel would have reported ~0% for each, on the most clinically
# actionable question this agent can be asked: a confident false negative, not a low frequency.
#
# sv status ∈ {none, fusion_in_frame, fusion_out_of_frame, intragenic, rearrangement}
SV_STATUSES = ("none", "fusion_in_frame", "fusion_out_of_frame", "intragenic", "rearrangement")

# Precedence when one sample carries several SVs in one gene. This is a SPECIFICITY ordering —
# how much the source told us — and deliberately NOT a severity claim: an intragenic deletion of
# SMAD4 is more consequential than an in-frame fusion of it, and this ranking does not say
# otherwise. It only decides which single label survives into the matrix cell; every event's
# annotation string is retained in `provenance`, so nothing is discarded.
_SV_RANK = {"none": 0, "rearrangement": 1, "intragenic": 2, "fusion_out_of_frame": 3, "fusion_in_frame": 4}

# Only somatic events are in scope (germline is a controlled-access question we do not touch).
SV_SOMATIC = "SOMATIC"


def _sv_text(event_info: str) -> str:
    """Lower-case an SV annotation and flatten the separators the sources disagree about.

    `pdac_msk_2024` writes BOTH "Protein Fusion: in frame" and "The rearrangement is an
    in-frame fusion between genes ATP1B1 Exon2 (NM_001677) and NRG1 Exon2" — the same claim,
    one hyphenated. Matching the spaced form alone read the hyphenated rows as carrying no
    frame claim at all, so all six somatic NRG1 fusions in that cohort — in-frame, named
    partners, one with a PMID — were reported as bare `rearrangement`, on the gene the whole
    modality exists to answer for (found in the 2026-08-05 click-through).

    Because the cohort also had spaced rows, `sv_annotation_depth` still said "characterized"
    and nothing flagged the loss: the under-call was invisible from inside the payload.
    Normalising here (and in the depth gate, which made the identical test) means the two agree
    by construction rather than by both being edited.
    """
    return re.sub(r"[\s_-]+", " ", (event_info or "").strip().lower())


def _frame_claim(text: str) -> str | None:
    """`fusion_out_of_frame` / `fusion_in_frame` / None, from one normalised annotation string."""
    if "out of frame" in text:
        return "fusion_out_of_frame"
    if "in frame" in text:
        return "fusion_in_frame"
    return None


def classify_sv(event_info: str, variant_class: str = "", annotation: str = "") -> str:
    """One structural-variant row -> an sv status.

    Reads cBioPortal's free-text `eventInfo` because that is where the frame information
    actually lives; `variantClass` (TRANSLOCATION / DELETION / …) describes the genomic event,
    not its consequence for the protein, and is `NA` for entire cohorts (every one of CCLE's 153
    panel-gene rows). A classifier keyed on `variantClass` would therefore have produced a
    single undifferentiated bucket for one of the only two cohorts that has this data at all.

    Anything the source does not characterize falls to `rearrangement` — an honest "an SV
    touches this gene here, and we cannot say what it does to the protein" — rather than being
    promoted to a fusion it may not be.
    """
    info = _sv_text(event_info)
    if not info:
        return "rearrangement" if (variant_class or "").strip() else "none"
    if "fusion" in info:
        # "Protein Fusion: in frame" / "out of frame". An antisense fusion is not a protein
        # fusion and must not read as one.
        claim = _frame_claim(info)
        if claim:
            return claim
        if "antisense" in info:
            return "rearrangement"
        # `eventInfo` carries no frame claim — but in `pdac_msk_2024` that does NOT mean the
        # source made none. Its Archer-panel rows put a bare label in `eventInfo`
        # ("ATP1B1-NRG1 Fusion - Archer") and the actual finding in `annotation`: "The
        # rearrangement is an in-frame fusion between genes ATP1B1 Exon2 (NM_001677) and NRG1
        # Exon2 (NM_004495)." Reading `eventInfo` alone reported 10 such events as bare
        # `rearrangement` — 6 NRG1, 2 NTRK3, 1 NTRK1, 1 RET, i.e. the entire KRAS-wild-type
        # actionable tier this modality was added to answer for (found 2026-08-05).
        #
        # `site2EffectOnFrame` looks like the structured way to do this and is not: it is 'NA'
        # on all 110 somatic panel rows in that cohort, the same trap `variantClass` set.
        #
        # Checked before trusting it: across both SV cohorts no `annotation` says "negative",
        # none says "out of frame", and CCLE's never mentions a frame at all — so this can
        # promote an event only where the source states a frame, and cannot promote CCLE.
        return _frame_claim(_sv_text(annotation)) or "rearrangement"
    if "intragenic" in info or "within transcript" in info:
        return "intragenic"
    return "rearrangement"


def sv_annotation_depth(rows: list[dict]) -> str:
    """Whether a cohort's SV calls carry frame annotation at all — a per-STUDY gate.

    ``"characterized"`` — at least one event states in/out of frame, so `fusion_in_frame` in
    this cohort means something. ``"uncharacterized"`` — the source labels events only as
    ``"GENEA-GENEB fusion"`` free text with no frame call, so EVERY event here lands on
    `rearrangement` and no in-frame claim can be made about the cohort at all.

    This is the SV analogue of the per-study modality gate and the per-gene assay gate, and it
    exists because the two cohorts that have SV data differ completely: `pdac_msk_2024` carries
    "Protein Fusion: in frame", while `ccle_broad_2019` carries `variantClass: NA` on all 153
    panel-gene rows and free-text partner labels. Without this distinction a reader comparing
    the two would take "0 in-frame fusions in CCLE" as a finding about cell lines, when it is a
    fact about CCLE's annotation.
    """
    for r in rows:
        # Same normalisation as `classify_sv` — a cohort annotated ONLY in the hyphenated form
        # would otherwise be judged uncharacterized here, and the per-cohort collapse below
        # would then flatten every one of its genuinely in-frame fusions to `rearrangement`.
        # Both fields, for the same reason `classify_sv` reads both: a cohort whose frame calls
        # live only in `annotation` (MSK's Archer rows) is characterized, and judging it
        # otherwise would make the per-cohort collapse below flatten those very events.
        if _frame_claim(_sv_text(r.get("eventInfo"))) or _frame_claim(_sv_text(r.get("annotation"))):
            return "characterized"
    return "uncharacterized"


def sv_gene_symbols(row: dict) -> list[str]:
    """The gene symbols an SV row implicates — BOTH ends of the event.

    A fusion is an event between two loci and cBioPortal reports it once, so a panel gene that
    is the 3' partner appears only in `site2HugoSymbol`. Reading site1 alone loses most NRG1 and
    NTRK3 fusions, which are exactly the events this modality was added for.
    """
    return [s for s in ((row.get("site1HugoSymbol") or ""), (row.get("site2HugoSymbol") or "")) if s]


# --- panel ----------------------------------------------------------------------------

_PANEL_PATH = Path(__file__).resolve().parents[1] / "resources" / "pdac_panel.tsv"


def load_panel() -> list[str]:
    """The v1 driver-gene panel (config, `src/resources/pdac_panel.tsv`)."""
    rows = _PANEL_PATH.read_text().splitlines()[1:]
    return [r.split("\t")[0] for r in rows if r.strip()]


# --- the status matrix ----------------------------------------------------------------


@dataclass
class StatusMatrix:
    """gene×samples status + provenance + which modalities are actually present.

    ``modalities`` records per-modality availability for THIS cohort — the per-study
    coverage gate (a cohort may lack CNV entirely; the spike found one). A tool must
    consult it and refuse/omit an absent modality rather than imply a null answer.
    """

    genes: list[str]
    samples: list[str]
    mutation: object | None  # gene×sample DataFrame of mutation status, or None if absent
    cnv: object | None  # gene×sample DataFrame of cnv status, or None if absent
    provenance: dict  # (gene, sample) -> specific variant string
    modalities: dict  # {"mutation": bool, "cnv": bool, "sv": bool, "expression": bool}
    grounded: bool  # True = curated registered source; False = BYOD upload
    source: str  # e.g. "cbioportal:paad_tcga" | "upload:<file>"
    terms: dict | None = None  # per-study license/access provenance (C2); None for BYOD
    # Which panel genes the cohort's assay actually covers. ``None`` means "no per-gene
    # restriction" — a genome/exome-wide cohort, where every panel gene was interrogated.
    # A list means the cohort is TARGETED and only those genes are answerable; the rest are
    # unassayed, and must be refused rather than reported as 0%.
    genes_assayed: list[str] | None = None
    # Per-modality PROFILED samples — the frequency denominator. ``{}``/absent modality means
    # "no narrower claim than the whole cohort". This is the sample-axis twin of
    # ``genes_assayed``: a sample a study never sequenced is not wild-type, and dividing by the
    # full cohort silently counts it as one (paad_tcga: 186 samples, 150 sequenced -> KRAS
    # reads 73% instead of its true 91%).
    profiled: dict = None  # {"mutation": [ids] | None, "cnv": [ids] | None}
    # Artifact age / upstream-release provenance (ADR-0007). None for BYOD and for a
    # freshly-curated in-memory matrix — an artifact that was never written has no age.
    curation: dict | None = None
    # gene×sample DataFrame of structural-variant status, or None if the cohort has no SV
    # profile. Baseline is "none".
    sv: object | None = None
    # "characterized" | "uncharacterized" | None (no SV modality). Whether this cohort's SV
    # calls carry frame annotation at all — see `sv_annotation_depth`. An uncharacterized cohort
    # can never report `fusion_in_frame`, and every answer must say so rather than let a reader
    # take the absence as biology.
    sv_annotation: str | None = None
    # (gene, sample) -> the SV's annotation string. Kept apart from `provenance` because both
    # are keyed the same way and one sample can carry a mutation AND a fusion in one gene.
    sv_provenance: dict = None

    def __post_init__(self):
        if self.profiled is None:
            self.profiled = {}
        if self.sv_provenance is None:
            self.sv_provenance = {}

    def denominator(self, modality: str) -> list[str]:
        """Samples that count for a `modality` frequency — profiled ∩ this (maybe filtered) cohort.

        Falls back to the full sample list when the source declares no per-modality list, which
        is the only honest reading of "no narrower claim was made".
        """
        ids = self.profiled.get(modality)
        if ids is None:
            return list(self.samples)
        allowed = set(ids)
        return [s for s in self.samples if s in allowed]


class ControlledAccessArtifactError(RuntimeError):
    """A curated artifact declares non-open-access terms, so it is refused at SERVE time.

    Defence in depth for condition C2. Curation already refuses controlled studies, so this
    should be unreachable — but "should be unreachable" is exactly the claim that ages badly
    once artifacts are files a human can add by hand or copy between deployments. The check is
    one comparison; discovering later that we served restricted data is not recoverable.
    """


class NotCuratedError(RuntimeError):
    """A registered study has no curated artifact, and the request path will not fetch live.

    This is a *refusal*, not a failure: ADR-0005 (condition C4) makes the cBioPortal API a
    curation-time dependency only. Falling back to a live fetch here would quietly reintroduce
    exactly the runtime coupling that decision removed, and it would do so under load — the
    worst moment to discover it.
    """


def build_status_matrix(source, *, sample_ids: list[str] | None = None) -> StatusMatrix:
    """Resolve a source (cBioPortal study id string, or a BYODSpec) to a StatusMatrix.

    **Registered path reads a curated artifact — never the API** (ADR-0005 C4). An uncurated
    study raises :class:`NotCuratedError` telling the operator to curate it; see
    :func:`curate_from_cbioportal`, which is the only function here that touches the network.

    ``sample_ids`` restricts the cohort to a subset (used by the lineage filter to keep
    only the pancreatic subset of a pan-cancer source). A ``BYODSpec`` routes to the
    ``grounded=False`` upload path (ADR-0003) and is unaffected by caching — an upload is
    curated in-process, per request, and never persisted.
    """
    # BYOD upload (ADR-0003) — a declared MAF/CNV spec, not a study id. Lazy import: byod
    # depends on this module, so importing it at top level would be circular.
    if not isinstance(source, str):
        from .byod import BYODSpec, curate_byod_upload

        if isinstance(source, BYODSpec):
            return curate_byod_upload(source)
        raise TypeError(f"Unsupported source type {type(source).__name__}.")

    if not source.startswith("cbioportal:"):
        raise NotImplementedError(f"Only cbioportal: and BYOD sources are implemented. Got {source!r}.")
    study = source.split(":", 1)[1]

    if not curated_store.is_curated(study):
        raise NotCuratedError(
            f"Study {study!r} has not been curated for this deployment, so it cannot be "
            f"answered offline. Available: {', '.join(curated_store.list_curated()) or '(none)'}. "
            f"An operator curates it with `python -m src.curate {study}`; the request path "
            "never calls the cBioPortal API (ADR-0005)."
        )
    payload = curated_store.read(study)
    terms = payload.get("terms") or {}
    if terms and terms.get("access_tier") != "open":
        raise ControlledAccessArtifactError(
            f"Curated artifact for {study!r} declares access_tier="
            f"{terms.get('access_tier')!r}; only open-access studies may be served "
            "(ADR-0005 condition C2). Re-curate or remove the artifact."
        )
    return _from_artifact(payload, restrict=sample_ids)


def _from_artifact(payload: dict, *, restrict: list[str] | None) -> StatusMatrix:
    """Rebuild a dense StatusMatrix from a sparse curated artifact. No network, no parsing."""
    genes = payload["panel_genes"]
    samples = list(restrict) if restrict is not None else list(payload["samples"])
    known = set(payload["samples"])
    # A restrict list may name samples the artifact does not have (e.g. a stale lineage list);
    # keep only real ones so the frame can never carry a phantom column.
    samples = [s for s in samples if s in known]

    def _dense(sparse: dict, baseline: str):
        frame = pd.DataFrame(baseline, index=genes, columns=samples)
        for gene, row in sparse.items():
            if gene not in frame.index:
                continue
            for sample, value in row.items():
                if sample in frame.columns:
                    frame.at[gene, sample] = value
        return frame

    modalities = payload["modalities"]
    sample_set = set(samples)

    def _prov(key_name: str) -> dict:
        out = {}
        for key, value in (payload.get(key_name) or {}).items():
            gene, _, sample = key.partition("\t")
            if sample in sample_set:
                out[(gene, sample)] = value
        return out

    provenance = _prov("provenance")

    return StatusMatrix(
        genes=genes,
        samples=samples,
        mutation=_dense(payload["mutation"], "WT") if modalities.get("mutation") else None,
        cnv=_dense(payload["cnv"], "neutral") if modalities.get("cnv") else None,
        sv=_dense(payload.get("sv") or {}, "none") if modalities.get("sv") else None,
        sv_annotation=payload.get("sv_annotation"),
        sv_provenance=_prov("sv_provenance"),
        provenance=provenance,
        modalities=modalities,
        grounded=True,
        source=payload["source"],
        terms=payload.get("terms"),
        genes_assayed=payload.get("genes_assayed"),
        profiled=payload.get("profiled") or {},
        curation=curated_store.freshness_from_payload(payload),
    )


def curate_from_cbioportal(study: str, source: str | None = None, *, restrict=None) -> StatusMatrix:
    """**Curation time only** — fetch from the cBioPortal API and classify.

    The single network-touching entry point in this module. Called by `python -m src.curate`
    and by tests that patch the client; never by a request handler.
    """
    return _build_from_cbioportal(study, source or f"cbioportal:{study}", restrict=restrict)


def assayed_panel_genes(study: str, genes: list[str]) -> list[str] | None:
    """Which of `genes` a study's assay actually interrogates — curation time only.

    Returns ``None`` for a genome/exome-wide cohort (TCGA, CCLE, QCMG, UTSW, CPTAC): there is no
    per-gene restriction to record, and every panel gene is answerable. Returns a **list** for a
    TARGETED cohort, where a gene off the panel was never looked at.

    Targeted cohorts are detected by the `GENE_PANEL` sample attribute, which is how cBioPortal
    records that a sample was sequenced with a named panel (MSK-IMPACT and friends). The result
    is the INTERSECTION across every panel version in the cohort, not the union: a gene present
    on IMPACT505 but absent from IMPACT341 has a denominator that is not the cohort, and quietly
    dividing by the cohort anyway is the failure mode this whole field exists to prevent. Where
    the versions agree — as all four IMPACT versions do on this panel — the distinction is moot.

    Failing to resolve a panel is deliberately fatal rather than shrugged off: silently falling
    back to "everything assayed" would reinstate exactly the 0%-for-an-unlooked-at-gene lie.
    """
    rows = cbioportal_io.clinical_data(study, ["GENE_PANEL"])
    panels = {(r.get("value") or "").strip() for r in rows}
    panels.discard("")
    if not panels:
        return None  # not a targeted cohort — nothing to restrict
    covered = [cbioportal_io.gene_panel_genes(p) for p in sorted(panels)]
    shared = set.intersection(*covered)
    return [g for g in genes if g in shared]


def _build_from_cbioportal(study: str, source: str, *, restrict: list[str] | None) -> StatusMatrix:
    genes = load_panel()
    entrez = cbioportal_io.entrez_ids(genes)
    # entrez -> symbol, keeping only panel genes the source actually knows.
    by_entrez = {eid: sym for sym, eid in entrez.items()}
    entrez_ids_list = list(by_entrez)

    mut_profile = cbioportal_io.profile_id(study, "MUTATION_EXTENDED")
    cnv_profile = cbioportal_io.profile_id(study, "COPY_NUMBER_ALTERATION", "DISCRETE")
    sv_profile = cbioportal_io.profile_id(study, "STRUCTURAL_VARIANT")
    expr_profile = cbioportal_io.profile_id(study, "MRNA_EXPRESSION")
    modalities = {
        "mutation": mut_profile is not None,
        "cnv": cnv_profile is not None,  # the per-study coverage gate lives here
        # Same gate, third modality — and it fires far more often: of the seven curated cohorts
        # only two (pdac_msk_2024, ccle_broad_2019) publish SV at all. The other five must yield
        # no fusion answer rather than a reassuring zero.
        "sv": sv_profile is not None,
        "expression": expr_profile is not None,
    }

    sample_list = cbioportal_io.default_sample_list(study)
    samples = list(restrict) if restrict is not None else cbioportal_io.sample_ids(sample_list)
    sample_set = set(samples)

    provenance: dict = {}

    mutation_df = None
    if mut_profile:
        mutation_df = pd.DataFrame("WT", index=genes, columns=samples)
        rows = cbioportal_io.fetch_mutations(mut_profile, entrez_ids_list, sample_list)
        for r in rows:
            sym = by_entrez.get(r["entrezGeneId"])
            sid = r["sampleId"]
            if sym is None or sid not in sample_set:
                continue
            status = classify_mutation(sym, r.get("proteinChange") or "", r.get("mutationType") or "")
            if status == "WT":
                continue
            # keep the strongest call per (gene, sample); retain its variant as provenance
            if _MUT_RANK[status] >= _MUT_RANK[mutation_df.at[sym, sid]]:
                mutation_df.at[sym, sid] = status
                pc = r.get("proteinChange") or r.get("mutationType") or "?"
                provenance[(sym, sid)] = f"{sym} {pc}"

    cnv_df = None
    if cnv_profile:
        cnv_df = pd.DataFrame("neutral", index=genes, columns=samples)
        rows = cbioportal_io.fetch_molecular_data(cnv_profile, entrez_ids_list, sample_list)
        for r in rows:
            sym = by_entrez.get(r["entrezGeneId"])
            sid = r["sampleId"]
            if sym is None or sid not in sample_set:
                continue
            status = classify_cnv(r["value"])
            if status is None:
                continue
            cnv_df.at[sym, sid] = status

    sv_df = None
    sv_annotation = None
    sv_provenance: dict = {}
    if sv_profile:
        sv_df = pd.DataFrame("none", index=genes, columns=samples)
        rows = [
            r
            for r in cbioportal_io.fetch_structural_variants(sv_profile, entrez_ids_list)
            # Somatic only. The SV endpoint returns whatever the study deposited, and a germline
            # event is a controlled-access question this agent does not answer.
            if (r.get("svStatus") or "").upper() == SV_SOMATIC
        ]
        sv_annotation = sv_annotation_depth(rows)
        for r in rows:
            sid = r.get("sampleId")
            if sid not in sample_set:
                continue
            status = classify_sv(
                r.get("eventInfo") or "",
                r.get("variantClass") or "",
                r.get("annotation") or "",
            )
            if status == "none":
                continue
            # An uncharacterized cohort cannot support a frame claim about ANY of its events,
            # even the odd row that happens to mention a frame. Collapsing to `rearrangement`
            # here keeps the cohort internally consistent, so a frequency is never a mix of
            # "genuinely in-frame" and "we could not tell".
            if sv_annotation == "uncharacterized":
                status = "rearrangement"
            # BOTH ends: a panel gene may be the 3' partner and appear only in site2.
            for sym in sv_gene_symbols(r):
                if sym not in sv_df.index:
                    continue
                if _SV_RANK[status] >= _SV_RANK[sv_df.at[sym, sid]]:
                    sv_df.at[sym, sid] = status
                    label = (r.get("annotation") or r.get("eventInfo") or "?").strip()
                    # A SEPARATE provenance dict, not the mutation one. The two are keyed
                    # identically on (gene, sample), and a gene can carry both a point mutation
                    # and a fusion in the same sample — sharing the dict would let whichever
                    # curation step ran last overwrite the other's variant string, so KRAS G12D
                    # could silently be reported as the provenance of a rearrangement.
                    sv_provenance[(sym, sid)] = label

    return StatusMatrix(
        genes=genes,
        samples=samples,
        mutation=mutation_df,
        cnv=cnv_df,
        sv=sv_df,
        sv_annotation=sv_annotation,
        sv_provenance=sv_provenance,
        provenance=provenance,
        modalities=modalities,
        grounded=True,
        source=source,
        genes_assayed=assayed_panel_genes(study, genes),
        profiled={
            m: cbioportal_io.profiled_sample_ids(study, m)
            for m, present in modalities.items()
            if present and m in cbioportal_io.PROFILED_SAMPLE_LISTS
        },
    )