icassp2027-experiment/evaluate_egospeak_opportunity_events.py

#!/usr/bin/env python3
"""Evaluate opportunity-conditioned ENTRY/YIELD from dense wearer forecasts.

Semantic review labels define which conversational opportunities end in a
wearer floor action.  The evaluated score remains the model's original dense
wearer-activity probability; this script does not train or assume a semantic
event head.

ENTRY opportunities are anchored at a reviewed preceding non-wearer release.
Positive opportunities resolve at the wearer entry; core hard negatives
resolve when the same or a different non-wearer speaks next.  A positive is
excluded if another non-wearer onset competes before (or exactly at) its wearer
entry, or if wearer activity already occurred after the selected release.  The
score is the maximum wearer probability over available 0.2--2.0 s forecasts up
to the uncontested resolution.

YIELD opportunities are anchored at a non-wearer onset during wearer speech.
The score is the maximum future wearer-inactivity probability over 0.2--0.6 s,
censored at that non-wearer's offset.  Reviews with multiple eligible
competing onsets are excluded until the annotation selects one.  Positive and
hard-negative examples therefore begin at the same kind of unambiguous
observed conversational opportunity.

Prediction arrays use the repository contract
``[target_time, class, future_horizon]``.  All scores are read diagonally from
one causal source row.  No input after the labeled outcome is used.
"""

from __future__ import annotations

import argparse
from collections import Counter
import csv
from dataclasses import dataclass
import json
import math
from pathlib import Path
from typing import Any, Iterable, Literal

import numpy as np

from evaluate_annotated_floor_events import (
    PROVISIONAL_LLM_SOURCE,
    _read_jsonl,
    _review_payload,
    load_session_predictions,
)
from evaluate_floor_state_changes import (
    CLASS_NAMES,
    NUM_HORIZONS,
    OFFICIAL_TEST_SESSIONS,
    OFFICIAL_TRAIN_SESSIONS,
    STEP_SECONDS,
    WEARER_INDEX,
    _operating_points,
    _safe_ap,
    parse_runs,
    validate_recalls,
)


Family = Literal["entry", "int"]
FAMILIES: tuple[Family, ...] = ("entry", "int")
ENTRY_CORE_HARD_NEGATIVES = frozenset(
    {"ENTRY_OTHER_MIDTURN_PAUSE", "ENTRY_OTHER_TO_OTHER"}
)
ENTRY_DIAGNOSTIC_HARD_NEGATIVES = frozenset({"ENTRY_WEARER_BACKCHANNEL"})
INT_HARD_NEGATIVES = frozenset(
    {
        "INT_OTHER_BACKCHANNEL",
        "INT_NON_TAKEOVER_OVERLAP",
        "INT_WEARER_HOLDS_FLOOR",
    }
)
ENTRY_MINED_SOURCES = {
    "ENTRY_OTHER_MIDTURN_PAUSE": "SAME_SPEAKER_PAUSE",
    "ENTRY_OTHER_TO_OTHER": "OTHER_TO_OTHER",
}
INT_MINED_SOURCES = frozenset({"OTHER_BACKCHANNEL", "NON_TAKEOVER_OVERLAP"})


@dataclass(frozen=True)
class OpportunityEvent:
    uid: str
    session: str
    family: Family
    positive: bool
    subtype: str
    candidate_source: str
    opportunity_anchor_seconds: float
    resolution_seconds: float
    censor_seconds: float
    behavior_anchor_seconds: float | None
    anchor_basis: str
    competing_intervals: int = 0


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--reviews", type=Path, action="append", required=True)
    parser.add_argument(
        "--run",
        action="append",
        required=True,
        metavar="NAME=RESULT_DIR",
        help="Inference directory containing pred/ and valid/. Repeat per method.",
    )
    parser.add_argument("--sessions", nargs="+", default=list(OFFICIAL_TEST_SESSIONS))
    parser.add_argument("--entry-max-seconds", type=float, default=2.0)
    parser.add_argument("--yield-max-seconds", type=float, default=0.6)
    parser.add_argument("--calibration-reviews", type=Path, action="append")
    parser.add_argument("--calibration-run", action="append", metavar="NAME=RESULT_DIR")
    parser.add_argument("--calibration-sessions", nargs="+")
    parser.add_argument("--matched-recalls", nargs="*", type=float, default=())
    parser.add_argument("--allow-in-sample-thresholds", action="store_true")
    parser.add_argument(
        "--output",
        type=Path,
        default=Path("results/easycom_egospeak_opportunity_events/summary.json"),
    )
    parser.add_argument("--csv-output", type=Path)
    parser.add_argument("--event-csv-output", type=Path)
    parser.add_argument("--print-event-examples", type=int, default=0, metavar="N")
    parser.add_argument("--allow-provisional-llm-reviews", action="store_true")
    return parser.parse_args()


def _finite_seconds(value: Any, *, field: str, uid: str) -> float:
    if isinstance(value, bool) or not isinstance(value, (int, float)):
        raise ValueError(f"{uid}: {field} must be numeric")
    result = float(value)
    if not math.isfinite(result) or result < 0:
        raise ValueError(f"{uid}: {field} must be finite and non-negative")
    return result


def _optional_seconds(value: Any, *, field: str, uid: str) -> float | None:
    if value is None:
        return None
    return _finite_seconds(value, field=field, uid=uid)


def _behavior_anchor(
    candidate: dict[str, Any],
    review: dict[str, Any],
    *,
    uid: str,
) -> float:
    evidence = candidate.get("temporal_evidence")
    evidence = evidence if isinstance(evidence, dict) else {}
    raw = evidence.get(
        "behavior_anchor_seconds",
        evidence.get(
            "wearer_offset_seconds",
            candidate.get("anchor_dense_seconds", candidate.get("event_time")),
        ),
    )
    result = _finite_seconds(raw, field="behavior_anchor_seconds", uid=uid)
    correction = float(review.get("boundary_correction_seconds") or 0.0)
    corrected = result + correction
    if corrected < 0:
        raise ValueError(f"{uid}: corrected behavior anchor precedes session")
    return corrected


def _candidate_region(
    candidate: dict[str, Any], *, uid: str
) -> tuple[float, float]:
    anchor = candidate.get("anchor_dense_seconds", candidate.get("event_time"))
    start = _finite_seconds(
        candidate.get("region_start_seconds", anchor),
        field="region_start_seconds",
        uid=uid,
    )
    end = _finite_seconds(
        candidate.get("region_end_seconds", anchor),
        field="region_end_seconds",
        uid=uid,
    )
    if end < start:
        raise ValueError(f"{uid}: opportunity region ends before it starts")
    return start, end


def _entry_selected_option(
    candidate: dict[str, Any],
    review: dict[str, Any],
    *,
    selected_release_seconds: float,
    uid: str,
) -> dict[str, Any] | None:
    evidence = candidate.get("temporal_evidence")
    evidence = evidence if isinstance(evidence, dict) else {}
    options = evidence.get("opportunity_options")
    if not isinstance(options, list):
        return None
    selected_uid = review.get("selected_opportunity_uid")
    matches: list[dict[str, Any]] = []
    for index, option in enumerate(options):
        if not isinstance(option, dict):
            continue
        option_seconds = _optional_seconds(
            option.get("anchor_seconds"),
            field=f"opportunity_options[{index}].anchor_seconds",
            uid=uid,
        )
        if selected_uid is not None and option.get("uid") == selected_uid:
            matches.append(option)
        elif selected_uid is None and option_seconds is not None and math.isclose(
            option_seconds, selected_release_seconds, abs_tol=1e-7
        ):
            matches.append(option)
    if len(matches) > 1:
        raise ValueError(f"{uid}: selected ENTRY opportunity is not unique")
    return matches[0] if matches else None


def _entry_competing_onsets(
    candidate: dict[str, Any],
    *,
    selected_release_seconds: float,
    behavior_anchor_seconds: float,
    uid: str,
) -> list[float]:
    """Return other onsets that resolve a release before the wearer entry."""

    evidence = candidate.get("temporal_evidence")
    evidence = evidence if isinstance(evidence, dict) else {}
    raw_intervals = evidence.get("gap_other_intervals")
    if not isinstance(raw_intervals, list):
        raw_intervals = []
    onsets: set[float] = set()
    for index, interval in enumerate(raw_intervals):
        if not isinstance(interval, dict):
            continue
        onset = _optional_seconds(
            interval.get("onset_seconds"),
            field=f"gap_other_intervals[{index}].onset_seconds",
            uid=uid,
        )
        if onset is None:
            continue
        if (
            onset >= selected_release_seconds - 1e-7
            and onset <= behavior_anchor_seconds + 1e-7
        ):
            onsets.add(onset)
    return sorted(onsets)


def _int_competing_intervals(
    candidate: dict[str, Any],
    *,
    behavior_anchor_seconds: float,
    uid: str,
) -> list[tuple[float, float]]:
    evidence = candidate.get("temporal_evidence")
    evidence = evidence if isinstance(evidence, dict) else {}
    raw_intervals = evidence.get("nearby_other_intervals")
    if not isinstance(raw_intervals, list):
        return []
    eligible: list[tuple[float, float]] = []
    for index, interval in enumerate(raw_intervals):
        if not isinstance(interval, dict):
            continue
        if interval.get("crosses_offset_with_tolerance") is not True:
            continue
        onset = _optional_seconds(
            interval.get("onset_seconds"),
            field=f"nearby_other_intervals[{index}].onset_seconds",
            uid=uid,
        )
        offset = _optional_seconds(
            interval.get("offset_seconds"),
            field=f"nearby_other_intervals[{index}].offset_seconds",
            uid=uid,
        )
        if onset is None or offset is None:
            continue
        if onset < behavior_anchor_seconds and offset > onset:
            eligible.append((onset, offset))
    return sorted(set(eligible))


def load_opportunity_events(
    paths: Iterable[Path],
    sessions: Iterable[str],
    *,
    allow_provisional_llm: bool = False,
) -> tuple[list[OpportunityEvent], dict[str, Any]]:
    selected_sessions = set(sessions)
    events: list[OpportunityEvent] = []
    seen: set[str] = set()
    skipped: Counter[str] = Counter()
    provisional = 0

    for row_number, row in enumerate(_read_jsonl(paths), 1):
        uid = str(row.get("candidate_id") or row.get("candidate_uid") or f"row-{row_number}")
        if uid in seen:
            raise ValueError(f"Duplicate reviewed candidate: {uid}")
        seen.add(uid)
        session = str(row.get("session", ""))
        if session not in selected_sessions:
            continue
        review, candidate = _review_payload(
            row, uid, allow_provisional_llm=allow_provisional_llm
        )
        review_source = row.get("review_source", review.get("review_source"))
        if review_source == PROVISIONAL_LLM_SOURCE:
            provisional += 1
        verdict = review.get("human_verdict")
        if verdict == "EXCLUDE":
            skipped["review_exclude"] += 1
            continue
        if verdict not in {"POSITIVE", "HARD_NEGATIVE"}:
            raise ValueError(f"{uid}: invalid review verdict {verdict!r}")
        family_value = row.get("candidate_family")
        if family_value not in FAMILIES:
            raise ValueError(f"{uid}: invalid candidate_family {family_value!r}")
        family: Family = family_value
        positive = verdict == "POSITIVE"
        expected_event_type = "WEARER_ENTRY" if family == "entry" else "INT_YIELD"
        if positive and review.get("human_event_type") != expected_event_type:
            raise ValueError(f"{uid}: positive {family} must be {expected_event_type}")
        subtype_value = (
            "WEARER_ENTRY" if family == "entry" and positive
            else "INT_YIELD" if family == "int" and positive
            else review.get("human_subtype")
        )
        if not isinstance(subtype_value, str):
            raise ValueError(f"{uid}: missing subtype")
        source = str(candidate.get("candidate_source", ""))
        behavior = _behavior_anchor(candidate, review, uid=uid)

        if family == "entry":
            if positive:
                selected_release = _optional_seconds(
                    review.get("selected_opportunity_seconds"),
                    field="selected_opportunity_seconds",
                    uid=uid,
                )
                if selected_release is None:
                    skipped["entry_positive_without_selected_release"] += 1
                    continue
                selected_option = _entry_selected_option(
                    candidate,
                    review,
                    selected_release_seconds=selected_release,
                    uid=uid,
                )
                if (
                    selected_option is not None
                    and selected_option.get(
                        "wearer_active_between_release_and_onset"
                    ) is True
                ):
                    skipped["entry_positive_post_outcome_wearer_activity"] += 1
                    continue
                competing_onsets = _entry_competing_onsets(
                    candidate,
                    selected_release_seconds=selected_release,
                    behavior_anchor_seconds=behavior,
                    uid=uid,
                )
                option_reports_competitor = (
                    selected_option is not None
                    and selected_option.get(
                        "other_onset_between_release_and_onset"
                    ) is True
                )
                if option_reports_competitor and not competing_onsets:
                    skipped["entry_positive_competing_onset_unresolved"] += 1
                    continue
                if competing_onsets:
                    skipped["entry_positive_competed_before_behavior"] += 1
                    continue
                anchor = selected_release
                resolution = behavior
                censor = behavior
                basis = "reviewed_preceding_other_release"
            elif subtype_value in ENTRY_CORE_HARD_NEGATIVES:
                if source != ENTRY_MINED_SOURCES[subtype_value]:
                    skipped["entry_hn_without_matching_mined_release"] += 1
                    continue
                anchor, resolution = _candidate_region(candidate, uid=uid)
                censor = resolution
                basis = "mined_nonwearer_release"
            elif subtype_value in ENTRY_DIAGNOSTIC_HARD_NEGATIVES:
                skipped["entry_backchannel_separate_diagnostic"] += 1
                continue
            else:
                skipped["entry_unsupported_hard_negative"] += 1
                continue
            competing_intervals = 0
        else:
            if positive:
                intervals = _int_competing_intervals(
                    candidate,
                    behavior_anchor_seconds=behavior,
                    uid=uid,
                )
                if not intervals:
                    skipped["int_positive_without_crossing_other"] += 1
                    continue
                if len(intervals) > 1:
                    skipped["int_positive_ambiguous_competing_others"] += 1
                    continue
                anchor, censor = intervals[0]
                competing_intervals = 1
                resolution = behavior
                basis = "unique_other_onset_crossing_wearer_offset"
            elif subtype_value in INT_HARD_NEGATIVES:
                if source in INT_MINED_SOURCES:
                    anchor, censor = _candidate_region(candidate, uid=uid)
                    resolution = censor
                    competing_intervals = 1
                    basis = "mined_competing_other_onset"
                else:
                    intervals = _int_competing_intervals(
                        candidate,
                        behavior_anchor_seconds=behavior,
                        uid=uid,
                    )
                    if not intervals:
                        skipped["int_hn_without_competing_other"] += 1
                        continue
                    if len(intervals) > 1:
                        skipped["int_hn_ambiguous_competing_others"] += 1
                        continue
                    anchor, censor = intervals[0]
                    competing_intervals = 1
                    resolution = behavior
                    basis = "unique_other_onset_crossing_non_yield_offset"
            else:
                skipped["int_unsupported_hard_negative"] += 1
                continue

        if resolution <= anchor:
            skipped[f"{family}_nonpositive_opportunity_duration"] += 1
            continue
        if censor <= anchor:
            skipped[f"{family}_nonpositive_censor_duration"] += 1
            continue
        events.append(
            OpportunityEvent(
                uid=uid,
                session=session,
                family=family,
                positive=positive,
                subtype=subtype_value,
                candidate_source=source,
                opportunity_anchor_seconds=anchor,
                resolution_seconds=resolution,
                censor_seconds=censor,
                behavior_anchor_seconds=behavior if positive else None,
                anchor_basis=basis,
                competing_intervals=competing_intervals,
            )
        )

    diagnostics: dict[str, Any] = {
        "num_review_rows": len(seen),
        "num_opportunity_events": len(events),
        "num_provisional_llm": provisional,
        "skipped": dict(sorted(skipped.items())),
    }
    return events, diagnostics


def _first_causal_source_row(anchor_seconds: float) -> tuple[int, float]:
    """First 5-Hz state whose right-edge cutoff observes the opportunity."""

    cutoff_step = int(math.ceil(anchor_seconds / STEP_SECONDS - 1e-7))
    source_row = cutoff_step - 1
    return source_row, cutoff_step * STEP_SECONDS


def _score_event(
    event: OpportunityEvent,
    arrays: dict[str, tuple[np.ndarray, np.ndarray]],
    *,
    entry_max_seconds: float,
    yield_max_seconds: float,
) -> tuple[float, dict[str, Any]] | None:
    prediction, valid = arrays[event.session]
    source_row, source_cutoff = _first_causal_source_row(
        event.opportunity_anchor_seconds
    )
    if source_row < 0 or source_row >= len(prediction):
        return None
    # A post-opportunity state that already contains the positive behavior is
    # outcome observation, not prediction.  Drop these quantized fast events.
    if event.positive and event.behavior_anchor_seconds is not None:
        if source_cutoff >= event.behavior_anchor_seconds - 1e-7:
            return None

    max_seconds = entry_max_seconds if event.family == "entry" else yield_max_seconds
    # A positive ENTRY may use the 5-Hz target bin containing the wearer onset.
    # Hard-negative ENTRY and both YIELD classes stop strictly at their
    # competing non-wearer boundary, as required by competing-risk censoring.
    censor_limit = event.censor_seconds
    if event.family == "entry" and event.positive:
        # Quantize only to the right edge of the 5-Hz bin containing the
        # onset.  An onset already on-grid must not admit one post-onset bin.
        censor_steps = int(math.ceil(censor_limit / STEP_SECONDS - 1e-7))
        censor_limit = censor_steps * STEP_SECONDS
    horizon_limit = min(source_cutoff + max_seconds, censor_limit)
    probabilities: list[float] = []
    used_horizons: list[float] = []
    target_rows: list[int] = []
    for horizon_index in range(NUM_HORIZONS):
        horizon_seconds = (horizon_index + 1) * STEP_SECONDS
        target_seconds = source_cutoff + horizon_seconds
        if target_seconds > horizon_limit + 1e-7:
            continue
        target_row = source_row + horizon_index + 1
        if target_row >= len(prediction) or not valid[target_row, horizon_index]:
            return None
        probabilities.append(
            float(prediction[target_row, WEARER_INDEX, horizon_index])
        )
        used_horizons.append(horizon_seconds)
        target_rows.append(target_row)
    if not probabilities:
        return None
    wearer = np.asarray(probabilities, dtype=np.float64)
    action_scores = wearer if event.family == "entry" else 1.0 - wearer
    best_index = int(np.argmax(action_scores))
    return float(action_scores[best_index]), {
        "source_row": source_row,
        "source_cutoff_seconds": source_cutoff,
        "used_horizons_seconds": used_horizons,
        "target_rows": target_rows,
        "wearer_probabilities": probabilities,
        "best_horizon_seconds": used_horizons[best_index],
        "best_wearer_probability": probabilities[best_index],
    }


def evaluate_family(
    events: list[OpportunityEvent],
    arrays: dict[str, tuple[np.ndarray, np.ndarray]],
    family: Family,
    recalls: list[float],
    *,
    entry_max_seconds: float,
    yield_max_seconds: float,
    fixed_thresholds: dict[float, float | None] | None = None,
    threshold_source: str = "not_requested",
) -> tuple[dict[str, Any], list[dict[str, Any]]]:
    rows: list[dict[str, Any]] = []
    invalid = 0
    for event in events:
        if event.family != family:
            continue
        scored = _score_event(
            event,
            arrays,
            entry_max_seconds=entry_max_seconds,
            yield_max_seconds=yield_max_seconds,
        )
        if scored is None:
            invalid += 1
            continue
        score, details = scored
        rows.append(
            {
                "uid": event.uid,
                "session": event.session,
                "family": event.family,
                "positive": event.positive,
                "subtype": event.subtype,
                "candidate_source": event.candidate_source,
                "score": score,
                "opportunity_anchor_seconds": event.opportunity_anchor_seconds,
                "resolution_seconds": event.resolution_seconds,
                "censor_seconds": event.censor_seconds,
                "behavior_anchor_seconds": event.behavior_anchor_seconds,
                "anchor_basis": event.anchor_basis,
                "competing_intervals": event.competing_intervals,
                **details,
            }
        )

    scores = np.asarray([row["score"] for row in rows], dtype=np.float64)
    positive = np.asarray([row["positive"] for row in rows], dtype=bool)
    negative = ~positive
    labels = positive.astype(np.int8)
    subtypes = sorted({str(row["subtype"]) for row in rows if not row["positive"]})
    extra_groups = {
        f"fpr_{subtype.lower()}": np.asarray(
            [not row["positive"] and row["subtype"] == subtype for row in rows],
            dtype=bool,
        )
        for subtype in subtypes
    }
    operating_points: list[dict[str, Any]] = []
    if recalls:
        operating_points = _operating_points(
            scores,
            positive,
            negative,
            recalls,
            extra_groups=extra_groups,
            fixed_thresholds=fixed_thresholds,
            threshold_source=threshold_source,
        )
    metric = {
        "family": family,
        "num_positive": int(positive.sum()),
        "num_hard_negative": int(negative.sum()),
        "num_invalid_or_post_outcome": invalid,
        "average_precision": _safe_ap(labels, scores),
        "operating_points": operating_points,
    }
    return metric, rows


def _thresholds(metric: dict[str, Any]) -> dict[float, float | None]:
    return {
        float(item["requested_positive_recall"]): item["threshold"]
        for item in metric["operating_points"]
    }


def _positive_grid_seconds(value: float, *, field: str) -> float:
    if not math.isfinite(value) or value <= 0:
        raise ValueError(f"{field} must be finite and positive")
    steps = round(value / STEP_SECONDS)
    if not math.isclose(value, steps * STEP_SECONDS, abs_tol=1e-7):
        raise ValueError(f"{field} must lie on the {STEP_SECONDS:g}-s grid")
    if value > NUM_HORIZONS * STEP_SECONDS:
        raise ValueError(f"{field} cannot exceed {NUM_HORIZONS * STEP_SECONDS:g} s")
    return float(value)


def _validate_calibration(
    args: argparse.Namespace,
    runs: list[tuple[str, Path]],
    recalls: list[float],
) -> tuple[list[tuple[str, Path]], bool]:
    paired = bool(args.calibration_reviews) == bool(args.calibration_sessions)
    if not paired:
        raise ValueError(
            "--calibration-reviews and --calibration-sessions must be supplied together"
        )
    has_calibration = bool(args.calibration_reviews and args.calibration_sessions)
    if args.calibration_run and not has_calibration:
        raise ValueError("--calibration-run requires calibration reviews and sessions")
    if has_calibration:
        unknown = sorted(set(args.calibration_sessions) - set(OFFICIAL_TRAIN_SESSIONS))
        if unknown:
            raise ValueError(f"Calibration must use Sessions 4--12, got {unknown}")
        overlap = sorted(set(args.sessions) & set(args.calibration_sessions))
        if overlap:
            raise ValueError(f"Evaluation and calibration sessions overlap: {overlap}")
    official_test = bool(set(args.sessions) & set(OFFICIAL_TEST_SESSIONS))
    if recalls and official_test and not has_calibration and not args.allow_in_sample_thresholds:
        raise ValueError(
            "Official Sessions 1--3 require training-side calibration for "
            "--matched-recalls; omit it for threshold-free AP"
        )
    calibration_runs = parse_runs(args.calibration_run) if args.calibration_run else runs
    if [name for name, _ in calibration_runs] != [name for name, _ in runs]:
        raise ValueError("--calibration-run names/order must match --run")
    return calibration_runs, has_calibration


def _write_metrics(path: Path, results: list[dict[str, Any]]) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    fields = [
        "run", "family", "num_positive", "num_hard_negative",
        "num_invalid_or_post_outcome", "average_precision", "requested_recall",
        "threshold", "threshold_source", "achieved_recall",
        "false_positive_rate", "false_positive_count",
    ]
    with path.open("w", encoding="utf-8", newline="") as handle:
        writer = csv.DictWriter(handle, fieldnames=fields)
        writer.writeheader()
        for run in results:
            for metric in run["metrics"]:
                points = metric["operating_points"] or [None]
                for point in points:
                    writer.writerow(
                        {
                            "run": run["name"],
                            "family": metric["family"],
                            "num_positive": metric["num_positive"],
                            "num_hard_negative": metric["num_hard_negative"],
                            "num_invalid_or_post_outcome": metric[
                                "num_invalid_or_post_outcome"
                            ],
                            "average_precision": metric["average_precision"],
                            "requested_recall": "" if point is None else point[
                                "requested_positive_recall"
                            ],
                            "threshold": "" if point is None else point["threshold"],
                            "threshold_source": "" if point is None else point[
                                "threshold_source"
                            ],
                            "achieved_recall": "" if point is None else point[
                                "achieved_positive_recall"
                            ],
                            "false_positive_rate": "" if point is None else point[
                                "negative_false_positive_rate"
                            ],
                            "false_positive_count": "" if point is None else point[
                                "negative_false_positive_count"
                            ],
                        }
                    )


def _write_events(path: Path, rows: list[dict[str, Any]]) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    fields = [
        "run", "uid", "session", "family", "positive", "subtype",
        "candidate_source", "score", "opportunity_anchor_seconds",
        "source_row", "source_cutoff_seconds", "resolution_seconds",
        "censor_seconds", "behavior_anchor_seconds", "anchor_basis",
        "competing_intervals", "best_horizon_seconds",
        "best_wearer_probability", "used_horizons_seconds", "target_rows",
        "wearer_probabilities",
    ]
    with path.open("w", encoding="utf-8", newline="") as handle:
        writer = csv.DictWriter(handle, fieldnames=fields)
        writer.writeheader()
        for row in rows:
            serialized = dict(row)
            for field in ("used_horizons_seconds", "target_rows", "wearer_probabilities"):
                serialized[field] = json.dumps(row[field])
            writer.writerow(serialized)


def _print_examples(rows: list[dict[str, Any]], limit: int) -> None:
    if limit <= 0:
        return
    print("\nOpportunity-conditioned examples (first run)")
    for family in FAMILIES:
        print(f"{family}:")
        shown = 0
        for row in rows:
            if row["family"] != family:
                continue
            label = "POS" if row["positive"] else "HN"
            print(
                f"  {label} {row['session']} {row['uid']} "
                f"anchor={row['opportunity_anchor_seconds']:.2f}s "
                f"source_cutoff={row['source_cutoff_seconds']:.2f}s "
                f"resolution={row['resolution_seconds']:.2f}s "
                f"censor={row['censor_seconds']:.2f}s "
                f"horizons={row['used_horizons_seconds']} "
                f"score={row['score']:.4f} subtype={row['subtype']}"
            )
            shown += 1
            if shown >= limit:
                break


def main() -> None:
    args = parse_args()
    entry_max = _positive_grid_seconds(args.entry_max_seconds, field="entry max")
    yield_max = _positive_grid_seconds(args.yield_max_seconds, field="yield max")
    recalls = validate_recalls(args.matched_recalls) if args.matched_recalls else []
    runs = parse_runs(args.run)
    calibration_runs, has_calibration = _validate_calibration(args, runs, recalls)
    allow_provisional = bool(args.allow_provisional_llm_reviews)

    events, review_diagnostics = load_opportunity_events(
        args.reviews,
        args.sessions,
        allow_provisional_llm=allow_provisional,
    )
    if not events:
        raise ValueError("No opportunity-conditioned events for evaluated sessions")
    calibration_events = None
    calibration_diagnostics = None
    if has_calibration:
        calibration_events, calibration_diagnostics = load_opportunity_events(
            args.calibration_reviews,
            args.calibration_sessions,
            allow_provisional_llm=allow_provisional,
        )
        if not calibration_events:
            raise ValueError("No opportunity-conditioned calibration events")

    results: list[dict[str, Any]] = []
    all_rows: list[dict[str, Any]] = []
    first_rows: list[dict[str, Any]] = []
    for run_index, ((name, run_dir), (_, calibration_dir)) in enumerate(
        zip(runs, calibration_runs)
    ):
        arrays = load_session_predictions(run_dir, args.sessions)
        calibration_arrays = (
            load_session_predictions(calibration_dir, args.calibration_sessions)
            if calibration_events is not None
            else None
        )
        metrics: list[dict[str, Any]] = []
        run_rows: list[dict[str, Any]] = []
        for family in FAMILIES:
            fixed_thresholds = None
            threshold_source = "evaluated_events_exploratory"
            if calibration_events is not None and calibration_arrays is not None:
                calibration_metric, _ = evaluate_family(
                    calibration_events,
                    calibration_arrays,
                    family,
                    recalls,
                    entry_max_seconds=entry_max,
                    yield_max_seconds=yield_max,
                    threshold_source="training_side_calibration",
                )
                fixed_thresholds = _thresholds(calibration_metric)
                threshold_source = "training_side_calibration"
            metric, rows = evaluate_family(
                events,
                arrays,
                family,
                recalls,
                entry_max_seconds=entry_max,
                yield_max_seconds=yield_max,
                fixed_thresholds=fixed_thresholds,
                threshold_source=threshold_source,
            )
            metrics.append(metric)
            run_rows.extend({"run": name, **row} for row in rows)
            if run_index == 0:
                first_rows.extend(rows)
        results.append({"name": name, "result_dir": str(run_dir), "metrics": metrics})
        all_rows.extend(run_rows)

    provisional_count = review_diagnostics["num_provisional_llm"] + (
        0 if calibration_diagnostics is None
        else calibration_diagnostics["num_provisional_llm"]
    )
    payload = {
        "schema_version": 2,
        "protocol": "egospeak-opportunity-conditioned-floor-actions-v2",
        "semantic_scope": (
            "LLM-assisted opportunity labels with dense wearer-activity scores; "
            "this is provisional and no human review is implied."
            if provisional_count
            else "Human-reviewed opportunity labels with dense wearer-activity scores; "
            "the model output is not a semantic intent probability."
        ),
        "provisional_llm": bool(provisional_count),
        "prediction_layout": "[target_time, class, future_horizon]",
        "class_names": list(CLASS_NAMES),
        "sessions": list(args.sessions),
        "calibration_sessions": args.calibration_sessions,
        "entry_score": (
            "max wearer probability until competing resolution, "
            f"capped at {args.entry_max_seconds:g} s"
        ),
        "yield_score": (
            "max (1 - wearer probability) until other offset, "
            f"capped at {args.yield_max_seconds:g} s"
        ),
        "entry_max_seconds": entry_max,
        "yield_max_seconds": yield_max,
        "entry_headline_hard_negatives": sorted(ENTRY_CORE_HARD_NEGATIVES),
        "entry_backchannel_policy": "separate diagnostic; excluded from headline",
        "entry_competing_risk_policy": (
            "exclude selected-release positives with an other onset at or before "
            "the wearer entry, or with intervening wearer activity"
        ),
        "int_multiple_competitor_policy": (
            "exclude until one competing onset is selected by review"
        ),
        "matched_recalls": recalls,
        "review_diagnostics": review_diagnostics,
        "calibration_review_diagnostics": calibration_diagnostics,
        "runs": results,
    }
    args.output.parent.mkdir(parents=True, exist_ok=True)
    args.output.write_text(
        json.dumps(payload, indent=2, ensure_ascii=False) + "\n",
        encoding="utf-8",
    )
    metrics_path = args.csv_output or args.output.with_name(
        args.output.stem + "_metrics.csv"
    )
    events_path = args.event_csv_output or args.output.with_name(
        args.output.stem + "_events.csv"
    )
    _write_metrics(metrics_path, results)
    _write_events(events_path, all_rows)
    _print_examples(first_rows, args.print_event_examples)
    print(f"Saved {args.output}")
    print(f"Saved {metrics_path}")
    print(f"Saved {events_path}")


if __name__ == "__main__":
    main()