Something went wrong. Try again.
a tiny pet with an even tinier brain, living in your browser
Something went wrong. Try again.
17 kB · 285 lines
Python
at main
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286"""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 Counterfrom copy import deepcopyimport hashlibimport jsonfrom pathlib import Pathimport sysimport tempfileimport unittest
sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "train"))import gen_data as Gimport model_data as Dimport semantic_data as Simport 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()