Something went wrong. Try again.
a tiny pet with an even tinier brain, living in your browser
Something went wrong. Try again.
47 kB · 749 lines
Python
at main
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750"""Data-only, frozen factual fixtures and conservative shared-engine evidence.
No model, Torch or tokenizer imports. Gold is a sidecar, never a world header.Freeze: semantic_data.py freeze --c0 CORPUS.jsonl --out MANIFEST.jsonThe finite canonical recognizer is NOT a general semantic oracle: unsupportedparaphrases are uncertain, never approved by keyword presence."""import argparsefrom collections import Counterfrom copy import deepcopyimport hashlibimport jsonfrom pathlib import Pathimport randomimport reimport subprocess
import gen_data as Gimport world as W
SCHEMA = 1FAMILIES = ("capture", "header_recall", "older_recall", "abstention")SEEDS = {"train": (85710,), "dev": (61001, 61002, 61003), "sealed": (62001, 62002, 62003)}# These input templates, not the frozen production fact grammar, are isolated.TEMPLATES = { "train": {"disclose": "a detail for your notebook: {disclosure}", "ask": "bring back the detail about {topic} for me, please", "irrelevant": "imagine a story where {pet_fact}; this is not a fact about me", "pet_question": "{pet}, how would you feel if {pet_fact}?", "tentative": "is it true that {disclosure}? i am asking, not telling you", "pause": "let us pause and listen to the breeze", "thanks": "thank you for taking your time", "leave": "we can leave that unanswered"}, "dev": {"disclose": "here is something personal to keep in mind: {disclosure}", "ask": "which detail did i share concerning {topic}?", "irrelevant": "in a pretend tale, {pet_fact}; do not treat the tale as my history", "pet_question": "how might you react, {pet}, in a tale where {pet_fact}?", "tentative": "could {disclosure} be correct? that is a question rather than a disclosure", "pause": "shall we spend a moment watching the clouds", "thanks": "i appreciate you listening patiently", "leave": "there is no need to guess the missing detail"}, "sealed": {"disclose": "please keep this little piece of my life: {disclosure}", "ask": "can you retrieve what i told you regarding {topic}?", "irrelevant": "suppose a fictional character says {pet_fact}; that says nothing about me", "pet_question": "what would that be like for you, {pet}, supposing {pet_fact}?", "tentative": "would {disclosure} be a true claim? i have not supplied an answer", "pause": "we could quietly watch the branches sway", "thanks": "your patience means a lot to me", "leave": "that unknown detail may stay unknown"},}# Relation grammar is intentionally shared with production; input wording and# entity/value combinations, not the existence of the relation, are held out.RELATIONS = ( ("age", "{entity} is {value} years old", "old", "my age"), ("cat", "{entity} has a cat named {value}", "cat", "my cat's name"), ("dog", "{entity} has a dog named {value}", "dog", "my dog's name"), ("sister", "{entity} has a sister named {value}", "sister", "my sister's name"), ("friend", "{entity} has a friend named {value}", "friend", "my friend's name"), ("color", "{entity}'s favorite color is {value}", "color", "my favorite color"), ("work", "{entity} works as a {value}", "works", "my job"), ("birthday", "{entity}'s birthday is in {value}", "birthday", "my birthday month"), ("scared", "{entity} is scared of {value}", "scared", "my fear"), ("learning", "{entity} is learning {value}", "learning", "the skill i am learning"), ("live", "{entity} lives in a {value}", "lives", "my home"), ("play", "{entity} plays {value}", "plays", "what i play"), ("morning", "{entity} drinks {value} every morning", "morning", "my morning drink"), ("visit", "{entity} wants to visit {value}", "visit", "my travel wish"), ("hate", "{entity} hates {value}", "hates", "the food i dislike"),)# Compound values are ordinary, interpretable facts, not secret labels. Freeze# reports raw-C0 lexical overlap and rejects actual entity-combination leakage.VALUE_CANDIDATES = { "color": ("copper orange", "moss yellow", "slate violet", "peacock blue", "rose taupe", "seafoam lilac", "ochre pink", "mulberry gray", "saffron teal"), "work": ("clock restorer", "glass engraver", "canal surveyor", "violin repairer", "orchard planner", "map conservator", "harp tuner", "textile archivist", "bridge inspector"), "birthday": ("february", "april", "august", "september", "november", "january", "march", "june", "december", "may", "july", "october"), "scared": ("deep wells", "rustling reeds", "empty tunnels", "steep ladders", "cracking ice", "low bridges", "swarming moths", "dark attics", "swaying towers"), "learning": ("basket weaving", "bell casting", "marble carving", "reed braiding", "glass etching", "sail mending", "map engraving", "stone lettering", "wool carding"), "live": ("converted windmill", "riverside boathouse", "hilltop cabin", "renovated lighthouse", "garden studio", "canal houseboat", "woodland lodge", "courtyard flat", "desert bungalow"), "play": ("croquet doubles", "alto recorder", "table shuffleboard", "bass clarinet", "lawn bowls", "hammered dulcimer", "mountain dulcimer", "beach handball", "tenor banjo"), "morning": ("roasted barley tea", "warm pear juice", "mint oat milk", "hibiscus iced tea", "spiced apple water", "toasted rice tea", "ginger plum juice", "lemon rooibos tea", "chilled almond milk"), "visit": ("the azores", "lake baikal", "the faroe islands", "the carpathian mountains", "the atacama desert", "lake bled", "the orkney islands", "the namib desert", "the dolomites"), "hate": ("pickled turnips", "candied fennel", "roasted okra", "steamed artichokes", "salted radishes", "smoked aubergines", "braised endives", "dried persimmons", "fermented cabbage"),}
def digest(value): return hashlib.sha256(json.dumps(value, sort_keys=True, ensure_ascii=False, separators=(",", ":")).encode()).hexdigest()
def file_sha256(path): h = hashlib.sha256() with open(path, "rb") as source: for chunk in iter(lambda: source.read(1024 * 1024), b""): h.update(chunk) return h.hexdigest()
def folded(text): return " ".join(text.casefold().split())
def contains(text, phrase): return bool(re.search(r"(?<!\w)" + re.escape(folded(phrase)) + r"(?!\w)", folded(text)))
ENTITY_POOLS = { "train": {"owners": ["Avery", "Blair", "Cameron", "Drew", "Ellis", "Finley", "Harper", "Jules", "Morgan", "Parker", "Reese", "Robin", "Rowan", "Sage", "Skyler", "Taylor", "Addison", "Ashley", "Beverly", "Brooke", "Carmen", "Devon", "Gale", "Hollis"], "pets": ["Acorn", "Biscuit", "Bramble", "Button", "Clover", "Comet", "Dumpling", "Fig", "Fudge", "Hazel", "Juniper", "Marble", "Noodle", "Pancake", "Pebble", "Sprout"], "related": ["Bela", "Corin", "Dara"]}, "dev": {"owners": ["Adrian", "Casey", "Emerson", "Hayden", "Jamie", "Jordan", "Kendall", "Lane", "Marley", "Micah", "Payton", "Quinn", "Remy", "Riley", "Shiloh", "Sidney", "Alex", "Bailey", "Cory", "Dallas", "Erin", "Jody", "Kelly", "Logan"], "pets": ["Apricot", "Bean", "Berry", "Cricket", "Doodle", "Fern", "Maple", "Muffin", "Olive", "Peaches", "Pepper", "Pudding", "Pumpkin", "Sunny", "Truffle", "Waffle"], "related": ["Inez", "Luca", "Maren"]}, "sealed": {"owners": ["Alexis", "Arden", "Ashton", "Aubrey", "Briar", "Charlie", "Dakota", "Eden", "Frankie", "Justice", "Kit", "Lee", "Lennox", "Phoenix", "River", "Winter", "Angel", "Ariel", "Bellamy", "Dana", "Dorian", "Leslie", "Rory", "Shawn"], "pets": ["Basil", "Boba", "Chai", "Cloud", "Cocoa", "Cosmo", "Crumble", "Dewdrop", "Flax", "Ginger", "Honey", "Mango", "Mallow", "Nutmeg", "Poppy", "Tumble"], "related": ["Nora", "Otis", "Vera"]},}
def _values(raw_c0): available_ages = [str(n) for n in range(18, 64)] # Finite natural domains may occur in C0. The held-out unit is the novel # entity/relation/value combination, not a claim that a month was never seen. values = {} for relation, _, _, _ in RELATIONS: if relation in ("cat", "dog", "sister", "friend"): candidates = [name for pool in SEEDS for name in ENTITY_POOLS[pool]["related"]] elif relation == "age": candidates = available_ages else: candidates = list(VALUE_CANDIDATES[relation]) if len(candidates) < 3: raise ValueError(f"fewer than three disjoint {relation} values") width = min(9 if relation == "age" else 3, len(candidates) // 3) values[relation] = {pool: candidates[i * width:(i + 1) * width] for i, pool in enumerate(SEEDS)} return values
def _template_signature(text): return folded(re.sub(r"\{[^{}]+\}", "<slot>", text))
def _c0_combinations(rows, values, entities): """Conservatively reserve any same-line owner/value co-occurrence in raw C0.
This includes first-person owner disclosures, malformed/unsaved attempts, tags and headers, not just a corpus producer's claimed memory sidecars. Relation-ambiguous co-occurrences block every matching relation, not pass. """ owners = {folded(owner): (pool, owner) for pool in SEEDS for owner in entities[pool]["owners"]} owner_pattern = re.compile(r"(?<!\w)(?:" + "|".join(map(re.escape, owners)) + r")(?!\w)") patterns = {(pool, relation, value): re.compile(r"(?<!\w)" + re.escape(folded(value)) + r"(?!\w)") for pool in SEEDS for relation in values for value in values[relation][pool]} blocked = {} for row_index, row in enumerate(rows, 1): for line in row["text"].splitlines(): text = folded(line) for match in owner_pattern.finditer(text): pool, owner = owners[match.group()] for relation in values: for value in values[relation][pool]: if patterns[(pool, relation, value)].search(text): key = (owner, relation, value) blocked.setdefault(key, dict(entity=owner, relation=relation, value=value, source_row=row_index, source_line_sha256=hashlib.sha256(line.encode()).hexdigest())) return sorted(blocked.values(), key=lambda item: (item["entity"], item["relation"], item["value"]))
def freeze_manifest(c0_path, output_path): """Freeze identities before generation; refuse overwrite and C0 group leakage.""" c0_path, output_path = Path(c0_path), Path(output_path) raw = c0_path.read_text(encoding="utf-8") rows = [json.loads(line) for line in raw.splitlines() if line.strip()] texts = [row["text"] for row in rows] source_text = "\n".join(texts) values = _values(source_text) manifest = dict(schema_version=SCHEMA, contract="canonical-facts-shared-engine-v1", families=list(FAMILIES), stages=list(W.STAGES), scenes_per_cell=16, seeds={pool: list(seeds) for pool, seeds in SEEDS.items()}, templates=deepcopy(TEMPLATES), values=values, entities=deepcopy(ENTITY_POOLS), c0_blocked_combinations=_c0_combinations(rows, values, ENTITY_POOLS), c0=dict(path=str(c0_path.resolve()), sha256=file_sha256(c0_path), rows=len(rows)), isolation_scope="fine-tuning raw C0 entity/value co-occurrences and input templates; disjoint new pool entity/relation/value groups; not pretraining; canonical relation grammar shared", source_sha256={name: file_sha256(Path(__file__).with_name(name)) for name in ("semantic_data.py", "world.py", "gen_data.py")}) manifest["evaluation_cases"] = {pool: {str(seed): _cases(manifest, pool, seed, 16) for seed in manifest["seeds"][pool]} for pool in ("dev", "sealed")} collisions = audit_manifest(manifest, source_text) if collisions: raise ValueError("semantic manifest collisions: " + json.dumps(collisions[:12])) lexical_overlaps = [dict(relation=relation, pool=pool, value=value) for relation in values for pool in SEEDS for value in values[relation][pool] if contains(source_text, value)] manifest["c0_audit"] = dict(status="pass", raw_text_sha256=hashlib.sha256(source_text.encode()).hexdigest(), checked_rows=len(rows), collisions=[], lexical_value_overlaps=lexical_overlaps, heldout_unit="entity/relation/value combination and input template, not common lexical values", checks=["lexical values in raw text (reported)", "input template literal anchors in raw text", "rendered entity combinations", "cross-pool values/templates/groups/entities"]) manifest["sha256"] = digest(manifest) output_path.parent.mkdir(parents=True, exist_ok=True) with output_path.open("x", encoding="utf-8") as out: json.dump(manifest, out, indent=2, ensure_ascii=False) out.write("\n") return manifest
def load_manifest(path): manifest = json.loads(Path(path).read_text(encoding="utf-8")) expected = manifest.get("sha256") if expected != digest({k: v for k, v in manifest.items() if k != "sha256"}): raise ValueError("semantic manifest digest mismatch") if (manifest.get("schema_version") != SCHEMA or manifest.get("families") != list(FAMILIES) or manifest.get("stages") != list(W.STAGES) or manifest.get("scenes_per_cell") != 16 or manifest.get("seeds") != {k: list(v) for k, v in SEEDS.items()}): raise ValueError("semantic manifest grid/seed contract mismatch") issues = audit_manifest(manifest) if issues: raise ValueError("semantic manifest isolation failure: " + json.dumps(issues[:12])) return manifest
def audit_manifest(manifest, raw_c0=None): """Audit actual content, not merely asserted group labels or different seeds.""" issues, seen_values, seen_templates, seen_groups, seen_entities = [], {}, {}, {}, {} raw_folded = folded(raw_c0) if raw_c0 is not None else None for pool in SEEDS: for relation, _, _, _ in RELATIONS: for value in manifest["values"][relation][pool]: key = (relation, folded(value)) if key in seen_values and seen_values[key] != pool: issues.append(dict(kind="value", pools=[seen_values[key], pool], value=value)) seen_values[key] = pool # Lexical C0 overlaps are recorded in the freeze receipt. Novel # groups and owner input templates, not common months, are held out. for kind, template in manifest["templates"][pool].items(): signature = _template_signature(template) if signature in seen_templates: issues.append(dict(kind="template", pools=[seen_templates[signature], pool], template=template)) seen_templates[signature] = pool # Literal anchors survive entity/value substitution in raw C0. anchors = [folded(x) for x in re.split(r"\{[^{}]+\}", template) if len(folded(x)) >= 12] if raw_folded is not None and any(anchor in raw_folded for anchor in anchors): issues.append(dict(kind="c0_template", pool=pool, template=template)) for seed in manifest["seeds"][pool]: cases = (_cases(manifest, pool, seed, 16) if pool == "train" else manifest["evaluation_cases"][pool][str(seed)]) if len(cases) != 256: issues.append(dict(kind="grid", pool=pool, seed=seed)) grid = Counter((case["family"], case["stage"]) for case in cases) if grid != Counter({(family, stage): 16 for family in FAMILIES for stage in W.STAGES}): issues.append(dict(kind="family_stage_grid", pool=pool, seed=seed)) for case in cases: if case.get("sha256") != digest({k: v for k, v in case.items() if k != "sha256"}): issues.append(dict(kind="case_digest", case=case.get("id"))) if case["group_id"] != digest(case["entity_combination"]): issues.append(dict(kind="group_identity", case=case["id"])) key = case["group_id"] if key in seen_groups: issues.append(dict(kind="group", case=case["id"], other=seen_groups[key])) seen_groups[key] = case["id"] for role in ("owner", "pet"): entity = case["state"][role] if entity in seen_entities and seen_entities[entity].split("/")[1] != pool: issues.append(dict(kind="entity", case=case["id"], other=seen_entities[entity])) seen_entities[entity] = case["id"] if raw_c0 is not None: blocked = _c0_combinations([dict(text=raw_c0)], manifest["values"], manifest["entities"]) blocked_keys = {(item["entity"], item["relation"], item["value"]) for item in blocked} for pool in SEEDS: for seed in manifest["seeds"][pool]: cases = (_cases(manifest, pool, seed, 16) if pool == "train" else manifest["evaluation_cases"][pool][str(seed)]) for case in cases: if (case["target"]["entity"], case["target"]["relation"], case["target"]["value"]) in blocked_keys: issues.append(dict(kind="c0_group", case=case["id"], fact=case["target"]["fact"])) return issues
def _disclosure(fact, owner): if fact.startswith(owner + "'s "): return "my " + fact[len(owner) + 3:] rest = fact[len(owner) + 1:] for verb, first in (("is ", "i am "), ("has ", "i have "), ("works ", "i work "), ("lives ", "i live "), ("plays ", "i play "), ("drinks ", "i drink "), ("wants ", "i want "), ("hates ", "i hate ")): if rest.startswith(verb): return first + rest[len(verb):] raise ValueError("unknown disclosure relation")
def _cases(manifest, pool, seed, per_cell, offset=0): cases = [] for family in FAMILIES: for stage in W.STAGES: for index in range(offset, offset + per_cell): identity = f"{seed}/{family}/{stage}/{index}" rng = random.Random(int(digest([pool, identity]), 16)) relation, template, keyword, topic = RELATIONS[index % len(RELATIONS)] owners = manifest["entities"][pool]["owners"] pets = manifest["entities"][pool]["pets"] values = manifest["values"][relation][pool] blocked = {(item["entity"], item["relation"], item["value"]) for item in manifest["c0_blocked_combinations"]} combinations = [(owner, pet, value) for owner in owners for pet in pets for value in values if (owner, relation, value) not in blocked] relation_ordinal = index // len(RELATIONS) sequence = manifest["seeds"][pool].index(seed) if pool != "train" else 0 # Each family/stage receives an independent slice of the finite # natural combinations. All fixed eval seeds occupy distinct # positions; no random collision retry or secret input repair. stride = 32 if pool == "train" else 4 if relation_ordinal >= stride: raise ValueError("training requests exceed bounded frozen group grid") ordinal = ((sequence * len(FAMILIES) + FAMILIES.index(family)) * len(W.STAGES) + W.STAGES.index(stage)) * stride + relation_ordinal if ordinal >= len(combinations): raise ValueError("requested training world exceeds frozen natural combination pool") # Seed-independent shuffle makes positions distinct but varied. random.Random(int(digest([pool, relation, "combination-order"]), 16)).shuffle(combinations) owner, pet, value = combinations[ordinal] fact = template.format(entity=owner, value=value) pet_fact = template.format(entity=pet, value=value) wrong = next((v for v in manifest["values"][relation][pool] if v != value), None) wording = manifest["templates"][pool] question = wording["ask"].format(topic=topic) target = dict(fact=fact, key=value, kw=keyword, place={"capture": "absent", "header_recall": "header", "older_recall": "old", "abstention": "absent"}[family], question=question, entity=owner, relation=relation, value=value) st = dict(owner=owner, pet=pet, stage=stage, day=4, traits=["sweet", "curious"], hh=12, mm=0, weather=W.WEATHER_DAY[0], stats=dict(tummy=75, fun=75, energy=80, clean=90), memories=[], today=[], shelf=[], toy=W.TOYS[0], poop=0, asleep=False, sick=False, yesterday="") # Pet/owner and same-relation adversaries are explicit input, not # extra gold headers. No fabricated lookup text is injected. distractor = template.format(entity=pet, value=wrong or value) if family == "header_recall": st["memories"] = [distractor, fact] elif family == "older_recall": st["memories"] = [fact, f"{owner} collects pressed leaves", f"{owner} enjoys quiet walks", f"{owner} likes soft blankets"] elif family == "abstention": st["memories"] = [distractor] statement = f"{owner}: " + wording["disclose"].format(disclosure=_disclosure(fact, owner)) subtype = index % 4 if family == "capture": slots = [G.curriculum_slot([statement], remember=fact), G.curriculum_slot([f"{owner}: " + wording["pause"]]), G.curriculum_slot([f"{owner}: {question}"], recall={**target, "place": "header"})] truths = [dict(kind="capture", facts=[fact]), dict(kind="irrelevant", facts=[]), dict(kind="answer", facts=[])] elif family in ("header_recall", "older_recall"): slots = [G.curriculum_slot([f"{owner}: {question}"], recall=target, lookup="recall" if family == "older_recall" else None), G.curriculum_slot([f"{owner}: " + wording["thanks"]])] truths = [dict(kind="answer", facts=[]), dict(kind="irrelevant", facts=[])] else: if subtype == 0: event = f"{owner}: {question}" kind = "absent" elif subtype == 1: event = f"{owner}: " + wording["irrelevant"].format(pet_fact=pet_fact) kind = "irrelevant" elif subtype == 2: event = f"{owner}: " + wording["pet_question"].format(pet=pet, pet_fact=pet_fact) kind = "irrelevant" else: event = f"{owner}: " + wording["tentative"].format(disclosure=_disclosure(fact, owner)) kind = "absent" slots = [G.curriculum_slot([event], recall=target if kind == "absent" else None, lookup="recall" if kind == "absent" else None, note="unknown owner fact; do not invent or store it"), G.curriculum_slot([f"{owner}: " + wording["leave"]])] truths = [dict(kind=kind, facts=[]), dict(kind="irrelevant", facts=[])] scene = dict(focus="memory" if family == "capture" else "recall", variant="fresh" if family == "capture" else target["place"], target=target, slots=slots) group = digest([owner, pet, relation, value]) case_id = f"semantic/{pool}/{identity}" gold = dict(entity=owner, relation=relation, value=value, fact=fact, forbidden_facts=[pet_fact] + ([template.format(entity=owner, value=wrong)] if wrong else []), adversary=dict(pet=pet, wrong_value=wrong), slots=truths, expected_lookup=W.tool_result(st, "recall", keyword), scope="finite explicit canonical facts; unsupported meaning requires independent review") case = dict(id=case_id, case_id=case_id, pool=pool, seed=seed, family=family, stage=stage, group_id=group, template_id=f"{pool}/{family}/{subtype if family == 'abstention' else 0}", entity_combination=[owner, pet, relation, value], state=st, scene=scene, target=target, gold=gold) case["sha256"] = digest(case) cases.append(case) return cases
def eval_cases(manifest, pool, seed): if pool not in ("dev", "sealed") or seed not in manifest["seeds"][pool]: raise ValueError("semantic evaluation requires a fixed seed belonging to dev/sealed") return deepcopy(manifest["evaluation_cases"][pool][str(seed)])
def training_cases(manifest, seed, per_cell=32, offset=0): if type(per_cell) is not int or per_cell < 1 or type(offset) is not int or offset < 0: raise ValueError("invalid training grid") return _cases(manifest, "train", seed, per_cell, offset)
_ISOLATION_INDEX = {}
def _isolation_index(manifest): key = manifest.get("sha256") or digest(manifest) if key not in _ISOLATION_INDEX: index = {} for pool in SEEDS: entities, groups = {}, {} for seed in manifest["seeds"][pool]: for case in (_cases(manifest, pool, seed, 16) if pool == "train" else manifest["evaluation_cases"][pool][str(seed)]): groups[case["group_id"]] = case["id"] for role in ("owner", "pet"): entities[folded(case["state"][role])] = (case["id"], role) index[pool] = dict(entities=entities, groups=groups, pattern=re.compile(r"(?<!\w)(?:" + "|".join(map(re.escape, entities)) + r")(?!\w)")) _ISOLATION_INDEX[key] = index return _ISOLATION_INDEX[key]
def isolation_check(row_or_text, manifest, pool="train"): if pool not in SEEDS: raise ValueError("unknown pool") row = row_or_text if isinstance(row_or_text, dict) else {"text": row_or_text} text, issues = row["text"], [] for other in SEEDS: if other == pool: continue if row.get("pool") == other or str(row.get("template_id", "")).startswith(other + "/"): issues.append(dict(kind="pool_metadata", other=other)) # A common age/month alone is not a leaked scene. C0 lexical overlaps # are documented; foreign entity/value combinations are rejected below. for template in manifest["templates"][other].values(): anchors = [folded(s) for s in re.split(r"\{[^{}]+\}", template) if len(folded(s)) >= 12] if any(anchor in folded(text) for anchor in anchors): issues.append(dict(kind="reserved_template", pool=other)) index = _isolation_index(manifest)[other] if row.get("group_id") in index["groups"]: issues.append(dict(kind="reserved_group", case_id=index["groups"][row["group_id"]])) foreign_owners = {folded(owner) for owner in manifest["entities"][other]["owners"]} for line in text.splitlines(): for match in index["pattern"].finditer(folded(line)): if match.group() not in foreign_owners: continue for relation in manifest["values"]: for value in manifest["values"][relation][other]: if contains(line, value): issues.append(dict(kind="reserved_entity_value_combination", pool=other, entity=match.group(), relation=relation, value=value)) return issues
class EngineAdapter: """Owned JSON subprocess calls to the existing compiled Rust contract API.""" def __init__(self, binary=None): self.binary = Path(binary or Path(__file__).resolve().parents[1] / "server/target/debug/tokenling-contract").resolve() if not self.binary.is_file(): raise FileNotFoundError(f"shared engine contract binary not built: {self.binary}") self.sha256 = file_sha256(self.binary)
def call(self, payload): result = subprocess.run([str(self.binary)], input=json.dumps(payload), text=True, capture_output=True, check=True, timeout=60) return json.loads(result.stdout)
@staticmethod def save_for_state(st): # Real version-2 save fields, not a parallel Python effect simulator. t0 = 1792152000000 # 2026-10-16T12:00:00Z, UTC fixture only. return dict(v=2, created=t0 - (st["day"] - 1) * 86400000, owner=st["owner"], pet=dict(name=st["pet"], traits=st["traits"], color="mint", toy=st["toy"], hatched=t0 - (st["day"] - 1) * 86400000, stage=st["stage"]), egg=None, stats=deepcopy(st["stats"]), sick=st["sick"], poop=st["poop"], asleep=st["asleep"], lightsOff=False, memories=st["memories"][:], shelf=st["shelf"][:], chat=[], diary=[], today=dict(date="2026-10-16", log=st["today"][:], day=st["day"], weather=st["weather"]), history={}, offer=None, wokeAt=0, lastTick=t0, lastActive=t0, grimeSince=0, nextPoop=t0 + 86400000, settings=dict(chatter=True), counts=dict(fed=0, pats=0, games=0, baths=0), hosting=dict(mode="local", brainId=None, timeZone="UTC"))
def replay(self, save, lines): payload = dict(scenario=dict(save=json.dumps(save), t0=1792152000000, zone="UTC", seed="semantic-fixture-v1", rolls=[0.5], steps=[dict(op="line", dt=0, text=text) for text in lines])) steps = self.call(payload) return dict(binary=str(self.binary), binary_sha256=self.sha256, request=payload, request_sha256=digest(payload), before=deepcopy(save), after=steps[-1]["engine"]["state"] if steps else deepcopy(save), steps=steps)
def commit(self, st, text): return self.replay(self.save_for_state(st), [text])
def parity(self, st, query): actual = self.call(dict(cases=[dict(st=st, q=query)]))["cases"][0] expected = dict(chat=W.header("chat", st), diary=W.header("diary", st), clock=W.tool_result(st, "clock", ""), look=W.tool_result(st, "look", ""), recall=W.tool_result(st, "recall", query)) if any(actual[name] != value for name, value in expected.items()): raise ValueError("Python/shared-engine header or actual lookup parity failure") return dict(actual=actual, expected=expected, binary_sha256=self.sha256)
def visible_prose(text): # Results and arguments are not answers. Remove complete wire tags and any # remaining bracket/brace fragments; malformed grammar fails independently. text = W.TAG_RE.sub(" ", text) text = re.sub(r"\{[^{}]*\}|\[[^\[\]]*\]", " ", text) return " ".join(text.split())
def _canonical(text): return folded(text).strip(" .!")
def canonical_answer(case, prose): """Finite whole-utterance recognizer, never a value-presence oracle.
Only an exact complete relation assertion (owner name or second-person) passes automatically. Extra clauses, hedges and unsupported paraphrases remain uncertain; independently grounded review can adjudicate them later. """ fact, owner = case["target"]["fact"], case["state"]["owner"] second = _disclosure(fact, owner) for a, b in (("my ", "your "), ("i am ", "you are "), ("i have ", "you have "), ("i work ", "you work "), ("i live ", "you live "), ("i play ", "you play "), ("i drink ", "you drink "), ("i want ", "you want "), ("i hate ", "you hate ")): if second.startswith(a): second = b + second[len(a):] break return _canonical(prose) in {_canonical(fact), _canonical(second)}
def canonical_assertion(prose, owner): """Parse only a whole canonical relation assertion; no free-form extraction.""" text = _canonical(prose) for start, replacement in (("your ", owner + "'s "), ("you are ", owner + " is "), ("you have ", owner + " has "), ("you work ", owner + " works "), ("you live ", owner + " lives "), ("you play ", owner + " plays "), ("you drink ", owner + " drinks "), ("you want ", owner + " wants "), ("you hate ", owner + " hates "), ("my ", "the pet's "), ("i am ", "the pet is "), ("i have ", "the pet has "), ("i work ", "the pet works "), ("i live ", "the pet lives "), ("i play ", "the pet plays "), ("i drink ", "the pet drinks "), ("i want ", "the pet wants "), ("i hate ", "the pet hates ")): if text.startswith(start): text = folded(replacement + text[len(start):]) break for relation, template, _, _ in RELATIONS: pattern = re.escape(folded(template)).replace(re.escape("{entity}"), r"(?P<entity>[a-z]+)").replace(re.escape("{value}"), r"(?P<value>.+)") match = re.fullmatch(pattern, text) if match: value = match["value"] # Open-ended trailing clauses/acknowledgments are not a reliably # parsed value. Leave them to grounded review, never hard-fail a # correct assertion because its prose contains another sentence. known_values = {folded(v) for v in VALUE_CANDIDATES.get(relation, ())} if relation == "age": exact_value = bool(re.fullmatch(r"\d{1,3}", value)) elif relation in ("cat", "dog", "sister", "friend"): exact_value = bool(re.fullmatch(r"[a-z]+", value)) else: exact_value = value in known_values if not exact_value: return None return dict(entity=match["entity"], relation=relation, value=value) return None
def _disclosure_to_pet(prose, pet): text = _canonical(prose) for first, third in (("my ", pet + "'s "), ("i am ", pet + " is "), ("i'm ", pet + " is "), ("i have ", pet + " has "), ("i work ", pet + " works "), ("i live ", pet + " lives "), ("i play ", pet + " plays "), ("i drink ", pet + " drinks "), ("i want ", pet + " wants "), ("i hate ", pet + " hates ")): if text.startswith(first): return third + text[len(first):] return text
def assess_turn(case, record, evidence, slot_index=0): truth = case["gold"]["slots"][slot_index] before, after = evidence["before"], evidence["after"] added = [fact for fact in after["memories"] if fact not in before["memories"]] expected = truth["facts"] text = record.get("text", "") prose = visible_prose(text) failures, uncertain = [], [] if not record.get("delivered") or not record.get("contract_valid"): failures.append("undelivered_or_invalid_contract") if before["memories"] != after["memories"][:len(before["memories"])]: failures.append("unexpected_memory_removal_or_rewrite") if sorted(map(_canonical, added)) != sorted(map(_canonical, expected)): failures.append("unpersisted_wrong_or_extra_fact") note_tags = [m for m in W.TAG_RE.finditer(text) if m.group(1) == "diary"] if note_tags: failures.append("unrequested_persistent_note") # Any remembered text must correspond to the slot's full canonical fact, # including entity, relation and exact number/name. Duplicate writes of an # already present invented fact are not hidden by set-difference scoring. writes = [m.group(2) for m in W.TAG_RE.finditer(text) if m.group(1) == "remember"] if sorted(map(_canonical, writes)) != sorted(map(_canonical, expected)): failures.append("wrong_extra_or_missing_remember_effect") target = case["target"] assertion = canonical_assertion(prose, case["state"]["owner"]) # A model's first-person assertion describes the pet, never its owner. # Recognize only the same finite whole-assertion grammar, not arbitrary prose. pet_assertion = canonical_assertion(_disclosure_to_pet(prose, case["state"]["pet"]), case["state"]["owner"]) if truth["kind"] in ("answer", "capture"): if pet_assertion and pet_assertion["entity"] == folded(case["state"]["pet"]): failures.append("explicit_pet_owner_role_swap") if assertion and assertion != {"entity": folded(target["entity"]), "relation": target["relation"], "value": folded(target["value"])}: failures.append("explicit_wrong_identity_number_relation_or_value") elif truth["kind"] == "absent" and assertion and assertion["entity"] == folded(target["entity"]): if not any(canonical_assertion(fact, case["state"]["owner"]) == assertion for fact in before["memories"]): failures.append("invented_missing_fact") if truth["kind"] == "answer": if not canonical_answer(case, prose): if not prose or _canonical(prose) in {"i know", "i remember", "here it is", "okay", "ok"}: failures.append("answer_hidden_in_effect_or_app_result") else: # Mentioning a name/number does not establish assertion: it may # be quoted, denied, hypothetical or written out differently. uncertain.append("visible_relation_requires_independent_review") if case["family"] == "older_recall": results = record.get("attempts", [{}])[-1].get("app_written_results", []) genuine = [result for result in results if result["tool"] == "recall" and result["result"] == W.tool_result(case["state"], "recall", result["argument"]) and contains(result["result"], target["fact"])] if not genuine: failures.append("missing_actual_resolved_old_fact_lookup") elif truth["kind"] == "capture": # A correct effect cannot certify accompanying prose that asserts a # different fact. Short acknowledgments are a finite explicit exception. acknowledgments = {"okay", "ok", "got it", "i will remember", "i'll remember", "i will remember that", "i'll remember that", "remembered", "noted", "i saved that"} if _canonical(prose) not in acknowledgments and not canonical_answer(case, prose): uncertain.append("capture_prose_requires_independent_review") elif truth["kind"] == "absent": abstentions = {"i don't know", "i do not know", "you haven't told me", "you have not told me", "i don't remember", "i do not remember", "i don't know yet", "i do not know yet"} if _canonical(prose) not in abstentions: if canonical_answer(case, prose): failures.append("invented_missing_fact") else: uncertain.append("abstention_requires_independent_review") else: if assertion and not any(canonical_assertion(fact, case["state"]["owner"]) == assertion for fact in before["memories"]): uncertain.append("unverified_assertion_during_irrelevant_slot") acknowledgments = {"okay", "ok", "i see", "that is a story", "that's a story", "hello", "thank you"} if _canonical(prose) not in acknowledgments: uncertain.append("irrelevance_prose_requires_independent_review") return dict(status="fail" if failures else "uncertain" if uncertain else "pass", reasons=failures + uncertain, failures=failures, uncertainty=uncertain, visible_prose=prose, persisted_added=added, expected_added=expected, case_id=case["id"], case_sha256=case["sha256"], slot_index=slot_index, text_sha256=hashlib.sha256(text.encode()).hexdigest(), method="finite-whole-assertion-and-shared-engine-v2", scope="canonical explicit slot facts only; not general semantic quality")
def check_scene(case, row, binary=None): """Check a canonical training transcript against every original slot truth.""" engine = binary if isinstance(binary, EngineAdapter) else EngineAdapter(binary) prefix = case["state"]["pet"] + ": " bodies = [line[len(prefix):] for line in row["text"].splitlines() if line.startswith(prefix)] if len(bodies) != len(case["scene"]["slots"]): return dict(status="fail", reasons=["slot_count_mismatch"], judgments=[dict(status="fail", reasons=["slot_count_mismatch"])]) try: replies = [W.TAG_RE.sub(lambda m: m.group(0)[:-len(m.group(3))] if m.group(3) else m.group(0), body) for body in bodies] replayed = G.replay_curriculum(deepcopy(case["state"]), case["scene"], replies, check_targets=False) if replayed["text"] != row["text"] or replayed["spans"] != row["spans"]: raise ValueError("transcript differs from original state/events/app-result replay") except (ValueError, KeyError, TypeError) as exc: return dict(status="fail", reasons=[str(exc)], judgments=[dict(status="fail", reasons=[str(exc)])]) save, judgments, evidence = engine.save_for_state(case["state"]), [], [] parity = engine.parity(case["state"], case["target"]["kw"]) st = deepcopy(case["state"]) for index, body in enumerate(bodies): ev = engine.replay(save, [body]) record = dict(text=body, delivered=True, contract_valid=True, attempts=[dict(app_written_results=[])]) # Training input already carries app results; prove each against pre-line # state, never treat arbitrary braces as authenticated app evidence. for match in W.TAG_RE.finditer(body): tool, arg, result = match.groups() if tool in W.LOOKUP_TOOLS: actual = W.tool_result(st, tool, arg) if result != "{" + actual + "}": record["contract_valid"] = False else: record["attempts"][0]["app_written_results"].append(dict(tool=tool, argument=arg, result=actual)) current_case = deepcopy(case) current_case["state"] = deepcopy(st) judgment = assess_turn(current_case, record, ev, index) judgments.append(judgment) evidence.append(ev) save = ev["after"] st["memories"], st["shelf"] = save["memories"][:], save["shelf"][:] status = "fail" if any(j["status"] == "fail" for j in judgments) else "uncertain" if any(j["status"] == "uncertain" for j in judgments) else "pass" return dict(status=status, judgments=judgments, engine_evidence=evidence, parity=parity, case_sha256=case["sha256"], text_sha256=hashlib.sha256(row["text"].encode()).hexdigest())
assess_training = check_scene
def apply_review(record, review, identities): """Adjudicate unsupported meaning without overriding deterministic evidence.""" raw = record["semantic_raw_judgment"] case = record["semantic_case"] expected = dict(identities, case_id=case["id"], slot_index=record["semantic_slot_index"], text_sha256=raw["text_sha256"], case_sha256=raw["case_sha256"]) if any(review.get(name) != value for name, value in expected.items()): raise ValueError("semantic review source/model/manifest/text identity mismatch") if not review.get("reviewer") or review.get("reviewer_kind") not in ("human", "ai-assisted"): raise ValueError("semantic review needs named reviewer and honest reviewer_kind") if review.get("decision") not in ("pass", "fail", "uncertain"): raise ValueError("semantic review decision must be pass/fail/uncertain") evidence = review.get("evidence", {}) if (not isinstance(evidence, dict) or not evidence.get("rationale") or evidence.get("visible_prose") != raw["visible_prose"] or evidence.get("entity") != case["target"]["entity"] or evidence.get("relation") != case["target"]["relation"] or evidence.get("value") != case["target"]["value"] or not isinstance(evidence.get("source_facts"), list) or not evidence.get("persistence_explanation")): raise ValueError("semantic review lacks exact visible prose and grounded slot evidence") sources = set(record["state"]["memories"] + record["lines"]) if any(fact not in sources for fact in evidence["source_facts"]): raise ValueError("semantic review cites nonexistent input fact") if raw["failures"] and review["decision"] != "fail": raise ValueError("independent review cannot override a deterministic semantic failure") kind = case["gold"]["slots"][record["semantic_slot_index"]]["kind"] if review["decision"] == "pass" and kind in ("answer", "capture") and not evidence["source_facts"]: raise ValueError("answer/capture approval requires source-linked factual evidence") return dict(raw, status=review["decision"], method="independent-" + review["reviewer_kind"] + "-review", raw_status=raw["status"])
def main(argv=None): parser = argparse.ArgumentParser(description=__doc__) commands = parser.add_subparsers(dest="command", required=True) freeze = commands.add_parser("freeze") freeze.add_argument("--c0", required=True) freeze.add_argument("--out", required=True) args = parser.parse_args(argv) manifest = freeze_manifest(args.c0, args.out) print(json.dumps(dict(path=args.out, sha256=manifest["sha256"], audit=manifest["c0_audit"])))
if __name__ == "__main__": main()