Document target resolution V1 diagnostic

tests/test_target_resolution_v1_diagnostic.py

@@ -0,0 +1,59 @@
+import argparse,copy,json,tempfile,unittest
+from pathlib import Path
+from unittest.mock import Mock
+import src.meeting_lab.controlled_semantic_derivation.experiment_target_resolution_v1 as module
+from src.meeting_lab.controlled_semantic_derivation.experiment_h import DerivationValidationError
+
+CASES=module.load_cases(Path("tests/gold/target_resolution_v1/cases.json")); BY={c["case_id"]:c for c in CASES}
+def self_output(text="Zusammenarbeit mit Dr. Schlummer fortsetzen"): return {"candidate_observation_id":"obs_1","normalized_target_text":text}
+def paired(case,target="obs_1",text="externe Lösung weiterverfolgen"): return {"candidate_observation_id":case["negative_act"]["observation_id"],"target_observation_id":target,"normalized_target_text":text}
+
+class TargetResolutionV1Tests(unittest.TestCase):
+ def test_self_linkage_is_deterministic_and_equals_candidate(self):
+  result=module.deterministic_self_link(BY["TR1-V1"]); self.assertEqual(result["linkage_source"],"deterministic"); self.assertEqual(result["target_observation_id"],result["candidate_observation_id"])
+ def test_self_llm_output_has_no_target_id(self):
+  self.assertEqual(module.SELF_KEYS,{"candidate_observation_id","normalized_target_text"}); self.assertNotIn("target_observation_id",module.output_schema(BY["TR1-V1"])["properties"])
+ def test_self_combination_records_link_and_normalization_separately(self):
+  combined=module.combine(BY["TR1-V1"],self_output()); self.assertEqual(combined["target_observation_id"],"obs_1"); self.assertIn("fortsetzen",combined["normalized_target_text"])
+ def test_self_link_rejects_paired_strategy(self):
+  with self.assertRaises(DerivationValidationError): module.deterministic_self_link(BY["TR2-V1"])
+ def test_paired_schema_enumerates_allowed_ids_and_null(self):
+  schema=module.output_schema(BY["TR2-V1"]); self.assertEqual(schema["properties"]["target_observation_id"]["enum"],["obs_1","obs_2",None]); self.assertFalse(schema["additionalProperties"])
+ def test_allowed_obs1_and_obs2_are_structurally_accepted(self):
+  case=BY["TR2-V1"]
+  module.validate_semantic_output(paired(case,"obs_1"),case); module.validate_semantic_output(paired(case,"obs_2"),case)
+ def test_unknown_and_string_null_targets_rejected(self):
+  case=BY["TR2-V1"]
+  for value in ("obs_9","null"):
+   with self.assertRaises(DerivationValidationError): module.validate_semantic_output(paired(case,value),case)
+ def test_json_null_accepted_and_requires_null_text(self):
+  case=BY["TR2-V1"]; valid=paired(case,None,None); self.assertEqual(module.validate_semantic_output(valid,case),valid)
+  with self.assertRaises(DerivationValidationError): module.validate_semantic_output(paired(case,None,"x"),case)
+ def test_non_null_requires_nonempty_text(self):
+  with self.assertRaises(DerivationValidationError): module.validate_semantic_output(paired(BY["TR2-V1"],"obs_1",""),BY["TR2-V1"])
+ def test_target_after_candidate_rejected(self):
+  case=copy.deepcopy(BY["TR2-V1"]); case["negative_act"]["observation_id"]="obs_1"
+  with self.assertRaises(DerivationValidationError): module.validate_semantic_output({"candidate_observation_id":"obs_1","target_observation_id":"obs_2","normalized_target_text":"x"},case)
+ def test_duplicate_ids_and_evidence_rejected(self):
+  for field in ("observation_id","evidence_id"):
+   obs=copy.deepcopy(BY["TR2-V1"]["observations"]); obs[1][field]=obs[0][field]
+   with self.assertRaises(DerivationValidationError): module.validate_observations(obs)
+ def test_forbidden_and_unknown_fields_rejected(self):
+  case=BY["TR2-V1"]
+  for extra in ({"status":"x"},{"nested":{"decision":True}},{"extra":1}):
+   data=paired(case); data.update(extra)
+   with self.assertRaises(DerivationValidationError): module.validate_semantic_output(data,case)
+ def test_payload_uses_true_schema_object(self):
+  schema=module.output_schema(BY["TR2-V1"]); payload=module.build_payload("qwen3.5:9B","p",schema,16384,1024); self.assertIs(payload["format"],schema); self.assertIsInstance(payload["format"],dict); self.assertEqual(payload["options"]["temperature"],0)
+ def test_prompt_has_typed_examples_and_allowed_ids(self):
+  prompt=module.build_prompt(BY["TR2-V1"]); self.assertIn('["obs_1", "obs_2"]',prompt); self.assertIn('"target_observation_id":null',prompt); self.assertIn('Never return the string "null"',prompt); self.assertNotIn('observation ID or null',prompt)
+ def test_self_normalization_must_be_positive(self):
+  case=BY["TR1-V1"]; semantic=self_output("Mit Dr. Schlummer arbeiten wir nicht weiter."); combined=module.combine(case,semantic); self.assertEqual(module.evaluate(case,semantic,combined)["classification"],"FAIL")
+ def test_no_rejection_derivation_exists(self):
+  self.assertFalse(hasattr(module,"derive")); self.assertNotIn("status",module.PAIRED_KEYS); self.assertNotIn("explicitly_rejected",module.PAIRED_KEYS)
+ def test_runner_artifacts_distinguish_linkage_and_normalization(self):
+  case=BY["TR1-V1"]; fixture={"schema_version":module.SCHEMA_VERSION,"cases":[case]}; caller=Mock(return_value=(json.dumps(self_output()),{"model":"qwen3.5:9B"}))
+  with tempfile.TemporaryDirectory() as tmp:
+   root=Path(tmp); path=root/"cases.json"; path.write_text(json.dumps(fixture)); out=root/"out"; args=argparse.Namespace(cases=path,output=out,endpoint="x",model="qwen3.5:9B",timeout=1,num_ctx=16384,num_predict=1024); summary=module.run(args,caller); self.assertEqual(summary["llm_call_count"],1); self.assertTrue((out/"tr1-v1"/"deterministic_linkage_result.json").exists()); self.assertTrue((out/"tr1-v1"/"normalized_target_result.json").exists())
+
+if __name__=="__main__": unittest.main()