"""Tests for wearer-entry/INT DeepSeek suggestion validation without network calls.""" from __future__ import annotations import unittest from unittest import mock from pathlib import Path import json import tempfile from adjudicate_easycom_eot_int_candidates import ( APIRequestError, BatchValidationError, PROMPT_VERSION, SYSTEM_PROMPT, _progress_line, _transport_rows, add_suggestions, build_user_prompt, exclude_reason_group, read_candidates, requires_semantic_adjudication, validate_decisions, ) class EntryINTAdjudicationTests(unittest.TestCase): @staticmethod def _rows(count: int) -> list[dict[str, object]]: return [ {"uid": f"u{index}", "candidate_source": "WEARER_ONSET", "task": "wearer_entry"} for index in range(count) ] @staticmethod def _decisions(rows: list[dict[str, object]]) -> list[dict[str, object]]: return [ { "uid": row["uid"], "verdict": "EXCLUDE", "event_type": None, "subtype": None, "exclude_reason": "INSUFFICIENT_EVIDENCE", "exclude_reason_group": "INSUFFICIENT_EVIDENCE", "selected_opportunity_uid": None, "confidence": 0.5, "reason": "Needs human context.", "needs_human_review": True, } for row in rows ] 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-v16") 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("whether the wearer activity run ending there ever held the floor", SYSTEM_PROMPT) self.assertIn("OUT_OF_SCOPE_WEARER_BACKCHANNEL_OFFSET", SYSTEM_PROMPT) self.assertIn("do not count its trailing offset 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("OUT_OF_SCOPE_WEARER_BACKCHANNEL_OFFSET"), "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"})) def test_validation_failure_recursively_splits_batch_to_singletons(self) -> None: rows = self._rows(4) batch_sizes: list[int] = [] def fake_post(**kwargs): batch = list(kwargs["rows"]) batch_sizes.append(len(batch)) if len(batch) > 1: raise BatchValidationError("one item has an invalid subtype") return self._decisions(batch), {"total_tokens": 10} with tempfile.TemporaryDirectory() as temporary, mock.patch( "adjudicate_easycom_eot_int_candidates._post", side_effect=fake_post ): output, pending, usage = add_suggestions( rows, api_key="test", model="model", base_url="https://example.test/v1", passes=1, batch_size=4, timeout_seconds=1, max_retries=0, checkpoint_path=Path(temporary) / "checkpoint.json", ) self.assertEqual(batch_sizes, [4, 2, 1, 1, 2, 1, 1]) self.assertEqual(len(output), 4) self.assertEqual(pending, []) self.assertEqual(usage["total_tokens"], 40) def test_singleton_validation_failure_goes_to_pending_queue(self) -> None: rows = self._rows(1) with tempfile.TemporaryDirectory() as temporary, mock.patch( "adjudicate_easycom_eot_int_candidates._post", side_effect=BatchValidationError("invalid singleton label"), ): checkpoint = Path(temporary) / "checkpoint.json" pending_path = Path(temporary) / "pending.jsonl" output, pending, _ = add_suggestions( rows, api_key="test", model="model", base_url="https://example.test/v1", passes=2, batch_size=8, timeout_seconds=1, max_retries=0, checkpoint_path=checkpoint, pending_output_path=pending_path, ) state = json.loads(checkpoint.read_text(encoding="utf-8")) saved_pending = [ json.loads(line) for line in pending_path.read_text(encoding="utf-8").splitlines() ] self.assertEqual(output, []) self.assertEqual([row["uid"] for row in pending], ["u0"]) self.assertIn("invalid singleton label", pending[0]["llm_adjudication_failure"]["error"]) self.assertEqual(state["status"], "complete") self.assertIn("u0", state["pending"]) self.assertEqual([row["uid"] for row in saved_pending], ["u0"]) def test_resume_skips_batches_already_saved_in_checkpoint(self) -> None: rows = self._rows(4) with tempfile.TemporaryDirectory() as temporary: checkpoint = Path(temporary) / "checkpoint.json" first_sizes: list[int] = [] def interrupted_post(**kwargs): batch = list(kwargs["rows"]) first_sizes.append(len(batch)) if len(first_sizes) == 2: raise APIRequestError("temporary provider outage") return self._decisions(batch), {"total_tokens": 20} with mock.patch( "adjudicate_easycom_eot_int_candidates._post", side_effect=interrupted_post ), self.assertRaises(APIRequestError): add_suggestions( rows, api_key="test", model="model", base_url="https://example.test/v1", passes=1, batch_size=2, timeout_seconds=1, max_retries=0, checkpoint_path=checkpoint, ) resumed_uids: list[list[str]] = [] def resumed_post(**kwargs): batch = list(kwargs["rows"]) resumed_uids.append([str(row["uid"]) for row in batch]) return self._decisions(batch), {"total_tokens": 20} with mock.patch( "adjudicate_easycom_eot_int_candidates._post", side_effect=resumed_post ): output, pending, usage = add_suggestions( rows, api_key="test", model="model", base_url="https://example.test/v1", passes=1, batch_size=2, timeout_seconds=1, max_retries=0, checkpoint_path=checkpoint, ) self.assertEqual(first_sizes, [2, 2]) self.assertEqual(resumed_uids, [["u2", "u3"]]) self.assertEqual(len(output), 4) self.assertEqual(pending, []) self.assertEqual(usage["total_tokens"], 40) if __name__ == "__main__": unittest.main()