"""Semantic behavior boundaries; python3 test/semantic_rules.py. Shared-engine integration cases need the existing tokenling-contract binary; they are explicitly skipped when it has not been built, never simulated. """ from collections import Counter from copy import deepcopy import hashlib import json from pathlib import Path import sys import tempfile import unittest sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "train")) import gen_data as G import model_data as D import semantic_data as S import world as W BINARY = Path(__file__).resolve().parents[1] / "server/target/debug/tokenling-contract" class SemanticRules(unittest.TestCase): @classmethod def setUpClass(cls): cls.temp = tempfile.TemporaryDirectory() root = Path(cls.temp.name) c0 = root / "c0.jsonl" c0.write_text(json.dumps({"text": "an independent old synthetic corpus"}) + "\n") cls.manifest_path = root / "manifest.json" cls.manifest = S.freeze_manifest(c0, cls.manifest_path) cls.cases = S.eval_cases(cls.manifest, "dev", S.SEEDS["dev"][0]) @classmethod def tearDownClass(cls): cls.temp.cleanup() def case(self, family="header_recall", relation="age"): return deepcopy(next(case for case in self.cases if case["family"] == family and case["target"]["relation"] == relation)) def record(self, text, results=None): return dict(text=text, delivered=True, contract_valid=True, attempts=[dict(app_written_results=results or [])]) def evidence(self, case, added=()): before = S.EngineAdapter.save_for_state(case["state"]) after = deepcopy(before) after["memories"].extend(added) return dict(before=before, after=after) def test_exact_grid_and_fixed_disjoint_seed_groups(self): groups, templates, entity_combinations = {}, {}, {} for pool in ("dev", "sealed"): for seed in S.SEEDS[pool]: cases = S.eval_cases(self.manifest, pool, seed) self.assertEqual(len(cases), 256) self.assertEqual(Counter((case["family"], case["stage"]) for case in cases), Counter({(family, stage): 16 for family in S.FAMILIES for stage in W.STAGES})) these = {case["group_id"] for case in cases} self.assertEqual(len(these), 256) self.assertFalse(any(these & previous for previous in groups.values())) groups[(pool, seed)] = these templates.setdefault(pool, set()).update(case["template_id"] for case in cases) entity_combinations.setdefault(pool, set()).update(tuple(case["entity_combination"]) for case in cases) self.assertFalse(templates["dev"] & templates["sealed"]) self.assertFalse(entity_combinations["dev"] & entity_combinations["sealed"]) self.assertEqual(S.load_manifest(self.manifest_path), self.manifest) with self.assertRaises(ValueError): S.eval_cases(self.manifest, "dev", S.SEEDS["sealed"][0]) def test_training_grid_has_real_slots_and_no_foreign_inputs(self): cases = S.training_cases(self.manifest, 85710, per_cell=32) self.assertEqual(len(cases), 512) self.assertEqual(len({case["group_id"] for case in cases}), 512) for case in cases: self.assertTrue(case["scene"]["slots"]) self.assertEqual(case["gold"]["slots"][0]["facts"], [case["target"]["fact"]] if case["family"] == "capture" else []) prompt = G.curriculum_prompt(case["state"], case["scene"]) self.assertEqual(S.isolation_check(prompt, self.manifest), []) self.assertNotIn("semantic/", W.header("chat", case["state"])) def test_group_value_and_template_collisions_are_rejected(self): broken = deepcopy(self.manifest) seed = str(S.SEEDS["sealed"][0]) broken["evaluation_cases"]["sealed"][seed][1] = deepcopy(broken["evaluation_cases"]["sealed"][seed][0]) self.assertTrue(any(issue["kind"] == "group" for issue in S.audit_manifest(broken))) broken = deepcopy(self.manifest) broken["templates"]["sealed"]["ask"] = broken["templates"]["dev"]["ask"] self.assertTrue(any(issue["kind"] == "template" for issue in S.audit_manifest(broken))) broken = deepcopy(self.manifest) broken["values"]["age"]["sealed"][0] = broken["values"]["age"]["dev"][0] self.assertTrue(any(issue["kind"] == "value" for issue in S.audit_manifest(broken))) case = self.case() self.assertTrue(S.isolation_check(dict(text=case["state"]["owner"], group_id=case["group_id"]), self.manifest)) raw_c0 = case["scene"]["slots"][0]["events"][0] self.assertTrue(any(issue["kind"] in ("c0_template", "c0_group") for issue in S.audit_manifest(self.manifest, raw_c0))) def test_wrong_number_name_and_role_cannot_pass(self): case = self.case() target = case["target"] for prose in (f"you are {int(target['value']) + 7} years old", target["fact"].replace(target["entity"], case["state"]["pet"]), f"i am {target['value']} years old"): judgment = S.assess_turn(case, self.record(prose), self.evidence(case)) self.assertEqual(judgment["status"], "fail", judgment) cat = self.case(relation="cat") wrong = cat["target"]["fact"].replace(cat["target"]["value"], "Changedname") self.assertEqual(S.assess_turn(cat, self.record(wrong), self.evidence(cat))["status"], "fail") def test_open_prose_is_not_a_fabricated_deterministic_contradiction(self): case = self.case("capture", "scared") fact = case["target"]["fact"] text = fact + " ohh cool detail!! [remember:" + fact + "]" judgment = S.assess_turn(case, self.record(text), self.evidence(case, [fact])) self.assertEqual(judgment["status"], "uncertain") self.assertFalse(judgment["failures"]) case = self.case("capture", "visit") fact = case["target"]["fact"] text = "hehe i remember that " + fact + " [remember:" + fact + "]" judgment = S.assess_turn(case, self.record(text), self.evidence(case, [fact])) self.assertEqual(judgment["status"], "uncertain") self.assertFalse(judgment["failures"]) case["state"]["memories"] = [fact] judgment = S.assess_turn(case, self.record(fact), self.evidence(case), slot_index=1) self.assertNotEqual(judgment["status"], "fail") def test_nonassertive_mentions_remain_reviewable_not_approved(self): absent = self.case("abstention", "age") prose = "i don't know whether you are " + absent["target"]["value"] + " years old" judgment = S.assess_turn(absent, self.record(prose), self.evidence(absent)) self.assertEqual(judgment["status"], "uncertain") self.assertFalse(judgment["failures"]) answer = self.case("header_recall", "cat") wrong = answer["gold"]["adversary"]["wrong_value"] prose = "not " + wrong + "; your cat is " + answer["target"]["value"] judgment = S.assess_turn(answer, self.record(prose), self.evidence(answer)) self.assertEqual(judgment["status"], "uncertain") self.assertFalse(judgment["failures"]) fact = answer["target"]["fact"] judgment = S.assess_turn(answer, self.record(fact + "?"), self.evidence(answer)) self.assertEqual(judgment["status"], "uncertain") capture = self.case("capture") question = capture["target"]["fact"] + "?" judgment = S.assess_turn(capture, self.record("got it [remember:" + question + "]"), self.evidence(capture, [question])) self.assertEqual(judgment["status"], "fail") fictional = self.case("abstention", "cat") prose = "in a fictional story, " + fictional["gold"]["forbidden_facts"][0] judgment = S.assess_turn(fictional, self.record(prose), self.evidence(fictional)) self.assertEqual(judgment["status"], "uncertain") self.assertFalse(judgment["failures"]) def test_written_age_is_not_a_false_hard_number_contradiction(self): for family in ("capture", "header_recall"): case = self.case(family) fact = case["state"]["owner"] + " is 30 years old" case["target"].update(value="30", key="30", fact=fact) case["gold"].update(value="30", fact=fact) case["gold"]["slots"][0]["facts"] = [fact] if family == "capture" else [] case["state"]["memories"] = [] if family == "capture" else [fact] text = "you are thirty years old" + (" [remember:" + fact + "]" if family == "capture" else "") judgment = S.assess_turn(case, self.record(text), self.evidence(case, [fact] if family == "capture" else [])) self.assertEqual(judgment["status"], "uncertain") self.assertFalse(judgment["failures"]) def test_hidden_answer_and_matching_keyword_are_not_answers(self): case = self.case() fact = case["target"]["fact"] for text in (f"[recall:old]{{{fact}}}", f"i know! [recall:old]{{{fact}}}", f"[remember:{fact}]"): self.assertEqual(S.assess_turn(case, self.record(text), self.evidence(case))["status"], "fail") self.assertEqual(S.assess_turn(case, self.record(case["target"]["value"]), self.evidence(case))["status"], "uncertain") self.assertEqual(S.assess_turn(case, self.record(fact), self.evidence(case))["status"], "pass") self.assertEqual(S.assess_turn(case, self.record(fact + ", or perhaps a different age"), self.evidence(case))["status"], "uncertain") def test_capture_needs_actual_complete_correct_persistence(self): case = self.case("capture") fact = case["target"]["fact"] record = self.record(f"got it [remember:{fact}]") self.assertEqual(S.assess_turn(case, record, self.evidence(case))["status"], "fail") self.assertEqual(S.assess_turn(case, record, self.evidence(case, [fact]))["status"], "pass") self.assertEqual(S.assess_turn(case, record, self.evidence(case, [fact, "Another false fact"]))["status"], "fail") self.assertEqual(S.assess_turn(case, self.record(f"got it [remember:{fact}] [diary:invented note]"), self.evidence(case, [fact]))["status"], "fail") swapped = fact.replace(case["state"]["owner"], case["state"]["pet"]) self.assertEqual(S.assess_turn(case, self.record(f"got it [remember:{swapped}]"), self.evidence(case, [swapped]))["status"], "fail") def test_absence_and_irrelevance_never_approve_owner_fact_invention(self): case = self.case("abstention") fact = case["target"]["fact"] self.assertEqual(S.assess_turn(case, self.record("i don't know"), self.evidence(case))["status"], "pass") self.assertEqual(S.assess_turn(case, self.record(fact), self.evidence(case))["status"], "fail") self.assertEqual(S.assess_turn(case, self.record(f"okay [remember:{fact}]"), self.evidence(case, [fact]))["status"], "fail") unrelated = deepcopy(next(c for c in self.cases if c["family"] == "abstention" and c["gold"]["slots"][0]["kind"] == "irrelevant")) fact = unrelated["target"]["fact"] self.assertEqual(S.assess_turn(unrelated, self.record(f"okay [remember:{fact}]"), self.evidence(unrelated, [fact]))["status"], "fail") def review_fixture(self, case, text): record = self.record(text) judgment = S.assess_turn(case, record, self.evidence(case)) record.update(state=deepcopy(case["state"]), lines=case["scene"]["slots"][0]["events"][:], semantic_case=case, semantic_slot_index=0, semantic_raw_judgment=judgment) identities = dict(source_report_sha256="source", manifest_sha256=self.manifest["sha256"], model_identity_sha256="model", graph_sha256="actual-graph") review = dict(identities, case_id=case["id"], slot_index=0, case_sha256=case["sha256"], text_sha256=hashlib.sha256(text.encode()).hexdigest(), reviewer="independent reviewer", reviewer_kind="ai-assisted", decision="pass", evidence=dict(visible_prose=judgment["visible_prose"], entity=case["target"]["entity"], relation=case["target"]["relation"], value=case["target"]["value"], source_facts=[case["target"]["fact"]], rationale="relation and value agree with supplied memory", persistence_explanation="no new memory; actual engine memories unchanged")) return record, review, identities def test_independent_review_is_exact_source_bound_and_cannot_waive_failures(self): case = self.case() record, review, identities = self.review_fixture(case, f"i remember your age: {case['target']['value']}!") self.assertEqual(record["semantic_raw_judgment"]["status"], "uncertain") judged = S.apply_review(record, review, identities) self.assertEqual(judged["status"], "pass") self.assertEqual(judged["raw_status"], "uncertain") for field in ("text_sha256", "case_sha256", "graph_sha256", "source_report_sha256", "manifest_sha256"): invalid = deepcopy(review) invalid[field] = "different" with self.assertRaises(ValueError): S.apply_review(record, invalid, identities) record, review, identities = self.review_fixture(case, f"you are {int(case['target']['value']) + 1} years old") with self.assertRaises(ValueError): S.apply_review(record, review, identities) invalid = deepcopy(review) invalid["evidence"]["source_facts"] = ["a nonexistent memory"] with self.assertRaises(ValueError): S.apply_review(record, invalid, identities) def test_historical_contract_is_unchanged_unless_explicit(self): old = D.evaluation_contract(123, 40, 8, 10) self.assertEqual(old["version"], 3) self.assertNotIn("semantic", old) new = D.evaluation_contract(123, 40, 8, 10, semantic={"manifest_sha256": "frozen"}) self.assertEqual(new["version"], 4) self.assertEqual({k: v for k, v in new.items() if k not in ("version", "semantic")}, {k: v for k, v in old.items() if k != "version"}) @unittest.skipUnless(BINARY.is_file(), "build existing Rust tokenling-contract for actual persistence") def test_actual_shared_engine_persists_and_rejects_uncommitted_effects(self): engine = S.EngineAdapter(BINARY) case = self.case("capture") fact = case["target"]["fact"] engine.parity(case["state"], case["target"]["kw"]) complete = engine.commit(case["state"], f"got it [remember:{fact}]") self.assertIn(fact, complete["after"]["memories"]) self.assertEqual(S.assess_turn(case, self.record(f"got it [remember:{fact}]"), complete)["status"], "pass") for body in (f"got it [remember:{fact}", f"got it [remember:{fact}]{{invented result}}"): evidence = engine.commit(case["state"], body) self.assertNotIn(fact, evidence["after"]["memories"]) sleeping = deepcopy(case["state"]) sleeping["asleep"] = True evidence = engine.commit(sleeping, f"got it [remember:{fact}]") self.assertNotIn(fact, evidence["after"]["memories"]) @unittest.skipUnless(BINARY.is_file(), "build existing Rust tokenling-contract for actual persistence") def test_fractional_body_stats_keep_native_header_thresholds(self): engine = S.EngineAdapter(BINARY) case = self.case() for value in (0.0, 20.49, 20.5, 40.49, 40.5, 60.49, 60.5, 80.49, 80.5, 100.0): state = deepcopy(case["state"]) state["stats"] = dict.fromkeys(("tummy", "fun", "energy", "clean"), value) engine.parity(state, case["target"]["kw"]) @unittest.skipUnless(BINARY.is_file(), "build existing Rust tokenling-contract for actual persistence") def test_actual_old_lookup_training_scene_uses_authentic_braces(self): case = self.case("older_recall") replies = [f"[recall:{case['target']['kw']}] {case['target']['fact']}", "okay"] row = G.replay_curriculum(deepcopy(case["state"]), case["scene"], replies) checked = S.check_scene(case, row, BINARY) self.assertEqual(checked["status"], "pass", checked) self.assertIn("{" + W.tool_result(case["state"], "recall", case["target"]["kw"]) + "}", row["text"]) wrong = deepcopy(row) wrong["text"] = wrong["text"].replace("{" + case["target"]["fact"] + "}", "{nothing}") self.assertEqual(S.check_scene(case, wrong, BINARY)["status"], "fail") if __name__ == "__main__": unittest.main()