Add controlled request-acceptance derivation experiment
This commit is contained in:
@@ -0,0 +1,215 @@
|
||||
import json
|
||||
import tempfile
|
||||
import unittest
|
||||
from copy import deepcopy
|
||||
from pathlib import Path
|
||||
from unittest.mock import patch
|
||||
|
||||
from src.meeting_lab.controlled_semantic_derivation.experiment_h import (
|
||||
SCHEMA_VERSION,
|
||||
DerivationValidationError,
|
||||
build_ollama_payload,
|
||||
derive_action,
|
||||
load_v3_observations,
|
||||
parse_model_json,
|
||||
run_experiment,
|
||||
validate_semantic_recognition,
|
||||
)
|
||||
|
||||
|
||||
ACCEPTED_H_PATH = Path(
|
||||
"artifacts/experiments/evidence_observations_v3/20260819_v3_single_run/"
|
||||
"h_resulting_action/parsed_observations.json"
|
||||
)
|
||||
|
||||
|
||||
class ControlledSemanticDerivationHTests(unittest.TestCase):
|
||||
def setUp(self) -> None:
|
||||
self.observations = [
|
||||
{
|
||||
"observation_id": "obs_1", "evidence_id": "e1",
|
||||
"content": "Antonius: Nina, übernimmst du die Prüfung der Messdaten bis Friday?",
|
||||
"speaker": "Antonius", "named_person": "Nina", "addressee": "Nina",
|
||||
},
|
||||
{
|
||||
"observation_id": "obs_2", "evidence_id": "e2",
|
||||
"content": "Nina: Ja, ich übernehme die Prüfung bis Freitag.",
|
||||
"speaker": "Nina", "named_person": None, "addressee": None,
|
||||
},
|
||||
]
|
||||
self.recognition = {
|
||||
"schema_version": SCHEMA_VERSION,
|
||||
"request": {
|
||||
"observation_id": "obs_1", "is_concrete_request": True,
|
||||
"normalized_action_text": "Prüfung der Messdaten",
|
||||
},
|
||||
"acceptance": {
|
||||
"observation_id": "obs_2", "is_explicit_commitment": True,
|
||||
"same_requested_work": True,
|
||||
"normalized_action_text": "die Prüfung",
|
||||
},
|
||||
}
|
||||
|
||||
def derive(self, observations=None, recognition=None):
|
||||
return derive_action(
|
||||
observations if observations is not None else self.observations,
|
||||
recognition if recognition is not None else self.recognition,
|
||||
)
|
||||
|
||||
def test_actual_accepted_v3_artifact_is_the_experiment_input(self):
|
||||
actual = load_v3_observations(ACCEPTED_H_PATH)
|
||||
self.assertEqual(actual, self.observations)
|
||||
|
||||
def test_valid_semantic_recognition_has_no_derivation_fields(self):
|
||||
validate_semantic_recognition(self.recognition, self.observations)
|
||||
serialized = json.dumps(self.recognition)
|
||||
for forbidden in ("responsible_person", "requested_actor", "status", "established", "action_item"):
|
||||
self.assertNotIn(forbidden, serialized)
|
||||
|
||||
def test_request_and_acceptance_provenance_survive(self):
|
||||
gates, result = self.derive()
|
||||
self.assertTrue(all(gates.values()))
|
||||
self.assertEqual(result["support"]["request"], {"observation_id": "obs_1", "evidence_id": "e1"})
|
||||
self.assertEqual(result["support"]["acceptance"], {"observation_id": "obs_2", "evidence_id": "e2"})
|
||||
|
||||
def test_valid_sequence_establishes_expected_action(self):
|
||||
recognition = deepcopy(self.recognition)
|
||||
recognition["request"]["normalized_action_text"] = "Prüfung der Messdaten bis Friday"
|
||||
_, result = self.derive(recognition=recognition)
|
||||
self.assertEqual(result["content"], "Prüfung der Messdaten")
|
||||
self.assertEqual(result["status"], "established")
|
||||
self.assertEqual(result["requested_actor"], "Nina")
|
||||
self.assertEqual(result["responsible_person"], "Nina")
|
||||
self.assertEqual(result["due"], "Freitag")
|
||||
|
||||
def test_lexical_identity_is_not_required(self):
|
||||
self.assertNotEqual(
|
||||
self.recognition["request"]["normalized_action_text"],
|
||||
self.recognition["acceptance"]["normalized_action_text"],
|
||||
)
|
||||
gates, result = self.derive()
|
||||
self.assertTrue(gates["same_requested_work"])
|
||||
self.assertIsNotNone(result)
|
||||
|
||||
def test_request_alone_does_not_establish(self):
|
||||
gates, result = self.derive(observations=self.observations[:1])
|
||||
self.assertFalse(gates["acceptance_observation_exists"])
|
||||
self.assertIsNone(result)
|
||||
|
||||
def test_noncommitting_or_acknowledging_response_does_not_establish(self):
|
||||
recognition = deepcopy(self.recognition)
|
||||
recognition["acceptance"]["is_explicit_commitment"] = False
|
||||
gates, result = self.derive(recognition=recognition)
|
||||
self.assertFalse(gates["acceptance_semantic_positive"])
|
||||
self.assertIsNone(result)
|
||||
|
||||
def test_different_response_speaker_does_not_establish(self):
|
||||
observations = deepcopy(self.observations)
|
||||
observations[1]["speaker"] = "Martin"
|
||||
gates, result = self.derive(observations=observations)
|
||||
self.assertFalse(gates["acceptance_speaker_matches_addressee"])
|
||||
self.assertIsNone(result)
|
||||
|
||||
def test_different_accepted_work_does_not_establish(self):
|
||||
recognition = deepcopy(self.recognition)
|
||||
recognition["acceptance"]["same_requested_work"] = False
|
||||
recognition["acceptance"]["normalized_action_text"] = "Angebot prüfen"
|
||||
gates, result = self.derive(recognition=recognition)
|
||||
self.assertFalse(gates["same_requested_work"])
|
||||
self.assertIsNone(result)
|
||||
|
||||
def test_tentative_acceptance_does_not_establish(self):
|
||||
recognition = deepcopy(self.recognition)
|
||||
recognition["acceptance"]["is_explicit_commitment"] = False
|
||||
_, result = self.derive(recognition=recognition)
|
||||
self.assertIsNone(result)
|
||||
|
||||
def test_named_person_speaker_or_addressee_alone_cannot_establish(self):
|
||||
recognition = deepcopy(self.recognition)
|
||||
recognition["acceptance"]["is_explicit_commitment"] = False
|
||||
gates, result = self.derive(recognition=recognition)
|
||||
self.assertEqual(self.observations[0]["named_person"], "Nina")
|
||||
self.assertEqual(self.observations[0]["addressee"], "Nina")
|
||||
self.assertEqual(self.observations[1]["speaker"], "Nina")
|
||||
self.assertFalse(gates["acceptance_semantic_positive"])
|
||||
self.assertIsNone(result)
|
||||
|
||||
def test_acceptance_must_follow_request(self):
|
||||
observations = list(reversed(deepcopy(self.observations)))
|
||||
gates, result = self.derive(observations=observations)
|
||||
self.assertFalse(gates["acceptance_after_request"])
|
||||
self.assertIsNone(result)
|
||||
|
||||
def test_conflicting_deadlines_do_not_establish(self):
|
||||
observations = deepcopy(self.observations)
|
||||
observations[1]["content"] = "Nina: Ja, ich übernehme die Prüfung bis Donnerstag."
|
||||
gates, result = self.derive(observations=observations)
|
||||
self.assertFalse(gates["deadline_consistent"])
|
||||
self.assertIsNone(result)
|
||||
|
||||
def test_unknown_observation_reference_is_rejected(self):
|
||||
recognition = deepcopy(self.recognition)
|
||||
recognition["acceptance"]["observation_id"] = "obs_9"
|
||||
with self.assertRaisesRegex(DerivationValidationError, "unknown observation"):
|
||||
validate_semantic_recognition(recognition, self.observations)
|
||||
|
||||
def test_inconsistent_evidence_provenance_is_rejected(self):
|
||||
data = json.loads(ACCEPTED_H_PATH.read_text())
|
||||
data["observations"][1]["evidence_id"] = "e1"
|
||||
with tempfile.TemporaryDirectory() as temporary:
|
||||
path = Path(temporary) / "observations.json"
|
||||
path.write_text(json.dumps(data), encoding="utf-8")
|
||||
with self.assertRaisesRegex(DerivationValidationError, "inconsistent evidence provenance"):
|
||||
load_v3_observations(path)
|
||||
|
||||
def test_responsibility_or_status_in_llm_output_is_rejected(self):
|
||||
for field in ("responsibility", "responsible_person", "status", "established"):
|
||||
recognition = deepcopy(self.recognition)
|
||||
recognition[field] = "forbidden"
|
||||
with self.subTest(field=field), self.assertRaisesRegex(DerivationValidationError, "forbidden semantic keys"):
|
||||
validate_semantic_recognition(recognition, self.observations)
|
||||
|
||||
def test_protocol_or_unrelated_semantic_concepts_are_rejected(self):
|
||||
for field in ("protocol_category", "decision", "unresolved_issue", "graph", "confidence"):
|
||||
recognition = deepcopy(self.recognition)
|
||||
recognition[field] = "forbidden"
|
||||
with self.subTest(field=field), self.assertRaisesRegex(DerivationValidationError, "forbidden semantic keys"):
|
||||
validate_semantic_recognition(recognition, self.observations)
|
||||
|
||||
def test_malformed_json_is_rejected(self):
|
||||
with self.assertRaises(json.JSONDecodeError):
|
||||
parse_model_json("{bad json")
|
||||
|
||||
def test_payload_has_one_call_controls(self):
|
||||
payload = build_ollama_payload("qwen3.5:9B", "prompt", 16384, 1024)
|
||||
self.assertFalse(payload["think"])
|
||||
self.assertFalse(payload["stream"])
|
||||
self.assertEqual(payload["options"]["temperature"], 0)
|
||||
|
||||
def test_run_preserves_all_artifacts_without_real_ollama(self):
|
||||
raw = json.dumps(self.recognition, ensure_ascii=False)
|
||||
from argparse import Namespace
|
||||
|
||||
with tempfile.TemporaryDirectory() as temporary:
|
||||
output = Path(temporary) / "run"
|
||||
args = Namespace(
|
||||
observations=ACCEPTED_H_PATH, output=output, model="qwen3.5:9B",
|
||||
endpoint="http://unused", timeout=1, num_ctx=16384, num_predict=1024,
|
||||
)
|
||||
with patch(
|
||||
"src.meeting_lab.controlled_semantic_derivation.experiment_h.call_ollama",
|
||||
return_value=(raw, {"model": "qwen3.5:9B"}),
|
||||
):
|
||||
summary = run_experiment(args)
|
||||
self.assertTrue(summary["action_established"])
|
||||
for filename in (
|
||||
"v3_input_observations.json", "prompt.txt", "raw_model_response.txt",
|
||||
"parsed_semantic_recognition.json", "structural_validation.json",
|
||||
"deterministic_gate_results.json", "final_derived_result.json",
|
||||
"ollama_metadata.json", "summary.json",
|
||||
):
|
||||
self.assertTrue((output / filename).is_file(), filename)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
Reference in New Issue
Block a user