Merge branch 'main' of http://182.92.86.220:8300/wx44wx/icassp2027-experiment

PROJECT_CONTEXT.md

@@ -51,6 +51,11 @@ result.
   pools the 0.6--2.0-s wearer trajectory over equal-width 0.6-s causal decision
   regions. It preserves one score per annotated opportunity and never treats
   the longer annotation evidence interval as a scoring window.
+- An opportunity-conditioned dense-output evaluator aligns ENTRY positives and
+  core negatives at non-wearer releases, aligns INT positives and hold-floor
+  negatives at competing non-wearer onsets, and applies strict competing-risk
+  censoring. It uses only the original wearer probabilities, excludes wearer
+  backchannels from headline ENTRY FPR, and records unanchored/ambiguous cases.
 
 ### Measured screening result
 

docs/README.md

@@ -34,6 +34,7 @@ tests.
 | [`easycom_competing_risk_floor_evaluation.md`](easycom_competing_risk_floor_evaluation.md) | Running and auditing the competing-risk entry/yield evaluator with training-side threshold calibration |
 | [`easycom_floor_event_diagnostics.md`](easycom_floor_event_diagnostics.md) | Separating boundary observability, event-objective mismatch and addressee information with remote commands |
 | [`easycom_event_review_tool.md`](easycom_event_review_tool.md) | Reviewing automatically mined EasyCom floor-event candidates against local video and transcripts |
+| [`easycom_opportunity_event_evaluation.md`](easycom_opportunity_event_evaluation.md) | Current opportunity-conditioned ENTRY/YIELD evaluation from EgoSpeak-compatible wearer forecasts |
 | [`easycom_egospeak_event_proxy.md`](easycom_egospeak_event_proxy.md) | Scoring the EgoSpeak wearer trajectory over equal-width causal decision regions as an activity-derived ENTRY/INT proxy |
 | [`easycom_annotated_floor_evaluation.md`](easycom_annotated_floor_evaluation.md) | Reproducing the legacy asymmetric point-positive/region-negative fixed-horizon diagnostic |
 

docs/easycom_egospeak_event_proxy.md

@@ -1,5 +1,10 @@
 # EgoSpeak dense-trajectory event proxy
 
+> **Status:** boundary-centered diagnostic retained for reproducibility. The
+> manuscript's current main behavioral protocol is the matched-opportunity
+> evaluator in
+> [`easycom_opportunity_event_evaluation.md`](easycom_opportunity_event_evaluation.md).
+
 `evaluate_egospeak_event_proxy.py` provides a deliberately limited bridge
 between EgoSpeak-compatible dense future-activity predictions and reviewed
 wearer ENTRY/INT opportunities. It does **not** claim that wearer voice

docs/easycom_opportunity_event_evaluation.md

@@ -0,0 +1,59 @@
+# EasyCom opportunity-conditioned ENTRY/YIELD evaluation
+
+[`evaluate_egospeak_opportunity_events.py`](../evaluate_egospeak_opportunity_events.py)
+uses LLM-assisted or human-reviewed semantic labels to define matched
+conversational opportunities while scoring every model only from its original
+EgoSpeak-compatible wearer probabilities. It does not train a semantic event
+head.
+
+## ENTRY
+
+A positive must have a reviewer-selected preceding non-wearer release. Its
+opportunity starts at that release and resolves at the reviewed wearer entry.
+Core hard negatives start at the same type of non-wearer release and resolve
+when the same speaker resumes (`ENTRY_OTHER_MIDTURN_PAUSE`) or a different
+non-wearer takes the next floor (`ENTRY_OTHER_TO_OTHER`). The score is the
+maximum projected wearer probability up to the competing resolution, capped at
+2.0 seconds.
+
+`ENTRY_WEARER_BACKCHANNEL` is excluded from headline ENTRY FPR. Predicting
+wearer activity for a real backchannel is correct under EgoSpeak's dense target;
+ENTRY-versus-backchannel remains a separate semantic-selectivity diagnostic.
+
+## YIELD
+
+An opportunity starts at a non-wearer onset during wearer speech. The score is
+the maximum `1 - p_wearer` over 0.2--0.6 seconds, censored strictly at that
+non-wearer's offset. Reviewed `INT_YIELD` items are positive; other
+backchannels, failed takeovers, and clear wearer-hold cases are hard negatives.
+
+Current INT reviews do not select one competing onset. The loader therefore
+uses the latest supplied other interval that crosses the wearer offset within
+the annotation tolerance. It records how many intervals were eligible so
+multi-competitor cases can be audited before the protocol is frozen.
+
+## Provisional EgoSpeak run
+
+Threshold-free AP on the official test sessions can be run without fitting a
+test threshold:
+
+```bash
+python3 evaluate_egospeak_opportunity_events.py \
+  --reviews reports/easycom_entry_int_sessions1_4_v16/easycom_eot_int_provisional_llm_reviews.jsonl \
+  --run egospeak=results/egospeak_miniroada_floor_eval \
+  --sessions Session_1 Session_2 Session_3 \
+  --allow-provisional-llm-reviews \
+  --print-event-examples 10 \
+  --output results/easycom_egospeak_opportunity_events/sessions1_3_provisional_summary.json
+```
+
+The output JSON records label coverage and every exclusion reason. The metrics
+CSV contains threshold-free ENTRY/YIELD AP. The event CSV contains anchors,
+causal source cutoffs, censor times, horizons, and wearer probabilities for
+timeline auditing.
+
+Do not request matched-recall operating points on Sessions 1--3 without
+disjoint Sessions 4--12 calibration reviews and out-of-fold predictions.
+Before manuscript reporting, freeze the 2.0-second ENTRY cap, 0.6-second YIELD
+cap, positive anchor-selection policy, and multiple-competitor handling using
+training-side data only.

evaluate_egospeak_opportunity_events.py

@@ -0,0 +1,785 @@
+#!/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.  The score is the
+maximum wearer probability over available 0.2--2.0 s forecasts up to that
+competing 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.  Positive and hard-negative examples
+therefore begin at the same kind of 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 _int_competing_interval(
+    candidate: dict[str, Any],
+    *,
+    behavior_anchor_seconds: float,
+    uid: str,
+) -> tuple[float, float, int] | None:
+    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 None
+    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))
+    if not eligible:
+        return None
+    # A semantic INT review does not currently select one competing onset.
+    # The latest interval crossing the wearer offset is the deterministic,
+    # most proximal causal candidate.  Multiplicity is surfaced for audit.
+    onset, offset = max(eligible, key=lambda item: (item[0], item[1]))
+    return onset, offset, len(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
+                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:
+                interval = _int_competing_interval(
+                    candidate,
+                    behavior_anchor_seconds=behavior,
+                    uid=uid,
+                )
+                if interval is None:
+                    skipped["int_positive_without_crossing_other"] += 1
+                    continue
+                anchor, censor, competing_intervals = interval
+                resolution = behavior
+                basis = "latest_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:
+                    interval = _int_competing_interval(
+                        candidate,
+                        behavior_anchor_seconds=behavior,
+                        uid=uid,
+                    )
+                    if interval is None:
+                        skipped["int_hn_without_competing_other"] += 1
+                        continue
+                    anchor, censor, competing_intervals = interval
+                    resolution = behavior
+                    basis = "latest_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": 1,
+        "protocol": "egospeak-opportunity-conditioned-floor-actions-v1",
+        "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",
+        "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()

manuscript/main/main.tex

@@ -329,47 +329,55 @@ configuration, trainable parameter counts, and inference latency.}
 We first retain dense future-activity evaluation for direct comparison with
 EgoSpeak, reporting AP at the 0.2--2.0~s horizons and their mean mAP.
 
-For dense-only models, behavioral evaluation begins with an explicitly
+For dense-only models, behavioral evaluation is opportunity-conditioned and
-activity-derived event proxy rather than treating wearer voice activity as a
+uses the model's original wearer-activity forecasts rather than treating them
-semantic event detector. Following the prediction-region principle used by
+as semantic intent probabilities. An entry opportunity begins at a reviewed
-VAP~\cite{ekstedt2022vap}, every adjudicated item contributes one score from a
+non-wearer release. It resolves positively when the wearer enters, or
-fixed 0.6-s (three-frame) causal decision region. A positive uses the three
+negatively when the same non-wearer resumes or a different non-wearer takes the
-source frames immediately preceding its corrected behavior anchor. An
+next floor. From the first causal 5-Hz state that observes the release, the
-independently mined hard negative uses the first three source frames after its
+entry score is the maximum projected wearer probability before that competing
-opportunity begins; a raw wearer boundary relabeled as a hard negative instead
+resolution, capped at 2.0~s. Positive entries without a reviewer-selected
-uses the same pre-boundary region as a positive. Items without a complete
+release are reported as unanchored and excluded from this matched-opportunity
-three-frame region are excluded and counted. The longer interval supplied to
+analysis; they remain available to the separate raw-onset analysis.
-the annotator is evidence context only and is never a model-scoring window.
 
-At each decision frame, we follow the target-aligned forecast tensor
+An INT opportunity begins when a non-wearer starts while the wearer holds the
-diagonally to recover the wearer probabilities emitted at that same causal
+floor. It resolves positively when the reviewed behavior is an INT-yield, and
-source time. We average the 0.6--2.0-s horizons to obtain a sustained wearer-
+negatively when the non-wearer contribution is a backchannel or failed
-activity score. This mean is the entry proxy, and one minus the mean is the
+takeover and the wearer retains the floor. The yield score is the maximum of
-INT-yield proxy. Max-pooling the three decision-frame scores gives exactly one
+one minus the projected wearer probability over the next 0.2--0.6~s, censored
-score per positive event or hard-negative opportunity. This construction can
+strictly at the competing non-wearer offset. All prediction values are read
-capture sustained activity versus a brief vocalization, but it cannot by
+diagonally from one causal source row of the target-aligned forecast tensor.
-itself establish semantic floor intent; in particular, wearer backchannels and
+No input after the observed action outcome is used. For current INT-positive
-within-turn pauses remain deliberately difficult negatives. A later linear
+annotations, which do not select a competing onset explicitly, the
-event probe, if used, is fit only on training-side annotations and compared at
+implementation uses the latest supplied non-wearer interval crossing the
-matched capacity across representations, following the separated downstream-
+wearer offset and surfaces cases with multiple eligible intervals for audit.
-probe logic of MuVAP~\cite{qi2026muvap}.
+
+The headline entry hard negatives are same-speaker mid-turn pauses and other-
+to-other transfers. Wearer backchannels are not pooled into headline false-
+entry rate: their high wearer probability is correct under the dense training
+target. We instead retain entry-versus-backchannel as a separate semantic-
+selectivity diagnostic. Thus, the semantic annotations define which matched
+opportunities are evaluated, while every model is scored solely through the
+same EgoSpeak-compatible dense output. A capacity-matched downstream probe,
+following MuVAP~\cite{qi2026muvap}, is secondary and is not needed for this
+zero-shot activity-derived comparison.
 
 \textbf{False-positive scoring rule.}
-A false positive is a hard-negative opportunity whose fixed-region proxy score
+A false positive is a hard-negative opportunity whose opportunity score
 exceeds the frozen threshold, regardless of whether the wearer subsequently
-exhibits an onset or offset. Max-pooling means that multiple high-scoring
+exhibits an unrelated later onset or offset. Each opportunity contributes one
-decision frames still count as one false-positive opportunity. Thus, false-positive
+score, and competing-risk censoring prevents later conversational outcomes
-evaluation includes conversational situations in which the correct wearer
+from changing its label. Thus, false-positive evaluation includes situations
-behavior is to remain silent or retain the floor.
+in which the correct wearer behavior is to remain silent or retain the floor.
 
 We report threshold-free proxy AP for entry and INT. The headline behavioral
 operating point additionally reports false-positive rate at matched recall
 $R=0.8$; thresholds are selected independently for entry and INT from training-
 side out-of-fold predictions. The other-to-other hard-negative subset tests
 whether local visual evidence specifically reduces erroneous wearer action
-scores during multiparty floor transfers. \todo{Freeze the 0.6-s decision
+scores during multiparty floor transfers. \todo{Freeze the 2.0-s entry and
-window and 0.6--2.0-s aggregation range using training-side data, run window
+0.6-s yield caps using training-side data, audit selected-release and competing-
-sensitivity analysis, and verify identical event denominators across methods.}
+onset coverage, and verify identical censored/valid denominators across methods.}
 
 A deployment-style committed-event analysis with one-to-one matching and
 signed latency is secondary and is omitted unless a threshold, persistence,
@@ -394,8 +402,8 @@ verify identical targets, causal validity, and evaluation.}
 
 \begin{table}[t]
 \caption{Confirmatory EasyCom results. Entry and INT false-positive rates use
-the fixed-region activity-derived proxy at matched recall $R=0.8$; E/I AP is
+the opportunity-conditioned activity score at matched recall $R=0.8$; E/I AP
-the corresponding event-balanced proxy AP.}
+is the corresponding event-balanced opportunity AP.}
 \label{tab:main}
 \centering
 \small
@@ -427,7 +435,7 @@ evidence primarily improves \todo{the interpretation supported by the data}.
 \textbf{Annotated behavior.}
 At matched wearer-entry recall $R=0.8$, the proposed adapter reduces entry FPR
 from \todo{$X$} to \todo{$Y$}; at matched INT-yield recall, it changes INT FPR
-from \todo{$A$} to \todo{$B$}. Its entry/INT activity-proxy APs are
+from \todo{$A$} to \todo{$B$}. Its entry/INT opportunity APs are
 \todo{$E/I$}. These recall-matched opportunity error rates test whether lower
 scores on confusing floor situations are obtained without suppressing
 legitimate entry or yield events.

test_evaluate_egospeak_opportunity_events.py

@@ -0,0 +1,396 @@
+from __future__ import annotations
+
+import argparse
+import json
+import tempfile
+import unittest
+from pathlib import Path
+from unittest import mock
+
+import numpy as np
+
+import evaluate_egospeak_opportunity_events as evaluator
+from evaluate_egospeak_opportunity_events import (
+    OpportunityEvent,
+    _first_causal_source_row,
+    _score_event,
+    evaluate_family,
+    load_opportunity_events,
+)
+from evaluate_floor_state_changes import NUM_CLASSES, NUM_HORIZONS
+
+
+def reviewed_row(
+    uid: str,
+    *,
+    family: str,
+    verdict: str,
+    subtype: str | None,
+    anchor: float,
+    region_start: float,
+    region_end: float,
+    selected_release: float | None = None,
+    nearby: list[dict[str, object]] | None = None,
+    candidate_source: str | None = None,
+) -> dict[str, object]:
+    event_type = None
+    if verdict == "POSITIVE":
+        event_type = "WEARER_ENTRY" if family == "entry" else "INT_YIELD"
+    return {
+        "candidate_id": uid,
+        "session": "Session_4",
+        "candidate_family": family,
+        "review": {
+            "human_status": "REVIEWED",
+            "human_verdict": verdict,
+            "human_event_type": event_type,
+            "human_subtype": subtype,
+            "selected_opportunity_seconds": selected_release,
+            "boundary_correction_seconds": 0,
+        },
+        "candidate": {
+            "candidate_source": candidate_source or (
+                "WEARER_ONSET" if family == "entry" else "WEARER_OFFSET"
+            ),
+            "anchor_dense_seconds": anchor,
+            "region_start_seconds": region_start,
+            "region_end_seconds": region_end,
+            "temporal_evidence": {
+                "behavior_anchor_seconds": anchor,
+                "wearer_offset_seconds": anchor,
+                "nearby_other_intervals": nearby or [],
+            },
+        },
+    }
+
+
+def prediction_arrays(length: int = 80) -> tuple[np.ndarray, np.ndarray]:
+    return (
+        np.zeros((length, NUM_CLASSES, NUM_HORIZONS), dtype=np.float32),
+        np.ones((length, NUM_HORIZONS), dtype=bool),
+    )
+
+
+class OpportunityEventTests(unittest.TestCase):
+    def write_reviews(self, rows: list[dict[str, object]]) -> tuple[tempfile.TemporaryDirectory[str], Path]:
+        temporary = tempfile.TemporaryDirectory()
+        path = Path(temporary.name) / "reviews.jsonl"
+        path.write_text(
+            "".join(json.dumps(row) + "\n" for row in rows),
+            encoding="utf-8",
+        )
+        return temporary, path
+
+    def test_first_source_row_observes_non_grid_opportunity(self) -> None:
+        source_row, cutoff = _first_causal_source_row(1.05)
+        self.assertEqual(source_row, 5)
+        self.assertAlmostEqual(cutoff, 1.2)
+        self.assertEqual(_first_causal_source_row(1.0), (4, 1.0))
+
+    def test_loader_uses_selected_release_and_excludes_entry_backchannel(self) -> None:
+        rows = [
+            reviewed_row(
+                "entry-positive",
+                family="entry",
+                verdict="POSITIVE",
+                subtype=None,
+                anchor=1.6,
+                region_start=0.0,
+                region_end=3.0,
+                selected_release=1.0,
+            ),
+            reviewed_row(
+                "entry-backchannel",
+                family="entry",
+                verdict="HARD_NEGATIVE",
+                subtype="ENTRY_WEARER_BACKCHANNEL",
+                anchor=3.0,
+                region_start=2.0,
+                region_end=4.0,
+            ),
+        ]
+        temporary, path = self.write_reviews(rows)
+        self.addCleanup(temporary.cleanup)
+        events, diagnostics = load_opportunity_events([path], ["Session_4"])
+        self.assertEqual(len(events), 1)
+        self.assertEqual(events[0].opportunity_anchor_seconds, 1.0)
+        self.assertEqual(events[0].resolution_seconds, 1.6)
+        self.assertEqual(
+            diagnostics["skipped"]["entry_backchannel_separate_diagnostic"], 1
+        )
+
+    def test_loader_selects_latest_crossing_other_for_int(self) -> None:
+        nearby = [
+            {
+                "onset_seconds": 1.0,
+                "offset_seconds": 3.5,
+                "crosses_offset_with_tolerance": True,
+            },
+            {
+                "onset_seconds": 2.0,
+                "offset_seconds": 3.8,
+                "crosses_offset_with_tolerance": True,
+            },
+        ]
+        temporary, path = self.write_reviews([
+            reviewed_row(
+                "yield-positive",
+                family="int",
+                verdict="POSITIVE",
+                subtype=None,
+                anchor=3.0,
+                region_start=0.0,
+                region_end=6.0,
+                nearby=nearby,
+            )
+        ])
+        self.addCleanup(temporary.cleanup)
+        events, _ = load_opportunity_events([path], ["Session_4"])
+        self.assertEqual(len(events), 1)
+        self.assertEqual(events[0].opportunity_anchor_seconds, 2.0)
+        self.assertEqual(events[0].censor_seconds, 3.8)
+        self.assertEqual(events[0].competing_intervals, 2)
+
+    def test_entry_hn_requires_matching_independently_mined_source(self) -> None:
+        rows = [
+            reviewed_row(
+                "valid",
+                family="entry",
+                verdict="HARD_NEGATIVE",
+                subtype="ENTRY_OTHER_TO_OTHER",
+                anchor=2.0,
+                region_start=2.0,
+                region_end=2.6,
+                candidate_source="OTHER_TO_OTHER",
+            ),
+            reviewed_row(
+                "invalid",
+                family="entry",
+                verdict="HARD_NEGATIVE",
+                subtype="ENTRY_OTHER_TO_OTHER",
+                anchor=4.0,
+                region_start=1.0,
+                region_end=7.0,
+                candidate_source="WEARER_ONSET",
+            ),
+        ]
+        temporary, path = self.write_reviews(rows)
+        self.addCleanup(temporary.cleanup)
+        events, diagnostics = load_opportunity_events([path], ["Session_4"])
+        self.assertEqual([event.uid for event in events], ["valid"])
+        self.assertEqual(
+            diagnostics["skipped"]["entry_hn_without_matching_mined_release"],
+            1,
+        )
+
+    def test_zero_duration_mined_opportunity_is_skipped_not_fatal(self) -> None:
+        temporary, path = self.write_reviews([
+            reviewed_row(
+                "zero-gap",
+                family="entry",
+                verdict="HARD_NEGATIVE",
+                subtype="ENTRY_OTHER_TO_OTHER",
+                anchor=2.0,
+                region_start=2.0,
+                region_end=2.0,
+                candidate_source="OTHER_TO_OTHER",
+            )
+        ])
+        self.addCleanup(temporary.cleanup)
+        events, diagnostics = load_opportunity_events([path], ["Session_4"])
+        self.assertEqual(events, [])
+        self.assertEqual(
+            diagnostics["skipped"]["entry_nonpositive_opportunity_duration"],
+            1,
+        )
+
+    def test_entry_score_uses_max_wearer_probability_before_resolution(self) -> None:
+        event = OpportunityEvent(
+            "entry", "Session_4", "entry", True, "WEARER_ENTRY", "WEARER_ONSET",
+            1.0, 1.6, 1.6, 1.6, "reviewed_release",
+        )
+        prediction, valid = prediction_arrays()
+        # source row 4, cutoff 1.0; h=.2/.4/.6 are admitted.  Since the onset
+        # is exactly on-grid, the h=.8 post-onset bin must be excluded.
+        prediction[5, 1, 0] = 0.1
+        prediction[6, 1, 1] = 0.2
+        prediction[7, 1, 2] = 0.8
+        prediction[8, 1, 3] = 0.3
+        scored = _score_event(
+            event,
+            {"Session_4": (prediction, valid)},
+            entry_max_seconds=2.0,
+            yield_max_seconds=0.6,
+        )
+        self.assertIsNotNone(scored)
+        self.assertAlmostEqual(scored[0], 0.8)  # type: ignore[index]
+        self.assertAlmostEqual(scored[1]["best_horizon_seconds"], 0.6)  # type: ignore[index]
+        self.assertEqual(len(scored[1]["used_horizons_seconds"]), 3)  # type: ignore[index]
+
+    def test_entry_positive_admits_bin_containing_off_grid_onset(self) -> None:
+        event = OpportunityEvent(
+            "entry", "Session_4", "entry", True, "WEARER_ENTRY", "WEARER_ONSET",
+            1.0, 1.65, 1.65, 1.65, "reviewed_release",
+        )
+        prediction, valid = prediction_arrays()
+        prediction[8, 1, 3] = 0.9  # target cutoff 1.8 contains onset 1.65.
+        scored = _score_event(
+            event,
+            {"Session_4": (prediction, valid)},
+            entry_max_seconds=2.0,
+            yield_max_seconds=0.6,
+        )
+        self.assertIsNotNone(scored)
+        self.assertAlmostEqual(scored[0], 0.9)  # type: ignore[index]
+        self.assertEqual(len(scored[1]["used_horizons_seconds"]), 4)  # type: ignore[index]
+
+    def test_entry_hard_negative_stops_before_competing_onset(self) -> None:
+        event = OpportunityEvent(
+            "negative", "Session_4", "entry", False, "ENTRY_OTHER_TO_OTHER",
+            "OTHER_TO_OTHER", 1.0, 1.6, 1.6, None, "mined_release",
+        )
+        prediction, valid = prediction_arrays()
+        prediction[5, 1, 0] = 0.1
+        prediction[6, 1, 1] = 0.2
+        prediction[7, 1, 2] = 0.3
+        prediction[8, 1, 3] = 0.95  # 1.8 s: after competing onset, must be ignored.
+        scored = _score_event(
+            event,
+            {"Session_4": (prediction, valid)},
+            entry_max_seconds=2.0,
+            yield_max_seconds=0.6,
+        )
+        self.assertIsNotNone(scored)
+        self.assertAlmostEqual(scored[0], 0.3)  # type: ignore[index]
+        self.assertEqual(scored[1]["used_horizons_seconds"], [0.2, 0.4, 0.6000000000000001])  # type: ignore[index]
+
+    def test_yield_score_is_near_term_inactivity_and_censored(self) -> None:
+        event = OpportunityEvent(
+            "yield", "Session_4", "int", True, "INT_YIELD", "WEARER_OFFSET",
+            2.0, 2.4, 2.5, 2.4, "crossing_other",
+        )
+        prediction, valid = prediction_arrays()
+        prediction[10, 1, 0] = 0.9
+        prediction[11, 1, 1] = 0.2
+        prediction[12, 1, 2] = 0.0  # target 2.6 crosses other offset 2.5.
+        scored = _score_event(
+            event,
+            {"Session_4": (prediction, valid)},
+            entry_max_seconds=2.0,
+            yield_max_seconds=0.6,
+        )
+        self.assertIsNotNone(scored)
+        self.assertAlmostEqual(scored[0], 0.8)  # type: ignore[index]
+        self.assertEqual(scored[1]["used_horizons_seconds"], [0.2, 0.4])  # type: ignore[index]
+
+    def test_positive_is_dropped_if_first_causal_state_observes_outcome(self) -> None:
+        event = OpportunityEvent(
+            "fast", "Session_4", "entry", True, "WEARER_ENTRY", "WEARER_ONSET",
+            1.05, 1.1, 1.1, 1.1, "reviewed_release",
+        )
+        prediction, valid = prediction_arrays()
+        self.assertIsNone(
+            _score_event(
+                event,
+                {"Session_4": (prediction, valid)},
+                entry_max_seconds=2.0,
+                yield_max_seconds=0.6,
+            )
+        )
+
+    def test_one_score_per_opportunity_and_ap(self) -> None:
+        events = [
+            OpportunityEvent(
+                "p", "Session_4", "entry", True, "WEARER_ENTRY", "WEARER_ONSET",
+                1.0, 1.6, 1.6, 1.6, "reviewed_release",
+            ),
+            OpportunityEvent(
+                "n", "Session_4", "entry", False, "ENTRY_OTHER_TO_OTHER",
+                "OTHER_TO_OTHER", 2.0, 2.6, 2.6, None, "mined_release",
+            ),
+        ]
+        prediction, valid = prediction_arrays()
+        prediction[7, 1, 2] = 0.9
+        metric, rows = evaluate_family(
+            events,
+            {"Session_4": (prediction, valid)},
+            "entry",
+            [],
+            entry_max_seconds=2.0,
+            yield_max_seconds=0.6,
+        )
+        self.assertEqual(len(rows), 2)
+        self.assertEqual(metric["num_positive"], 1)
+        self.assertEqual(metric["num_hard_negative"], 1)
+        self.assertAlmostEqual(metric["average_precision"], 1.0)
+
+    def test_main_writes_summary_metrics_and_event_audit(self) -> None:
+        nearby = [{
+            "onset_seconds": 5.0,
+            "offset_seconds": 6.0,
+            "crosses_offset_with_tolerance": True,
+        }]
+        rows = [
+            reviewed_row(
+                "ep", family="entry", verdict="POSITIVE", subtype=None,
+                anchor=1.6, region_start=0.0, region_end=3.0,
+                selected_release=1.0,
+            ),
+            reviewed_row(
+                "en", family="entry", verdict="HARD_NEGATIVE",
+                subtype="ENTRY_OTHER_TO_OTHER", anchor=3.0,
+                region_start=3.0, region_end=3.8,
+                candidate_source="OTHER_TO_OTHER",
+            ),
+            reviewed_row(
+                "ip", family="int", verdict="POSITIVE", subtype=None,
+                anchor=5.4, region_start=3.0, region_end=7.0, nearby=nearby,
+            ),
+            reviewed_row(
+                "in", family="int", verdict="HARD_NEGATIVE",
+                subtype="INT_OTHER_BACKCHANNEL", anchor=7.0,
+                region_start=7.0, region_end=7.8,
+                candidate_source="OTHER_BACKCHANNEL",
+            ),
+        ]
+        temporary, reviews = self.write_reviews(rows)
+        self.addCleanup(temporary.cleanup)
+        root = Path(temporary.name)
+        run_dir = root / "run"
+        (run_dir / "pred").mkdir(parents=True)
+        (run_dir / "valid").mkdir()
+        prediction, valid = prediction_arrays()
+        np.save(run_dir / "pred" / "Session_4.npy", prediction)
+        np.save(run_dir / "valid" / "Session_4.npy", valid)
+        output = root / "summary.json"
+        args = argparse.Namespace(
+            reviews=[reviews],
+            run=[f"egospeak={run_dir}"],
+            sessions=["Session_4"],
+            entry_max_seconds=2.0,
+            yield_max_seconds=0.6,
+            calibration_reviews=None,
+            calibration_run=None,
+            calibration_sessions=None,
+            matched_recalls=[],
+            allow_in_sample_thresholds=False,
+            output=output,
+            csv_output=None,
+            event_csv_output=None,
+            print_event_examples=0,
+            allow_provisional_llm_reviews=False,
+        )
+        with mock.patch.object(evaluator, "parse_args", return_value=args):
+            evaluator.main()
+        payload = json.loads(output.read_text(encoding="utf-8"))
+        self.assertEqual(
+            payload["protocol"],
+            "egospeak-opportunity-conditioned-floor-actions-v1",
+        )
+        self.assertTrue((root / "summary_metrics.csv").is_file())
+        self.assertTrue((root / "summary_events.csv").is_file())
+        self.assertEqual(len(payload["runs"][0]["metrics"]), 2)
+
+
+if __name__ == "__main__":
+    unittest.main()