264 lines
13 KiB
Python
264 lines
13 KiB
Python
"""Tests for wearer-entry/INT DeepSeek suggestion validation without network calls."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import unittest
|
|
from pathlib import Path
|
|
import json
|
|
import tempfile
|
|
|
|
from adjudicate_easycom_eot_int_candidates import (
|
|
PROMPT_VERSION, SYSTEM_PROMPT, _progress_line, _transport_rows, build_user_prompt,
|
|
exclude_reason_group, read_candidates,
|
|
requires_semantic_adjudication, validate_decisions,
|
|
)
|
|
|
|
|
|
class EntryINTAdjudicationTests(unittest.TestCase):
|
|
def test_progress_line_reports_items_percent_pass_and_elapsed_time(self) -> None:
|
|
line = _progress_line(25, 100, label="pass 1/2 · batch 4/13", elapsed_seconds=65)
|
|
self.assertIn("25/100 ( 25.0%)", line)
|
|
self.assertIn("pass 1/2 · batch 4/13", line)
|
|
self.assertTrue(line.endswith("01:05"))
|
|
|
|
def test_prompt_contains_task_and_context(self) -> None:
|
|
prompt = build_user_prompt(
|
|
[{"uid": "u1", "candidate_source": "WEARER_ONSET", "task": "wearer_entry",
|
|
"temporal_evidence": {"wearer_duration": 2.0},
|
|
"context": [{"transcription": "Yes."}]}],
|
|
0,
|
|
)
|
|
self.assertIn("WEARER_ONSET", prompt)
|
|
self.assertIn("wearer_duration", prompt)
|
|
self.assertIn("Yes.", prompt)
|
|
|
|
def test_system_prompt_treats_laughter_as_nonlexical_floor_feedback(self) -> None:
|
|
self.assertEqual(PROMPT_VERSION, "easycom-wearer-boundary-semantic-prelabel-v15")
|
|
self.assertIn("opportunity_options", SYSTEM_PROMPT)
|
|
self.assertIn("t_w < t_o < t_y never proves takeover", SYSTEM_PROMPT)
|
|
self.assertIn("Every supplied WEARER_ONSET and WEARER_OFFSET is a raw observed", SYSTEM_PROMPT)
|
|
self.assertIn("a transcription consisting of [L] denotes laughter-only vocalization", SYSTEM_PROMPT)
|
|
self.assertIn("ENTRY_WEARER_BACKCHANNEL", SYSTEM_PROMPT)
|
|
self.assertIn("INT_OTHER_BACKCHANNEL", SYSTEM_PROMPT)
|
|
self.assertIn("Subtype precedence", SYSTEM_PROMPT)
|
|
self.assertIn("same non-wearer resumes", SYSTEM_PROMPT)
|
|
self.assertIn("different non-wearer takes", SYSTEM_PROMPT)
|
|
self.assertIn("This is not an uncertainty fallback", SYSTEM_PROMPT)
|
|
self.assertIn("Apply this adjacency-pair rule before using brevity", SYSTEM_PROMPT)
|
|
self.assertIn("Floor entry here means making the committed conversational response", SYSTEM_PROMPT)
|
|
self.assertIn("does not answer a question, accept or reject a proposal", SYSTEM_PROMPT)
|
|
self.assertIn('the wearer answers "Nope." at 54.15 s', SYSTEM_PROMPT)
|
|
self.assertIn("participant 3 later speaks again", SYSTEM_PROMPT)
|
|
self.assertIn("does not require a preceding non-wearer release", SYSTEM_PROMPT)
|
|
self.assertIn("duration alone never determines floor function", SYSTEM_PROMPT)
|
|
self.assertIn("normal wearer completion", SYSTEM_PROMPT)
|
|
self.assertIn("relevant non-wearer actually takes", SYSTEM_PROMPT)
|
|
self.assertIn("must not retroactively invalidate the preceding INT_YIELD", SYSTEM_PROMPT)
|
|
self.assertIn("do not infer that a visible smile occurred", SYSTEM_PROMPT)
|
|
self.assertIn("POSITIVE / INT_YIELD: subtype=null; exclude_reason=null", SYSTEM_PROMPT)
|
|
self.assertIn("OUT_OF_SCOPE_WITHIN_TURN_CONTINUATION", SYSTEM_PROMPT)
|
|
self.assertIn("DATA_TIMESTAMP_CORRUPTION", SYSTEM_PROMPT)
|
|
self.assertNotIn("NOT_APPLICABLE", SYSTEM_PROMPT)
|
|
self.assertNotIn("UNSURE", SYSTEM_PROMPT)
|
|
self.assertNotIn('"needs_human_review"', SYSTEM_PROMPT)
|
|
|
|
def test_decisions_require_exact_uid_and_vocabulary(self) -> None:
|
|
payload = {
|
|
"items": [
|
|
{
|
|
"uid": "u1",
|
|
"verdict": "HARD_NEGATIVE",
|
|
"event_type": None,
|
|
"subtype": "INT_OTHER_BACKCHANNEL",
|
|
"confidence": 0.91,
|
|
"reason": "Brief feedback while the wearer continues.",
|
|
"needs_human_review": False,
|
|
}
|
|
]
|
|
}
|
|
decisions = validate_decisions(payload, ["u1"])
|
|
self.assertEqual(decisions[0]["subtype"], "INT_OTHER_BACKCHANNEL")
|
|
self.assertTrue(decisions[0]["needs_human_review"])
|
|
with self.assertRaisesRegex(ValueError, "mismatch"):
|
|
validate_decisions({"items": []}, ["u1"])
|
|
|
|
def test_model_needs_human_review_type_drift_is_ignored(self) -> None:
|
|
payload = {
|
|
"items": [{
|
|
"uid": "u1", "verdict": "EXCLUDE", "event_type": None,
|
|
"subtype": None, "exclude_reason": "INSUFFICIENT_EVIDENCE",
|
|
"confidence": 0.4, "reason": "Ambiguous context.",
|
|
"needs_human_review": "true",
|
|
}]
|
|
}
|
|
decision = validate_decisions(payload, ["u1"])[0]
|
|
self.assertTrue(decision["needs_human_review"])
|
|
|
|
def test_removed_fallback_verdicts_are_rejected(self) -> None:
|
|
for verdict in ("UNSURE", "NOT_APPLICABLE"):
|
|
payload = {"items": [{
|
|
"uid": "u1", "verdict": verdict, "event_type": None,
|
|
"subtype": None, "confidence": 0.4, "reason": "Ambiguous context.",
|
|
}]}
|
|
with self.subTest(verdict=verdict), self.assertRaisesRegex(ValueError, "Invalid verdict"):
|
|
validate_decisions(payload, ["u1"])
|
|
|
|
def test_entry_positive_may_select_only_a_supplied_release(self) -> None:
|
|
item = {
|
|
"uid": "u1", "verdict": "POSITIVE", "event_type": "WEARER_ENTRY",
|
|
"subtype": None, "selected_opportunity_uid": "release-2",
|
|
"confidence": 0.9, "reason": "The wearer answers the second release.",
|
|
}
|
|
decision = validate_decisions(
|
|
{"items": [item]}, ["u1"], {"u1": {"release-1", "release-2"}}
|
|
)[0]
|
|
self.assertEqual(decision["selected_opportunity_uid"], "release-2")
|
|
with self.assertRaisesRegex(ValueError, "only a supplied opportunity"):
|
|
validate_decisions(
|
|
{"items": [{**item, "selected_opportunity_uid": "invented"}]},
|
|
["u1"],
|
|
{"u1": {"release-1", "release-2"}},
|
|
)
|
|
optional = validate_decisions(
|
|
{"items": [{**item, "selected_opportunity_uid": None}]},
|
|
["u1"], {"u1": {"release-1", "release-2"}},
|
|
)[0]
|
|
self.assertIsNone(optional["selected_opportunity_uid"])
|
|
|
|
def test_irrelevant_selected_opportunity_is_cleared_without_retrying_label(self) -> None:
|
|
for verdict, event_type, subtype in (
|
|
("POSITIVE", "INT_YIELD", None),
|
|
("HARD_NEGATIVE", None, "INT_OTHER_BACKCHANNEL"),
|
|
("EXCLUDE", None, None),
|
|
):
|
|
with self.subTest(verdict=verdict):
|
|
item = {
|
|
"uid": "u1",
|
|
"verdict": verdict,
|
|
"event_type": event_type,
|
|
"subtype": subtype,
|
|
"exclude_reason": (
|
|
"INSUFFICIENT_EVIDENCE" if verdict == "EXCLUDE" else None
|
|
),
|
|
"selected_opportunity_uid": "copied-release",
|
|
"confidence": 0.8,
|
|
"reason": "Evidence-based decision.",
|
|
}
|
|
decision = validate_decisions(
|
|
{"items": [item]}, ["u1"],
|
|
{"u1": {"copied-release"}},
|
|
{"u1": "INT"},
|
|
)[0]
|
|
self.assertIsNone(decision["selected_opportunity_uid"])
|
|
|
|
def test_exclude_reason_taxonomy_separates_scope_evidence_and_data(self) -> None:
|
|
self.assertEqual(
|
|
exclude_reason_group("OUT_OF_SCOPE_NORMAL_COMPLETION"), "OUT_OF_SCOPE"
|
|
)
|
|
self.assertEqual(
|
|
exclude_reason_group("INSUFFICIENT_EVIDENCE"), "INSUFFICIENT_EVIDENCE"
|
|
)
|
|
self.assertEqual(
|
|
exclude_reason_group("DATA_ANNOTATION_ERROR"), "DATA_QUALITY"
|
|
)
|
|
base = {
|
|
"uid": "u1", "verdict": "EXCLUDE", "event_type": None,
|
|
"subtype": None, "selected_opportunity_uid": None,
|
|
"confidence": 0.4, "reason": "Not a scored event.",
|
|
}
|
|
with self.assertRaisesRegex(ValueError, "valid exclude_reason"):
|
|
validate_decisions({"items": [base]}, ["u1"])
|
|
with self.assertRaisesRegex(ValueError, "valid exclude_reason"):
|
|
validate_decisions(
|
|
{"items": [{**base, "exclude_reason": "ANNOTATION_ERROR"}]}, ["u1"]
|
|
)
|
|
|
|
def test_event_and_hard_negative_subtypes_match_candidate_family(self) -> None:
|
|
positive = {
|
|
"uid": "u1", "verdict": "POSITIVE", "event_type": "INT_YIELD",
|
|
"subtype": None, "selected_opportunity_uid": None,
|
|
"confidence": 0.8, "reason": "Wrong family.",
|
|
}
|
|
with self.assertRaisesRegex(ValueError, "ENTRY candidate requires WEARER_ENTRY"):
|
|
validate_decisions(
|
|
{"items": [positive]}, ["u1"], expected_families={"u1": "ENTRY"}
|
|
)
|
|
hard_negative = {
|
|
"uid": "u1", "verdict": "HARD_NEGATIVE", "event_type": None,
|
|
"subtype": "ENTRY_OTHER_MIDTURN_PAUSE", "selected_opportunity_uid": None,
|
|
"confidence": 0.8, "reason": "Wrong family.",
|
|
}
|
|
with self.assertRaisesRegex(ValueError, "INT candidate requires an INT_"):
|
|
validate_decisions(
|
|
{"items": [hard_negative]}, ["u1"], expected_families={"u1": "INT"}
|
|
)
|
|
|
|
def test_markdown_escaped_uid_is_normalized_but_still_unique(self) -> None:
|
|
item = {
|
|
"uid": "Session\\_4/wearer\\_onset/00000024",
|
|
"verdict": "POSITIVE",
|
|
"event_type": "WEARER_ENTRY",
|
|
"subtype": None,
|
|
"confidence": 0.9,
|
|
"reason": "Sustained wearer response after the other speaker stops.",
|
|
"needs_human_review": False,
|
|
}
|
|
expected = "Session_4/wearer_onset/00000024"
|
|
decision = validate_decisions({"items": [item]}, [expected])[0]
|
|
self.assertEqual(decision["uid"], expected)
|
|
with self.assertRaisesRegex(ValueError, "duplicate"):
|
|
validate_decisions({"items": [item, {**item, "uid": expected}]}, [expected])
|
|
|
|
def test_duplicate_rule_uids_are_disambiguated_with_provenance(self) -> None:
|
|
with tempfile.TemporaryDirectory() as temporary:
|
|
path = Path(temporary) / "candidates.jsonl"
|
|
base = {"uid": "Session_4/other_to_other/00003600", "task": "wearer_entry"}
|
|
rows = [
|
|
{**base, "previous_other_participant_ids": [6], "incoming_other_participant_ids": [4]},
|
|
{**base, "previous_other_participant_ids": [3], "incoming_other_participant_ids": [6]},
|
|
]
|
|
path.write_text("".join(json.dumps(row) + "\n" for row in rows), encoding="utf-8")
|
|
loaded = read_candidates(path)
|
|
self.assertEqual(len({row["uid"] for row in loaded}), 2)
|
|
self.assertTrue(all(row["uid_disambiguated_from"] == base["uid"] for row in loaded))
|
|
|
|
def test_transport_ids_are_short_unique_and_reversible(self) -> None:
|
|
rows = [{"uid": "Session_4/other_to_other/00003600/a"},
|
|
{"uid": "Session_4/other_to_other/00003600/b"}]
|
|
transported, mapping = _transport_rows(rows)
|
|
self.assertEqual([row["uid"] for row in transported], ["item-0000", "item-0001"])
|
|
self.assertEqual(mapping["item-0001"], rows[1]["uid"])
|
|
|
|
def test_every_current_wearer_offset_is_sent_for_adjudication(self) -> None:
|
|
control = {
|
|
"candidate_source": "WEARER_OFFSET",
|
|
"temporal_evidence": {
|
|
"active_other_ids_before_offset": [],
|
|
"active_other_ids_after_offset": [],
|
|
},
|
|
}
|
|
crossing = {
|
|
"candidate_source": "WEARER_OFFSET",
|
|
"temporal_evidence": {
|
|
"active_other_ids_before_offset": [4],
|
|
"active_other_ids_after_offset": [4],
|
|
},
|
|
}
|
|
speaker_change = {
|
|
"candidate_source": "WEARER_OFFSET",
|
|
"temporal_evidence": {
|
|
"active_other_ids_before_offset": [3],
|
|
"active_other_ids_after_offset": [4],
|
|
},
|
|
}
|
|
self.assertTrue(requires_semantic_adjudication(control))
|
|
self.assertTrue(requires_semantic_adjudication(crossing))
|
|
self.assertTrue(requires_semantic_adjudication(speaker_change))
|
|
self.assertFalse(requires_semantic_adjudication({"candidate_source": "WEARER_OFFSET_CONTROL"}))
|
|
self.assertFalse(requires_semantic_adjudication({"candidate_source": "WEARER_ONSET_CONTROL"}))
|
|
self.assertFalse(requires_semantic_adjudication({"candidate_source": "WEARER_RESUME"}))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|