icassp2027-experiment/adjudicate_easycom_eot_int_candidates.py

#!/usr/bin/env python3
"""Add conservative DeepSeek pre-labels to wearer entry/INT candidates.

The model receives the candidate source, structured participant timing and the
transcript excerpt.  It cannot change any source timestamp.  Its output is
stored as an auditable suggestion beside the untouched rule candidate and must
still be accepted, relabeled, or excluded by a human.
"""

from __future__ import annotations

import argparse
from collections import Counter
import hashlib
import json
import os
from pathlib import Path
import random
import re
import sys
import time
from typing import Any, Sequence, TextIO
import urllib.error
import urllib.request

from easycom_event_taxonomy import EXCLUDE_REASONS, exclude_reason_group
from prepare_easycom_eot_int_review import requires_semantic_adjudication


PROMPT_VERSION = "easycom-wearer-boundary-semantic-prelabel-v15"
DEFAULT_MODEL = "deepseek-v4-flash"
DEFAULT_BASE_URL = "https://api.deepseek.com"
VERDICTS = {"POSITIVE", "HARD_NEGATIVE", "EXCLUDE"}
EVENT_TYPES = {"WEARER_ENTRY", "INT_YIELD"}
HARD_NEGATIVE_TYPES = {
    "ENTRY_OTHER_MIDTURN_PAUSE",
    "ENTRY_OTHER_TO_OTHER",
    "ENTRY_WEARER_BACKCHANNEL",
    "INT_OTHER_BACKCHANNEL",
    "INT_NON_TAKEOVER_OVERLAP",
    "INT_WEARER_HOLDS_FLOOR",
    "OTHER_HARD_NEGATIVE",
}


class BatchValidationError(RuntimeError):
    """The provider responded, but the batch output violated the label schema."""


class APIRequestError(RuntimeError):
    """The provider could not return a usable response after transport retries."""


def _progress_line(
    completed: int, total: int, *, label: str, elapsed_seconds: float, width: int = 26
) -> str:
    fraction = 1.0 if total == 0 else min(1.0, max(0.0, completed / total))
    filled = round(width * fraction)
    bar = "█" * filled + "░" * (width - filled)
    minutes, seconds = divmod(max(0, round(elapsed_seconds)), 60)
    return (
        f"DeepSeek [{bar}] {completed}/{total} ({fraction * 100:5.1f}%) "
        f"{label} · {minutes:02d}:{seconds:02d}"
    )


class _ProgressBar:
    """Dependency-free terminal progress with readable redirected logs."""

    def __init__(self, total: int, *, stream: TextIO | None = None) -> None:
        self.total = total
        self.stream = stream or sys.stderr
        self.started = time.monotonic()
        self.is_tty = bool(getattr(self.stream, "isatty", lambda: False)())
        self.last_logged = -1

    def update(self, completed: int, label: str, *, final: bool = False) -> None:
        line = _progress_line(
            completed,
            self.total,
            label=label,
            elapsed_seconds=time.monotonic() - self.started,
        )
        if self.is_tty:
            self.stream.write("\r" + line)
            if final:
                self.stream.write("\n")
            self.stream.flush()
            return
        # Avoid producing one permanent log line per API batch in redirected jobs.
        step = max(1, self.total // 10)
        if final or self.last_logged < 0 or completed - self.last_logged >= step:
            self.stream.write(line + "\n")
            self.stream.flush()
            self.last_logged = completed


SYSTEM_PROMPT = """You conservatively pre-label candidate regions for a scientific study of wearer-relative conversational floor control.

Evidence rules:
1. Reconstruct the event chronologically before assigning semantics. Use exact start/end timestamps to identify (a) the proposed release or competing onset, (b) the wearer behavior anchor, (c) which same participant continues across the anchor, and (d) who produces the next substantive turn. Then use transcript content and Target_of_Speech, when supplied, to decide conversational function. Timing has priority over speculative transcript interpretation, but duration alone never determines floor function.
2. Never alter or propose an event timestamp. Do not infer prosody, gaze, intention, or acoustic cues that were not supplied.
3. Every supplied WEARER_ONSET and WEARER_OFFSET is a raw observed wearer activity boundary. Candidate generation has made no floor-entry, within-turn-resumption, interruption, or takeover decision. A raw wearer onset is not automatically a floor entry; a wearer offset is not automatically a yield; an other onset is not automatically a takeover. A pause is not automatically a release, overlap is not automatically interruption, and words such as "yeah" are not automatically backchannels. Determine what the utterance does in context, not only how long it lasts.
4. EXCLUDE is the sole non-scoring outcome. Use it both for insufficient/unresolvable evidence and for clear out-of-scope non-events that should not enter evaluation, including a normal wearer completion that merely coincides with laughter or feedback. Never force such a region into HARD_NEGATIVE for coverage.
5. For every WEARER_ONSET, decide whether the wearer genuinely enters or re-enters the conversational floor. First inspect the previous wearer offset and silence gap, gap_other_intervals, nearby other activity, transcript, and Target_of_Speech. A short within-turn pause followed by continuation of the same wearer contribution is not a new entry; use EXCLUDE unless the region is a valid named false-positive opportunity. A POSITIVE WEARER_ENTRY does not require a preceding non-wearer release: it may occur after a release, from silence, or while another speaker is active. temporal_evidence.opportunity_options contains nearby preceding other offsets with factual relation flags, not pre-approved causes. If one supplied release clearly prompted the entry, selected_opportunity_uid may contain its uid; otherwise return null. Never invent a trigger or timestamp.

   Apply this adjacency-pair rule before using brevity or later speaker activity as evidence: when a participant addresses the wearer with a question, request, proposal, or other first-pair part and the wearer supplies its contextually complete second-pair part, classify the wearer contribution as POSITIVE / WEARER_ENTRY. A polar-question answer such as "yes", "no", or "Nope." is complete even if it is one word and lasts less than one second. Reciprocal Target_of_Speech evidence (question targets wearer; answer targets questioner) strengthens this reading. Floor entry here means making the committed conversational response; it does not require monopolizing or retaining the floor afterward. Therefore, the questioner or another participant speaking after the answer does not turn that completed answer into a backchannel.

   Use HARD_NEGATIVE / ENTRY_WEARER_BACKCHANNEL only when the wearer vocalization functions merely as feedback on another participant's still-ongoing contribution and does not answer a question, accept or reject a proposal, provide requested information, or otherwise complete an adjacency pair. Lexical form alone is insufficient: "yeah" may answer a yes/no question or may be a continuer, depending on the preceding utterance and addressee evidence. For the supplied audit pattern, participant 3 asks the wearer "Have you ever analyzed documents?" and releases at 54.00 s; the wearer answers "Nope." at 54.15 s. This is POSITIVE / WEARER_ENTRY, not ENTRY_WEARER_BACKCHANNEL, even though it is brief and participant 3 later speaks again.
6. For every WEARER_OFFSET, answer the central question: did the wearer stop because a non-wearer genuinely took over the conversational floor? Reconstruct the wearer activity-run start and offset, every nearby other onset and offset, which others are active immediately before and after the offset, whether each other is pre-existing or starts during wearer speech, whether it crosses the offset, overlap duration, who speaks next, transcript content, and Target_of_Speech. All temporal booleans are descriptive evidence, not prerequisite labels. A valid INT_YIELD may involve a newly entering or pre-existing competitor if the chronology and semantics establish genuine takeover and wearer relinquishment. Conversely, crossing the offset or satisfying t_w < t_o < t_y never proves takeover by itself. Verify that the relevant non-wearer actually takes and sustains the next floor. A different participant's later substantive turn is not evidence that the proposed competitor took over. Do not label INT_WEARER_HOLDS_FLOOR merely because the wearer speaks again later: if a competitor establishes an intervening substantive floor and the wearer has an intervening silent interval, the later wearer onset is a new WEARER_ENTRY and does not invalidate the earlier INT_YIELD. Natural completion, completion-to-silence, a within-turn wearer pause, or a post-completion other onset that could not have caused the stop is normally EXCLUDE unless the region is a valid named hard-negative opportunity. If the wearer reaches a semantically complete natural ending while overlapping activity is only [L] or feedback and that contributor does not take the next substantive floor, return EXCLUDE.
7. Transcript event markers are evidence, not lexical turns. In EasyCom, a transcription consisting of [L] denotes laughter-only vocalization. Do not treat [L] as a sustained floor entry or interruption takeover merely because its activity interval creates an onset, offset, or overlap candidate. Specifically:
   - wearer [L] during a non-wearer's retained turn is normally HARD_NEGATIVE / ENTRY_WEARER_BACKCHANNEL, not WEARER_ENTRY;
   - non-wearer [L] overlapping a wearer's retained turn is normally HARD_NEGATIVE / INT_OTHER_BACKCHANNEL, not INT_YIELD;
   - wearer [L] ending while a non-wearer continues does not establish that the wearer held and yielded the floor.
   Use EXCLUDE if the surrounding timing does not establish whose floor was retained. Do not EXCLUDE a valid laughter-only annotation merely because it is non-lexical, and do not infer that a visible smile occurred: [L] supplies laughter evidence only. Do not assign semantics to other bracketed markers unless the supplied context establishes them.

Top-level verdicts:
- POSITIVE: a target WEARER_ENTRY or INT_YIELD occurs and the system should commit.
- HARD_NEGATIVE: a deliberately retained, valid false-positive opportunity matching one of the behavioral counterexamples below; it enters evaluation as a negative.
- EXCLUDE: a non-scoring region. This includes clear out-of-scope/non-target opportunities such as natural wearer completion coincident with feedback, as well as insufficient evidence, annotation corruption, a recording boundary, missing context, a non-speech artifact, timestamp corruption, or another unresolvable defect.

EXCLUDE requires exactly one exclude_reason from this closed taxonomy:
- OUT_OF_SCOPE_NORMAL_COMPLETION: the wearer naturally completes; no interruption-yield target occurs.
- OUT_OF_SCOPE_WITHIN_TURN_CONTINUATION: a raw boundary is only a pause/resumption within the same wearer contribution.
- OUT_OF_SCOPE_POST_COMPLETION_ACTIVITY: another contribution begins only after wearer completion and cannot have caused the boundary.
- OUT_OF_SCOPE_NON_EVENT: another clear, interpretable non-target boundary not covered above.
- INSUFFICIENT_EVIDENCE: the supplied valid evidence cannot resolve conversational function.
- DATA_ANNOTATION_ERROR: participant, speech, transcript, or target annotation is evidently erroneous or contradictory.
- DATA_RECORDING_BOUNDARY: the recording/clip boundary prevents reliable interpretation.
- DATA_MISSING_CONTEXT: required context is absent from the supplied annotation window or source.
- DATA_NON_SPEECH_ARTIFACT: the boundary is caused by an artifact rather than an annotated conversational vocalization.
- DATA_TIMESTAMP_CORRUPTION: timestamps are corrupted or irreconcilably misaligned.
The OUT_OF_SCOPE_* reasons describe valid data outside the evaluation target. DATA_* reasons describe bad or unusable data. Never substitute one family for the other.

POSITIVE event_type must be exactly WEARER_ENTRY or INT_YIELD. WEARER_ENTRY means the wearer genuinely enters or re-enters the floor with a contextually complete response or substantive contribution; no prior release is required and the contribution need not be long. INT_YIELD means a non-wearer genuinely takes over during wearer speech and the wearer relinquishes the floor. A later WEARER_ENTRY after an intervening other floor is a new event and must not retroactively invalidate the preceding INT_YIELD.

HARD_NEGATIVE subtypes are behavioral counterexamples, not data-quality labels:
- ENTRY_OTHER_MIDTURN_PAUSE: a non-wearer merely pauses inside that same speaker's continuing floor episode; the same non-wearer resumes or clearly retains the turn, while the wearer does not enter.
- ENTRY_OTHER_TO_OTHER: one non-wearer releases and a different non-wearer takes the next floor; the wearer does not enter.
- ENTRY_WEARER_BACKCHANNEL: the wearer vocalizes, but only as brief feedback, acknowledgement, laughter, or another non-floor-taking response; the non-wearer retains or resumes the floor.
- INT_OTHER_BACKCHANNEL: the competing non-wearer contribution during wearer speech is brief feedback, acknowledgement, laughter, or another backchannel, and the wearer clearly continues or resumes the same floor episode. If the wearer instead naturally completes at the overlap boundary and the feedback speaker does not take the next substantive floor, use EXCLUDE.
- INT_WEARER_HOLDS_FLOOR: the competing contribution is not best characterized as a backchannel, but the wearer demonstrably continues or promptly resumes the same floor episode after the overlap, so no yield occurs.
- INT_NON_TAKEOVER_OVERLAP: the competing contribution is not merely a backchannel, yet it still fails to establish a sustained takeover (for example, a collision, abandoned attempt, or overlap that dies out); use this only when wearer continuation is not the clearer defining evidence for INT_WEARER_HOLDS_FLOOR.
- OTHER_HARD_NEGATIVE: a valid, interpretable false-positive opportunity in the correct ENTRY or INT family that does not fit any named subtype. State the unmatched behavior explicitly. This is not an uncertainty fallback; use EXCLUDE when the evidence is insufficient or unresolvable.

Subtype precedence when descriptions overlap:
1. Use the family matching the candidate task: ENTRY candidates take only ENTRY_* subtypes and INT candidates take only INT_* subtypes.
2. For ENTRY, prefer ENTRY_WEARER_BACKCHANNEL when an actual wearer vocalization is feedback rather than a floor entry; otherwise distinguish same-other resumption (ENTRY_OTHER_MIDTURN_PAUSE) from a different-other takeover (ENTRY_OTHER_TO_OTHER).
3. For INT, use INT_OTHER_BACKCHANNEL when the other contribution is feedback and wearer continuation/resumption makes it a scored false-positive opportunity. If feedback merely coincides with a natural wearer completion, use EXCLUDE. Otherwise use INT_WEARER_HOLDS_FLOOR when wearer continuation/resumption clearly demonstrates retained floor; use INT_NON_TAKEOVER_OVERLAP for the remaining interpretable failed-takeover overlaps.
4. Use OTHER_HARD_NEGATIVE only after the named subtypes have been ruled out. Never use it merely because confidence is low.

Return exactly one JSON object with this shape and no markdown:
{"items":[{"uid":"...","verdict":"EXCLUDE","event_type":null,"subtype":null,"exclude_reason":"INSUFFICIENT_EVIDENCE","selected_opportunity_uid":null,"confidence":0.25,"reason":"brief evidence-based sentence"}]}
Field combinations are strict:
- POSITIVE / WEARER_ENTRY: subtype=null; exclude_reason=null; selected_opportunity_uid is one supplied opportunity uid or null.
- POSITIVE / INT_YIELD: subtype=null; exclude_reason=null; selected_opportunity_uid=null.
- HARD_NEGATIVE: event_type=null; exclude_reason=null; selected_opportunity_uid=null; subtype is one valid subtype from the candidate's ENTRY or INT family.
- EXCLUDE: event_type=null; subtype=null; selected_opportunity_uid=null; exclude_reason is exactly one taxonomy value above.
Return every supplied uid exactly once. Copy uid byte-for-byte and never add a backslash before an underscore. confidence must be a number from 0 to 1."""


def _normalize_returned_uid(value: Any) -> Any:
    """Undo DeepSeek's occasional Markdown escaping without weakening identity checks."""
    if not isinstance(value, str):
        return value
    return re.sub(r"\\+_", "_", value)


def _participant_uid_suffix(row: dict[str, Any], ordinal: int) -> str:
    previous = row.get("previous_other_participant_ids") or []
    incoming = row.get("incoming_other_participant_ids") or []
    previous_key = "-".join(f"p{x}" for x in previous) or "none"
    incoming_key = "-".join(f"p{x}" for x in incoming) or "none"
    return f"prev-{previous_key}__in-{incoming_key}__row-{ordinal}"


def read_candidates(path: Path) -> list[dict[str, Any]]:
    rows: list[dict[str, Any]] = []
    with path.open("r", encoding="utf-8") as handle:
        for line_number, line in enumerate(handle, 1):
            if not line.strip():
                continue
            row = json.loads(line)
            if not isinstance(row, dict) or not isinstance(row.get("uid"), str):
                raise ValueError(f"Expected candidate object with uid at {path}:{line_number}")
            if not (row.get("candidate_source") or row.get("task")):
                raise ValueError(f"Missing candidate source/task at {path}:{line_number}")
            rows.append(row)
    if not rows:
        raise ValueError(f"No candidates in {path}")
    by_uid: dict[str, list[int]] = {}
    for index, row in enumerate(rows):
        by_uid.setdefault(row["uid"], []).append(index)
    for original_uid, indices in by_uid.items():
        if len(indices) == 1:
            continue
        for ordinal, index in enumerate(indices, 1):
            row = dict(rows[index])
            row["uid_disambiguated_from"] = original_uid
            row["uid"] = f"{original_uid}/{_participant_uid_suffix(row, ordinal)}"
            row["event_id"] = row["uid"]
            rows[index] = row
    if len({row["uid"] for row in rows}) != len(rows):
        raise ValueError(f"Could not deterministically disambiguate duplicate candidate UIDs in {path}")
    return rows


def candidate_for_prompt(row: dict[str, Any]) -> dict[str, Any]:
    fields = (
        "uid",
        "candidate_source",
        "rule_candidate_type",
        "task",
        "anchor_type",
        "anchor_dense_seconds",
        "event_time",
        "region_start_seconds",
        "region_end_seconds",
        "temporal_evidence",
        "speaker_timeline",
        "context",
    )
    payload = {field: row.get(field) for field in fields if row.get(field) is not None}
    # Older v1 files remain reviewable and can still be pre-labeled.
    if "candidate_source" not in payload:
        payload["candidate_source"] = row.get("anchor_type", row.get("event_type", "UNKNOWN"))
    if "event_time" not in payload:
        payload["event_time"] = row.get("anchor_dense_seconds")
    return payload


def build_user_prompt(rows: Sequence[dict[str, Any]], pass_index: int) -> str:
    instruction = (
        "Primary pass: classify conservatively."
        if pass_index == 0
        else "Independent skeptical pass: seek alternative explanations and choose EXCLUDE when semantics are unsupported."
    )
    return (
        instruction
        + "\nCandidate JSON:\n"
        + json.dumps([candidate_for_prompt(row) for row in rows], ensure_ascii=False, separators=(",", ":"))
    )


def _transport_rows(rows: Sequence[dict[str, Any]]) -> tuple[list[dict[str, Any]], dict[str, str]]:
    """Use simple batch-local IDs so the model never has to copy path-like UIDs."""
    transported: list[dict[str, Any]] = []
    original_by_transport: dict[str, str] = {}
    for index, row in enumerate(rows):
        transport_uid = f"item-{index:04d}"
        original_by_transport[transport_uid] = row["uid"]
        transported.append({**row, "uid": transport_uid})
    return transported, original_by_transport


def validate_decisions(
    payload: Any,
    expected_uids: Sequence[str],
    allowed_opportunities: dict[str, set[str]] | None = None,
    expected_families: dict[str, str] | None = None,
) -> list[dict[str, Any]]:
    if not isinstance(payload, dict) or not isinstance(payload.get("items"), list):
        raise ValueError("Response must be an object with an items list")
    expected = set(expected_uids)
    decisions: dict[str, dict[str, Any]] = {}
    for item in payload["items"]:
        if not isinstance(item, dict):
            raise ValueError("Every decision must be an object")
        raw_uid = item.get("uid")
        uid = _normalize_returned_uid(raw_uid)
        verdict = item.get("verdict")
        event_type = item.get("event_type")
        subtype = item.get("subtype")
        exclude_reason = item.get("exclude_reason")
        selected_opportunity_uid = item.get("selected_opportunity_uid")
        confidence = item.get("confidence")
        reason = item.get("reason")
        if uid not in expected or uid in decisions:
            raise ValueError(f"Unexpected or duplicate uid: {raw_uid!r}")
        if verdict not in VERDICTS:
            raise ValueError(f"Invalid verdict for {uid}: {verdict!r}")
        if verdict == "POSITIVE" and event_type not in EVENT_TYPES:
            raise ValueError(f"POSITIVE requires WEARER_ENTRY or INT_YIELD for {uid}")
        if verdict != "POSITIVE" and event_type is not None:
            raise ValueError(f"Only POSITIVE may have event_type for {uid}")
        if verdict == "HARD_NEGATIVE" and subtype not in HARD_NEGATIVE_TYPES:
            raise ValueError(
                f"HARD_NEGATIVE requires a valid subtype for {uid}; got {subtype!r}"
            )
        if verdict != "HARD_NEGATIVE" and subtype is not None:
            raise ValueError(f"Only HARD_NEGATIVE may have subtype for {uid}")
        if verdict == "EXCLUDE" and exclude_reason not in EXCLUDE_REASONS:
            raise ValueError(f"EXCLUDE requires a valid exclude_reason for {uid}")
        if verdict != "EXCLUDE" and exclude_reason is not None:
            raise ValueError(f"Only EXCLUDE may have exclude_reason for {uid}")
        family = None if expected_families is None else expected_families.get(str(uid))
        if verdict == "POSITIVE" and family in {"ENTRY", "INT"}:
            required_event_type = "WEARER_ENTRY" if family == "ENTRY" else "INT_YIELD"
            if event_type != required_event_type:
                raise ValueError(f"{family} candidate requires {required_event_type} for {uid}")
        if verdict == "HARD_NEGATIVE" and family in {"ENTRY", "INT"}:
            if subtype != "OTHER_HARD_NEGATIVE" and not str(subtype).startswith(f"{family}_"):
                raise ValueError(f"{family} candidate requires an {family}_* hard-negative subtype for {uid}")
        allowed = set() if allowed_opportunities is None else allowed_opportunities.get(str(uid), set())
        if verdict != "POSITIVE" or event_type != "WEARER_ENTRY":
            # This field is auxiliary attribution metadata, not part of the
            # semantic label. DeepSeek occasionally copies an opportunity uid
            # into an INT/HN/EXCLUDE result despite the output contract. Drop
            # that irrelevant value instead of retrying and losing the batch.
            selected_opportunity_uid = None
        elif selected_opportunity_uid is not None:
            if selected_opportunity_uid not in allowed:
                raise ValueError(f"WEARER_ENTRY may select only a supplied opportunity uid for {uid}")
        if isinstance(confidence, bool) or not isinstance(confidence, (int, float)) or not 0 <= confidence <= 1:
            raise ValueError(f"confidence must be in [0, 1] for {uid}")
        if not isinstance(reason, str) or not reason.strip():
            raise ValueError(f"Missing reason for {uid}")
        decisions[str(uid)] = {
            "uid": str(uid),
            "verdict": str(verdict),
            "event_type": event_type,
            "subtype": subtype,
            "exclude_reason": exclude_reason,
            "exclude_reason_group": exclude_reason_group(exclude_reason),
            "selected_opportunity_uid": selected_opportunity_uid,
            "confidence": float(confidence),
            "reason": reason.strip(),
            # Every AI decision enters the human queue. This is deliberately
            # local rather than model-generated, so harmless JSON type drift
            # cannot invalidate and retry an otherwise valid batch.
            "needs_human_review": True,
        }
    if set(decisions) != expected:
        raise ValueError(f"Response uid mismatch; missing={sorted(expected - set(decisions))}")
    return [decisions[uid] for uid in expected_uids]


def _post(
    *,
    rows: Sequence[dict[str, Any]],
    pass_index: int,
    api_key: str,
    model: str,
    base_url: str,
    timeout_seconds: float,
    max_retries: int,
) -> tuple[list[dict[str, Any]], dict[str, Any]]:
    transported, original_by_transport = _transport_rows(rows)
    body = {
        "model": model,
        "messages": [
            {"role": "system", "content": SYSTEM_PROMPT},
            {"role": "user", "content": build_user_prompt(transported, pass_index)},
        ],
        "response_format": {"type": "json_object"},
        "thinking": {"type": "disabled"},
        "temperature": 0,
        "max_tokens": 2048,
        "stream": False,
    }
    request = urllib.request.Request(
        base_url.rstrip("/") + "/chat/completions",
        data=json.dumps(body, ensure_ascii=False).encode("utf-8"),
        headers={"Authorization": f"Bearer {api_key}", "Content-Type": "application/json"},
        method="POST",
    )
    last_error: Exception | None = None
    for attempt in range(max_retries + 1):
        try:
            with urllib.request.urlopen(request, timeout=timeout_seconds) as response:
                envelope = json.loads(response.read().decode("utf-8"))
        except (
            urllib.error.HTTPError,
            urllib.error.URLError,
            TimeoutError,
        ) as error:
            last_error = error
            if attempt >= max_retries:
                raise APIRequestError(
                    f"API request failed after {max_retries + 1} attempts: {last_error}"
                ) from error
            time.sleep(min(30.0, 2.0**attempt))
            continue
        try:
            content = envelope["choices"][0]["message"]["content"]
            parsed = json.loads(content)
            allowed_opportunities = {}
            expected_families = {}
            for row in transported:
                evidence = row.get("temporal_evidence") or {}
                options = evidence.get("opportunity_options") or []
                allowed_opportunities[row["uid"]] = {
                    str(option["uid"])
                    for option in options
                    if isinstance(option, dict) and option.get("uid") is not None
                }
                if not allowed_opportunities[row["uid"]] and evidence.get("opportunity_source_uid"):
                    allowed_opportunities[row["uid"]].add(str(evidence["opportunity_source_uid"]))
                task = str(row.get("task", "")).lower()
                if task.startswith("wearer_entry") or task.startswith("eot"):
                    expected_families[row["uid"]] = "ENTRY"
                elif task.startswith("int"):
                    expected_families[row["uid"]] = "INT"
            decisions = validate_decisions(
                parsed, list(original_by_transport), allowed_opportunities, expected_families
            )
            for decision in decisions:
                decision["uid"] = original_by_transport[decision["uid"]]
            return decisions, envelope.get("usage", {})
        except (
            KeyError,
            TypeError,
            json.JSONDecodeError,
            ValueError,
        ) as error:
            last_error = error
            # A malformed multi-item answer is immediately bisected. Retrying
            # the identical large prompt is both expensive and unlikely to
            # identify which item caused the schema violation. A singleton is
            # retried before it is placed in the human queue.
            if len(rows) > 1:
                raise BatchValidationError(str(error)) from error
            if attempt >= max_retries:
                raise BatchValidationError(
                    f"singleton validation failed after {max_retries + 1} attempts: {error}"
                ) from error
            time.sleep(min(30.0, 2.0**attempt))
    raise RuntimeError(f"Unreachable request state: {last_error}")


def _atomic_write_json(path: Path, payload: dict[str, Any]) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    temporary = path.with_name(f".{path.name}.{os.getpid()}.tmp")
    try:
        temporary.write_text(
            json.dumps(payload, ensure_ascii=False, indent=2) + "\n", encoding="utf-8"
        )
        os.replace(temporary, path)
    finally:
        if temporary.exists():
            temporary.unlink()


def _atomic_write_jsonl(path: Path, rows: Sequence[dict[str, Any]]) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    temporary = path.with_name(f".{path.name}.{os.getpid()}.tmp")
    try:
        with temporary.open("w", encoding="utf-8") as handle:
            for row in rows:
                handle.write(json.dumps(row, ensure_ascii=False) + "\n")
        os.replace(temporary, path)
    finally:
        if temporary.exists():
            temporary.unlink()


def _run_fingerprint(
    rows: Sequence[dict[str, Any]], *, model: str, base_url: str, passes: int, batch_size: int
) -> str:
    payload = {
        "prompt_version": PROMPT_VERSION,
        "model": model,
        "base_url": base_url.rstrip("/"),
        "passes": passes,
        "batch_size": batch_size,
        "candidates": [candidate_for_prompt(row) for row in rows],
    }
    return hashlib.sha256(
        json.dumps(payload, ensure_ascii=False, sort_keys=True, separators=(",", ":")).encode(
            "utf-8"
        )
    ).hexdigest()


def _initial_checkpoint(
    rows: Sequence[dict[str, Any]], *, model: str, base_url: str, passes: int,
    batch_size: int,
) -> dict[str, Any]:
    return {
        "schema_version": 1,
        "status": "running",
        "run_fingerprint": _run_fingerprint(
            rows, model=model, base_url=base_url, passes=passes, batch_size=batch_size
        ),
        "config": {
            "prompt_version": PROMPT_VERSION,
            "model": model,
            "base_url": base_url.rstrip("/"),
            "passes": passes,
            "batch_size": batch_size,
        },
        "candidate_uids": [row["uid"] for row in rows],
        "decisions": {},
        "prompt_hashes": {},
        "pending": {},
        "usage": {},
    }


def add_suggestions(
    rows: Sequence[dict[str, Any]],
    *,
    api_key: str,
    model: str,
    base_url: str,
    passes: int,
    batch_size: int = 8,
    timeout_seconds: float,
    max_retries: int,
    checkpoint_path: Path | None = None,
    pending_output_path: Path | None = None,
    resume: bool = True,
) -> tuple[list[dict[str, Any]], list[dict[str, Any]], Counter[str]]:
    if batch_size <= 0:
        raise ValueError("batch_size must be positive")
    expected_state = _initial_checkpoint(
        rows, model=model, base_url=base_url, passes=passes, batch_size=batch_size
    )
    if checkpoint_path is not None and resume and checkpoint_path.exists():
        state = json.loads(checkpoint_path.read_text(encoding="utf-8"))
        if state.get("run_fingerprint") != expected_state["run_fingerprint"]:
            raise ValueError(
                f"Checkpoint does not match this input/model/prompt: {checkpoint_path}. "
                "Pass --overwrite to start a new run."
            )
    else:
        state = expected_state
        if checkpoint_path is not None:
            _atomic_write_json(checkpoint_path, state)

    decisions_by_uid: dict[str, dict[str, dict[str, Any]]] = state["decisions"]
    hashes_by_uid: dict[str, dict[str, str]] = state["prompt_hashes"]
    pending_by_uid: dict[str, dict[str, Any]] = state["pending"]
    usage: Counter[str] = Counter(state.get("usage", {}))
    source_by_uid = {row["uid"]: row for row in rows}

    def materialize_pending() -> list[dict[str, Any]]:
        result: list[dict[str, Any]] = []
        for uid in state["candidate_uids"]:
            if uid not in pending_by_uid:
                continue
            pending = dict(source_by_uid[uid])
            pending.update(
                human_status="UNREVIEWED",
                human_verdict=None,
                human_event_type=None,
                human_subtype=None,
                exclude_reason=None,
            )
            pending["llm_adjudication_failure"] = pending_by_uid[uid]
            result.append(pending)
        return result

    def save_checkpoint() -> None:
        state["usage"] = dict(usage)
        if checkpoint_path is not None:
            _atomic_write_json(checkpoint_path, state)
        if pending_output_path is not None:
            _atomic_write_jsonl(pending_output_path, materialize_pending())

    total_work = len(rows) * passes
    completed_work = len(pending_by_uid) * passes + sum(
        len(pass_decisions)
        for uid, pass_decisions in decisions_by_uid.items()
        if uid not in pending_by_uid
    )
    progress = _ProgressBar(total_work)
    if completed_work:
        print(
            f"Resuming {checkpoint_path}: completed={completed_work}/{total_work} "
            f"pending_human={len(pending_by_uid)}"
        )
    progress.update(completed_work, f"pass 1/{passes} · waiting")

    def process_batch(batch: Sequence[dict[str, Any]], pass_index: int) -> None:
        nonlocal completed_work
        if not batch:
            return
        transported, _ = _transport_rows(batch)
        prompt_hash = hashlib.sha256(
            (SYSTEM_PROMPT + "\n" + build_user_prompt(transported, pass_index)).encode("utf-8")
        ).hexdigest()
        progress.update(completed_work, f"pass {pass_index + 1}/{passes} · request {len(batch)}")
        try:
            decisions, pass_usage = _post(
                rows=batch,
                pass_index=pass_index,
                api_key=api_key,
                model=model,
                base_url=base_url,
                timeout_seconds=timeout_seconds,
                max_retries=max_retries,
            )
        except BatchValidationError as error:
            if len(batch) > 1:
                midpoint = len(batch) // 2
                progress.update(
                    completed_work,
                    f"validation split {len(batch)}→{midpoint}+{len(batch) - midpoint}",
                )
                process_batch(batch[:midpoint], pass_index)
                process_batch(batch[midpoint:], pass_index)
                return
            uid = batch[0]["uid"]
            pending_by_uid[uid] = {
                "failed_pass": pass_index + 1,
                "error_type": type(error).__name__,
                "error": str(error),
                "model": model,
                "base_url": base_url.rstrip("/"),
                "prompt_version": PROMPT_VERSION,
                "request_prompt_sha256": prompt_hash,
            }
            completed_work += passes - pass_index
            save_checkpoint()
            progress.update(completed_work, f"queued for human review: {uid}")
            return

        pass_key = str(pass_index)
        for decision in decisions:
            uid = decision["uid"]
            decisions_by_uid.setdefault(uid, {})[pass_key] = decision
            hashes_by_uid.setdefault(uid, {})[pass_key] = prompt_hash
        for key, value in pass_usage.items():
            if isinstance(value, int):
                usage[key] += value
        completed_work += len(batch)
        save_checkpoint()
        progress.update(completed_work, f"pass {pass_index + 1}/{passes} · saved {len(batch)}")

    try:
        for pass_index in range(passes):
            eligible = [
                row for row in rows
                if row["uid"] not in pending_by_uid
                and str(pass_index) not in decisions_by_uid.get(row["uid"], {})
            ]
            for start in range(0, len(eligible), batch_size):
                process_batch(eligible[start:start + batch_size], pass_index)
    except Exception:
        progress.update(completed_work, "failed; checkpoint retained", final=True)
        raise
    progress.update(completed_work, "complete", final=True)

    output: list[dict[str, Any]] = []
    for source in rows:
        uid = source["uid"]
        if uid in pending_by_uid:
            continue
        pass_decisions = decisions_by_uid.get(uid, {})
        missing_passes = [index + 1 for index in range(passes) if str(index) not in pass_decisions]
        if missing_passes:
            raise RuntimeError(f"Internal checkpoint error: {uid} missing passes {missing_passes}")
        decisions = [pass_decisions[str(index)] for index in range(passes)]
        labels = {
            (d["verdict"], d["event_type"], d["subtype"], d["exclude_reason"])
            for d in decisions
        }
        unanimous = len(labels) == 1
        if unanimous:
            verdict, event_type, subtype, exclude_reason = next(iter(labels))
            selected_options = {d["selected_opportunity_uid"] for d in decisions}
            selected_opportunity_uid = (
                next(iter(selected_options)) if len(selected_options) == 1 else None
            )
            confidence = min(float(item["confidence"]) for item in decisions)
            reason = " | ".join(item["reason"] for item in decisions)
            needs_human_review = any(item["needs_human_review"] for item in decisions)
        else:
            verdict, event_type, subtype, exclude_reason, selected_opportunity_uid = (
                "EXCLUDE", None, None, "INSUFFICIENT_EVIDENCE", None
            )
            confidence = min(float(item["confidence"]) for item in decisions)
            reason = "The independent DeepSeek passes disagreed."
            needs_human_review = True
        opportunity_options = (source.get("temporal_evidence") or {}).get("opportunity_options") or []
        selected_opportunity = next(
            (
                option for option in opportunity_options
                if isinstance(option, dict) and option.get("uid") == selected_opportunity_uid
            ),
            None,
        )
        selected_opportunity_seconds = (
            None if selected_opportunity is None else selected_opportunity.get("anchor_seconds")
        )
        behavior_seconds = (source.get("temporal_evidence") or {}).get(
            "behavior_anchor_seconds", source.get("anchor_dense_seconds")
        )
        enriched = dict(source)
        enriched["llm_suggestion"] = {
            "verdict": verdict,
            "event_type": event_type,
            "subtype": subtype,
            "exclude_reason": exclude_reason,
            "exclude_reason_group": exclude_reason_group(exclude_reason),
            "selected_opportunity_uid": selected_opportunity_uid,
            "selected_opportunity_seconds": selected_opportunity_seconds,
            "valid_prediction_interval_seconds": (
                None
                if selected_opportunity_seconds is None
                else [selected_opportunity_seconds, behavior_seconds]
            ),
            "confidence": confidence,
            "reason": reason,
            "needs_human_review": needs_human_review,
            "unanimous": unanimous,
        }
        enriched.update(
            ai_verdict=verdict,
            ai_event_type=event_type,
            ai_subtype=subtype,
            ai_exclude_reason=exclude_reason,
            ai_exclude_reason_group=exclude_reason_group(exclude_reason),
            ai_selected_opportunity_uid=selected_opportunity_uid,
            ai_selected_opportunity_seconds=selected_opportunity_seconds,
            ai_confidence=confidence,
            ai_reason=reason,
            human_status="UNREVIEWED",
            human_verdict=None,
            human_event_type=None,
            human_subtype=None,
            exclude_reason=None,
        )
        enriched["llm_adjudication"] = {
            "model": model,
            "base_url": base_url.rstrip("/"),
            "prompt_version": PROMPT_VERSION,
            "passes": passes,
            "request_prompt_sha256": [
                hashes_by_uid[uid][str(index)] for index in range(passes)
            ],
            "decisions": decisions,
            "human_review_required": True,
        }
        output.append(enriched)
    state["status"] = "complete"
    save_checkpoint()
    pending_output = materialize_pending()
    return output, pending_output, usage


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--input", type=Path, required=True)
    parser.add_argument("--output", type=Path, required=True)
    parser.add_argument("--model", default=os.environ.get("DEEPSEEK_MODEL", DEFAULT_MODEL))
    parser.add_argument("--base-url", default=os.environ.get("DEEPSEEK_BASE_URL", DEFAULT_BASE_URL))
    parser.add_argument("--passes", type=int, default=2)
    parser.add_argument("--batch-size", type=int, default=8)
    parser.add_argument("--max-events", type=int)
    parser.add_argument("--selection-seed", type=int, default=2027)
    parser.add_argument("--timeout-seconds", type=float, default=120.0)
    parser.add_argument("--max-retries", type=int, default=3)
    parser.add_argument(
        "--checkpoint",
        type=Path,
        help="Resume state (default: OUTPUT.checkpoint.json).",
    )
    parser.add_argument(
        "--pending-output",
        type=Path,
        help="Singleton validation failures (default: OUTPUT stem + _pending.jsonl).",
    )
    parser.add_argument("--dry-run", action="store_true")
    parser.add_argument("--allow-external-transcript-upload", action="store_true")
    parser.add_argument("--overwrite", action="store_true")
    return parser.parse_args()


def main() -> None:
    args = parse_args()
    if args.passes <= 0 or args.batch_size <= 0 or args.max_retries < 0:
        raise ValueError("passes/batch-size must be positive and max-retries non-negative")
    input_rows = read_candidates(args.input.expanduser().resolve())
    rows = [row for row in input_rows if requires_semantic_adjudication(row)]
    skipped_controls = len(input_rows) - len(rows)
    if args.max_events is not None:
        if args.max_events < 0:
            raise ValueError("--max-events must be non-negative")
        rng = random.Random(args.selection_seed)
        rows = rng.sample(rows, min(args.max_events, len(rows)))
        rows.sort(key=lambda row: row["uid"])
    print(
        f"Candidates={len(rows)} skipped_deterministic_controls={skipped_controls} "
        f"model={args.model} passes={args.passes}"
    )
    if args.dry_run:
        print("DRY RUN: no API request will be made.\n")
        print(build_user_prompt(rows, 0))
        return
    if not args.allow_external_transcript_upload:
        raise PermissionError(
            "Refusing to upload transcript excerpts; inspect --dry-run, then pass "
            "--allow-external-transcript-upload when authorized."
        )
    api_key = os.environ.get("DEEPSEEK_API_KEY")
    if not api_key:
        raise RuntimeError("Set DEEPSEEK_API_KEY in the environment")
    output = args.output.expanduser().resolve()
    if output.exists() and not args.overwrite:
        raise FileExistsError(f"Refusing to overwrite {output}; pass --overwrite")
    checkpoint = (
        args.checkpoint.expanduser().resolve()
        if args.checkpoint is not None
        else output.with_suffix(output.suffix + ".checkpoint.json")
    )
    pending_output_path = (
        args.pending_output.expanduser().resolve()
        if args.pending_output is not None
        else output.with_name(output.stem + "_pending.jsonl")
    )
    enriched, pending, usage = add_suggestions(
        rows,
        api_key=api_key,
        model=args.model,
        base_url=args.base_url,
        passes=args.passes,
        batch_size=args.batch_size,
        timeout_seconds=args.timeout_seconds,
        max_retries=args.max_retries,
        checkpoint_path=checkpoint,
        pending_output_path=pending_output_path,
        resume=not args.overwrite,
    )
    _atomic_write_jsonl(output, enriched)
    _atomic_write_jsonl(pending_output_path, pending)
    print(f"Saved {len(enriched)} suggested candidates to {output}")
    print(f"Saved {len(pending)} pending-human candidates to {pending_output_path}")
    print(f"Checkpoint: {checkpoint}")
    print("Usage:", dict(usage))


if __name__ == "__main__":
    main()