Something went wrong. Try again.
a tiny pet with an even tinier brain, living in your browser
Something went wrong. Try again.
51 kB · 874 lines
Python
at main
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875"""Regression cases for the training contract. Run with python3 test/data_rules.py."""import copyimport hashlibimport jsonimport osimport randomimport reimport subprocessimport sysimport tempfileimport unittestsys.path.insert(0, os.path.join(os.path.dirname(__file__), '..', 'train'))import gen_data as Gimport world as Wimport migrate_v6 as Mimport model_data as Dimport experiments as Efrom unittest.mock import patch
class ByteOffsetTokenizer: """Deterministic byte-token offsets, including UTF-8's repeated char spans.""" eos_token_id = 0
def __init__(self, merges=()): self.merges = dict(merges)
def __call__(self, text, return_offsets_mapping=True, add_special_tokens=True): ids, offsets = [], [] at = 0 while at < len(text): end = self.merges.get(at, at + 1) raw = text[at:end].encode('utf-8') if end > at + 1: ids.append(1000 + sum(raw)) offsets.append((at, end)) else: for value in raw: ids.append(value + 1) offsets.append((at, end)) at = end if add_special_tokens: ids = [0] + ids + [0] offsets = [(0, 0)] + offsets + [(0, 0)] return dict(input_ids=ids, offset_mapping=offsets)
class Rules(unittest.TestCase): def setUp(self): self.st = dict(owner='Sam', pet='Mochi', stage='kid', day=4, traits=['sleepy', 'curious'], hh=10, mm=0, weather='sunny', stats=dict(tummy=95, fun=50, energy=10, clean=80), memories=[], today=[], shelf=[], toy='ball', poop=0, asleep=True, sick=False, yesterday='')
def check(self, raw, error=None): if error: with self.assertRaisesRegex(ValueError, error): G.check_chat(copy.deepcopy(self.st), raw) else: text, spans = G.check_chat(copy.deepcopy(self.st), raw) G.recheck(dict(kind='chat', text=text, spans=spans)) return text, spans
def test_sleep(self): self.check('Sam: goodnight\nMochi: mmm... [dream:moon]\nMochi: [dream:cookie]') for body in ['hello awake world [emote:happy]', '[sleep]', '[dream:moon] [walk:left]', 'one two three four five [dream:moon]', '[dream:moon] [dream:heart]', '[dream:invented]']: self.check('Sam: goodnight\nMochi: '+body+'\nMochi: [dream:cookie]', 'sleep dream grammar|dream invented')
def test_sleep_offer(self): offer='* Sam offered Mochi a cookie *\n' self.check(offer+'Mochi: mmm [no] [dream:cookie]\nMochi: [dream:moon]') self.check(offer+'Mochi: cookie?! [yes]\n* Mochi woke up *\nMochi: yummy', 'exhausted food wake') self.st['stats']['tummy']=10 self.check(offer+'Mochi: cookie?! [yes]\n* Mochi woke up *\nMochi: yummy [emote:happy]') self.check(offer+'Mochi: [yes]\nMochi: hello', 'missing wake') self.check(offer+'Mochi: [yes] [dream:cookie]\n* Mochi woke up *\nMochi: hello', 'other tags') self.check('* Mochi woke up *\n'+offer+'Mochi: [yes]\nMochi: hello', 'wake before offer') self.check(offer+'Mochi: [dream:cookie]\nMochi: [dream:moon]', 'offer answer')
def test_wake(self): for action in ['Sam: hey sleepyhead', '* Sam patted Mochi *']: self.check('* Mochi woke up *\n'+action+'\nMochi: hi!\nMochi: [emote:wave]') self.check(action+'\n* Mochi woke up *\nMochi: hi!\nMochi: [emote:wave]', 'wake after message/pat') self.check('* Sam turned on the lights *\n* Mochi woke up *\nMochi: bright!\nMochi: [emote:wave]') self.st['asleep']=False self.check('Sam: hi\nMochi: [dream:moon]\nMochi: hello', 'awake dream') self.check('Sam: night\nMochi: [sleep]\n* Mochi woke up *\nMochi: hi', 'instant wake') self.check('Sam: night\nMochi: [sleep]\n* Mochi woke up *\nSam: hello\nMochi: hi')
def test_memory_and_hunger(self): self.st['asleep']=False for body, err in [('[remember:bug on window]', 'memory without owner'), ('[remember:Sam lost his hat]', 'gendered'), ('[recall:Sam has cat]{nothing}', 'long recall'), ("i am literally starving!", 'hunger while full')]: self.check('Sam: hello\nMochi: '+body+'\nMochi: [emote:happy]', err) self.check('Sam: hello\nMochi: i am not hungry [no]\nMochi: okay', 'answer without offer') self.check('Sam: hello\nMochi: i am not hungry\nMochi: okay') self.check('Sam: ...\nMochi: hello\nMochi: okay', 'owner without words') self.st['memories']=['Sam is 29 years old'] text, spans=self.check('Sam: how old am i?\nMochi: [recall:old]{Sam is 29 years old} 29!\nSam: thanks\nMochi: hehe') start=text.index('{Sam'); end=text.index('}',start)+1 self.assertTrue(all(b<=start or a>=end for a,b in spans))
def test_lookups_see_only_committed_effects(self): self.st.update(asleep=False, shelf=['star'], poop=2) raw = ('Mochi: [gift:star] [look]\n' 'Sam: what is on the shelf?\n' 'Mochi: [look] two stars\n' '* Sam cleaned up the poop *\n' 'Mochi: [look] all clean now') text, spans = self.check(raw) results = re.findall(r'\[look\]\{([^}]+)\}', text) self.assertEqual(results, ['shelf: star · toy: ball · window: sunny · poop x2', 'shelf: star, star · toy: ball · window: sunny · poop x2', 'shelf: star, star · toy: ball · window: sunny']) self.assertEqual(self.st['shelf'], ['star']) self.assertEqual(self.st['poop'], 2) for result in results: a = text.index('{' + result + '}') b = a + len(result) + 2 self.assertTrue(all(y <= a or x >= b for x, y in spans))
def test_recall_waits_for_current_line_commit(self): self.st.update(asleep=False, memories=[]) text, _ = self.check('Sam: I love frogs\n' 'Mochi: [remember:Sam loves frogs] [recall:frogs]\n' 'Sam: can you remember now?\n' 'Mochi: [recall:frogs] frogs!') self.assertEqual(re.findall(r'\[recall:frogs\]\{([^}]+)\}', text), ['nothing', 'Sam loves frogs']) self.assertEqual(self.st['memories'], [])
def test_medicine(self): self.st.update(asleep=False, sick=False) body='* Sam offered Mochi medicine *\nMochi: okay [yes]\nMochi: thank you Sam' self.check(body, 'medicine while healthy') self.check(body.replace('okay [yes]', 'no thanks [no]')) self.st['sick']=True self.check(body)
def test_headers_diary(self): self.st['today']=['ate a cookie', 'Sam said: goodnight']*4 self.assertNotIn('Sam said:', W.header('chat',self.st)) self.assertIn('Sam said:', W.header('diary',self.st)) for body in ['dear diary,\n[', 'dear diary,\ndoodle:star', 'dear diary,\n- PET', 'dear diary,\n[doodle:banana]']: with self.assertRaises(ValueError): G.check_diary_body(body) G.check_diary_body('dear diary,\ni dreamed of cookies.\n- Mochi\n[doodle:star]') self.st.update(hh=22, weather='sunny') text, spans = G.check_chat(self.st, 'Sam: goodnight little one\nMochi: zzz [dream:moon]\nMochi: [dream:star]') with self.assertRaisesRegex(ValueError, 'weather for time'): G.recheck(dict(kind='chat', text=text, spans=spans))
def stored_rows(self): st = copy.deepcopy(self.st) st['asleep'] = False rows = [] for kind, raw in [('chat', 'Sam: hello\n* Sam patted Mochi *\n' 'Mochi: [clock] hello [emote:happy]\nMochi: hi [emote:wave]'), ('diary', 'dear diary,\ni dreamed of cookies.\n- Mochi\n[doodle:star]')]: text, spans = (G.check_chat if kind == 'chat' else G.check_diary)(st, raw) rows.append(dict(kind=kind, text=text, spans=spans)) return rows
def test_recheck_masks(self): for row in self.stored_rows(): with self.subTest(kind=row['kind'], mask='canonical'): original = copy.deepcopy(row) G.recheck(row) self.assertEqual(row, original) missing = {k: v for k, v in row.items() if k != 'spans'} shortened = copy.deepcopy(row['spans']) shortened[0][0] += 1 cases = [('missing', missing), ('empty', {**row, 'spans': []}), ('all text', {**row, 'spans': [[0, len(row['text'])]]}), ('missing trained character', {**row, 'spans': shortened})] for label, bad in cases: with self.subTest(kind=row['kind'], mask=label): original = copy.deepcopy(bad) with self.assertRaises(ValueError): G.recheck(bad) self.assertEqual(bad, original)
def test_recheck_untrained_characters(self): for row in self.stored_rows(): text = row['text'] positions = [('header', 0), ('separator', text.index('---\n'))] if row['kind'] == 'chat': positions += [('owner', text.index('Sam: hello')), ('event', text.index('* Sam patted Mochi *')), ('pet prefix', text.index('Mochi: [clock]')), ('lookup brace', text.index('{10:00 morning}')), ('lookup result', text.index('10:00 morning}', text.index('---\n')))] else: positions.append(('diary opener', text.index('dear diary,'))) for label, at in positions: with self.subTest(kind=row['kind'], character=label): bad = {**row, 'spans': row['spans'] + [[at, at + 1]]} with self.assertRaisesRegex(ValueError, 'loss mask'): G.recheck(bad)
def test_recheck_bad_spans(self): for row in self.stored_rows(): n = len(row['text']) cases = [('null', None), ('number', 1), ('boolean', True), ('string', '0,1'), ('object', {}), ('null span', [None]), ('short span', [[0]]), ('long span', [[0, 1, 2]]), ('string bound', [['0', 1]]), ('float bound', [[0, 1.5]]), ('boolean start', [[False, 1]]), ('boolean end', [[0, True]]), ('negative bound', [[-1, 1]]), ('reversed bounds', [[1, 0]]), ('past end', [[n, n + 1]]), ('past end empty', [[n + 1, n + 1]])] for label, spans in cases: with self.subTest(kind=row['kind'], spans=label): with self.assertRaises(ValueError): G.recheck({**row, 'spans': spans})
def test_recheck_equivalent_masks(self): for row in self.stored_rows(): merged = [] for a, b in row['spans']: if merged and a == merged[-1][1]: merged[-1][1] = b else: merged.append([a, b]) split = [] for a, b in merged: split.extend([[a, a + 1], [a + 1, b]] if b - a > 1 else [[a, b]]) for label, spans in [('merged', merged), ('split reversed', list(reversed(split))), ('overlapping', split + [split[0]])]: with self.subTest(kind=row['kind'], spans=label): equivalent = {**row, 'spans': spans} original = copy.deepcopy(equivalent) G.recheck(equivalent) self.assertEqual(equivalent, original)
def test_recheck_shifted_text(self): chat, diary = self.stored_rows() at = chat['text'].index('Mochi: [clock]') indented = chat['text'][:at] + ' ' + chat['text'][at:] shifted = [[a + 2, b + 2] for a, b in chat['spans']] cases = [('indented, old spans', {**chat, 'text': indented}), ('indented, shifted spans', {**chat, 'text': indented, 'spans': shifted}), ('chat blank line', {**chat, 'text': chat['text'].replace('Sam: hello\n', 'Sam: hello\n\n')}), ('stage directions', {**chat, 'text': chat['text'].replace('Mochi: [clock]', 'Mochi: *purrs* [clock]')}), ('chat trailing spaces', {**chat, 'text': chat['text'].replace('Sam: hello\n', 'Sam: hello \n')}), ('diary opener case', {**diary, 'text': diary['text'].replace('dear diary,', 'Dear diary,')}), ('diary blank line', {**diary, 'text': diary['text'].replace('dear diary,\n', 'dear diary,\n\n')}), ('diary trailing newline', {**diary, 'text': diary['text'] + '\n'})] for label, bad in cases: with self.subTest(text=label): original = copy.deepcopy(bad) with self.assertRaisesRegex(ValueError, 'noncanonical text'): G.recheck(bad) self.assertEqual(bad, original)
def test_recheck_keeps_lookup_results(self): self.st['asleep'] = False text, spans = G.check_body('Sam', 'Mochi', W.header('chat', self.st), 'Sam: hello\nMochi: [clock]{23:59 night} late!\nMochi: hello', None) row = dict(kind='chat', text=text, spans=spans) original = copy.deepcopy(row) G.recheck(row) self.assertEqual(row, original) self.assertIn('[clock]{23:59 night}', row['text'])
def test_encode_lookup_boundary_crossings(self): row = self.stored_rows()[0] original = copy.deepcopy(row) start = row['text'].index('{10:00 morning}') end = start + len('{10:00 morning}') # Real BPEs can contain ']{' or '} hello' punctuation/word merges. for label, span in [('call/result', (start - 1, start + 1)), ('result/continuation', (end - 1, end + 2))]: for cap in [2, len(row['text']) * 4]: with self.subTest(boundary=label, cap=cap): with self.assertRaisesRegex(D.MaskBoundaryError, 'unsafe loss boundary') as caught: D.encode(ByteOffsetTokenizer([span]), row, cap) issue = caught.exception.issues[0] self.assertGreater(issue['trained_characters'], 0) self.assertGreater(issue['untrained_characters'], 0) self.assertEqual(issue['boundary_positions'], [start if label == 'call/result' else end]) self.assertEqual(row, original) item = D.inspect_row(ByteOffsetTokenizer([(start - 1, start + 1)]), row) crossing = item['mixed_tokens'][0]['token_index'] self.assertEqual(item['labels'][crossing], -100)
def test_encode_generated_separator_and_untrained_colon_boundary(self): row = self.stored_rows()[0] at = row['text'].index('Mochi: hi') + len('Mochi:') ids, labels, offsets = D.tokenize_row(ByteOffsetTokenizer([(at, at + 3)]), row) separator = next(i for i, (a, b) in enumerate(offsets) if a == at) self.assertEqual(labels[separator], ids[separator]) with self.assertRaisesRegex(D.MaskBoundaryError, 'trained and .*untrained'): D.encode(ByteOffsetTokenizer([(at - 1, at + 1)]), row, len(row['text']) * 4) for token_id, label, (a, b) in zip(ids, labels, offsets): if row['text'][a:b] == 'M' and row['text'][a:a + 6] == 'Mochi:': self.assertEqual(label, -100)
def test_continuation_mask_migration_preserves_text_and_rejects_other_mask_damage(self): row = self.stored_rows()[0] coverage = {at for a, b in row['spans'] for at in range(a, b)} spaces = {match.end() for match in re.finditer(r'(?m)^Mochi:', row['text'])} legacy = {**row, 'spans': [[at, at + 1] for at in sorted(coverage - spaces)]} with self.assertRaisesRegex(ValueError, 'loss mask mismatch'): G.recheck(legacy) fixed = M.continuation_masks(legacy) self.assertEqual(fixed['text'], row['text']) self.assertEqual(G._span_coverage(fixed['spans'], len(fixed['text'])), G._span_coverage(row['spans'], len(row['text']))) G.recheck(fixed) damaged = {**legacy, 'spans': legacy['spans'][1:]} with self.assertRaisesRegex(ValueError, 'corrupt masks'): M.continuation_masks(damaged)
def test_encode_unicode_offsets_and_masked_results(self): self.st['asleep'] = False text, spans = G.check_body('Sam', 'Mochi', W.header('chat', self.st), 'Sam: café🙂\nMochi: [look]{shelf: 漢字🙂 e\u0301} hé🙂!\nMochi: coucou', None) row = dict(kind='chat', text=text, spans=spans) G.recheck(row) ids, labels, offsets = D.tokenize_row(ByteOffsetTokenizer(), row) self.assertGreater(len(ids), len(text)) trained = {at for a, b in spans for at in range(a, b)} for token_id, label, (a, b) in zip(ids, labels, offsets): self.assertEqual(label, token_id if set(range(a, b)) <= trained else -100) emoji = text.index('🙂', text.index('{shelf')) emoji_tokens = [label for label, (a, b) in zip(labels, offsets) if (a, b) == (emoji, emoji + 1)] self.assertEqual(emoji_tokens, [-100] * len('🙂'.encode('utf-8'))) spoken_emoji = text.index('🙂', text.index(' hé')) spoken_tokens = [label for label, (a, b) in zip(labels, offsets) if (a, b) == (spoken_emoji, spoken_emoji + 1)] self.assertEqual(len(spoken_tokens), len('🙂'.encode('utf-8'))) self.assertTrue(all(label != -100 for label in spoken_tokens)) accent = text.index('e\u0301', text.index('{shelf')) self.assertTrue(all(label == -100 for label, (a, b) in zip(labels, offsets) if accent <= a < accent + 2))
def test_language_data_rejects_pet_masks(self): pet = self.stored_rows()[0] for label, row in [('canonical pet', pet), ('mask without kind', {k: v for k, v in pet.items() if k != 'kind'}), ('kind without mask', {k: v for k, v in pet.items() if k != 'spans'})]: with self.subTest(record=label), tempfile.TemporaryDirectory() as directory: path = os.path.join(directory, 'input.jsonl') with open(path, 'w', encoding='utf-8') as f: f.write(json.dumps(row, ensure_ascii=False) + '\n') with self.assertRaisesRegex(ValueError, 'pet kind/spans are forbidden'): D.load_language_rows(path)
def test_encode_diary_eos(self): row = self.stored_rows()[1] tok = ByteOffsetTokenizer() ids, labels, offsets = D.tokenize_row(tok, row) self.assertEqual((ids[-1], labels[-1], offsets[-1]), (tok.eos_token_id, tok.eos_token_id, (len(row['text']), len(row['text'])))) opener_end = row['text'].index('dear diary,') + len('dear diary,') self.assertTrue(all(label == -100 for label, (a, b) in zip(labels[:-1], offsets[:-1]) if b <= opener_end)) self.assertTrue(all(label != -100 for label, (a, b) in zip(labels[:-1], offsets[:-1]) if a >= opener_end)) truncated_ids, truncated_labels = D.encode(tok, row, len(ids) - 1) self.assertNotIn(tok.eos_token_id, truncated_ids) self.assertNotIn(tok.eos_token_id, truncated_labels)
def test_recall_templates(self): for _ in range(500): st=G.rand_state() G.recall_target(st) keys=[G.memory_template(m, st['owner']) for m in st['memories']] self.assertNotIn(None, keys) self.assertEqual(len(keys), len(set(keys)))
def test_sleep_teacher_standin(self): self.st['stats']['tummy']=10 slots = [([], 'DREAM'), ([], 'DREAM'), (['* Sam offered Mochi a cookie *'], 'OFFER')] with patch.object(G, 'sleep_scenario', return_value=slots): row = G.make_sleep_chat(lambda *a, **k: '["zzz [dream:moon]", "[dream:star]", "cookie? [yes]"]', self.st) self.assertTrue(row['text'].endswith('Mochi: cookie? [yes]\n* Mochi woke up *\n')) G.recheck(row) wake = row['text'].rindex('* Mochi woke up *') self.assertTrue(all(b <= wake for a, b in row['spans']))
def curriculum_fixture(self, focus, variant): st = copy.deepcopy(self.st) return st, G.curriculum_scenario(st, focus, variant, random.Random(8721))
def curriculum_teacher(self, replies): def teacher(prompt, **kwargs): return json.dumps(replies) return teacher
def test_curriculum_grammar_rejects_invented_and_sleep_invalid_tags(self): awake = re.compile(G.curriculum_reply_pattern('AWAKE')) for text in ('hi [emote:happy]', '[remember:Sam likes frogs]', '[recall:cat] Bo!'): self.assertIsNotNone(awake.fullmatch(text)) for text in ('[happy]', '[emote:curious]', '[dream:moon]', '[yes]', '[sleep]', '[look]{room}'): self.assertIsNone(awake.fullmatch(text)) dreaming = re.compile(G.curriculum_reply_pattern('DREAM')) self.assertIsNotNone(dreaming.fullmatch('mmm soft little moon [dream:moon]')) for text in ('one two three four five [dream:moon]', '[dream:moon] [dream:star]', '[emote:sleepy]'): self.assertIsNone(dreaming.fullmatch(text)) offer = re.compile(G.curriculum_reply_pattern('OFFER', True)) for text in ('[dream:cookie] [no]', 'mmm cookie [yes]'): self.assertIsNotNone(offer.fullmatch(text)) for text in ('[no]', '[dream:cookie] [yes]', '[yes] [emote:happy]'): self.assertIsNone(offer.fullmatch(text))
def test_curriculum_sleeping_offer_decision(self): for accept in (False, True): with self.subTest(accept=accept): st, scene = self.curriculum_fixture('care', 'food-starving-asleep') replies = (['zzz [dream:cloud]', 'mmm [yes]', 'hello Sam', '[emote:happy]'] if accept else ['zzz [dream:cloud]', 'mmm [dream:cookie] [no]', 'zzz [dream:moon]', '[dream:star]']) row = G.make_curriculum_chat(self.curriculum_teacher(replies), st, scene) G.recheck(row) wake = '* Mochi woke up *\n' self.assertEqual(row['text'].count(wake), int(accept)) if accept: self.assertIn('Mochi: mmm [yes]\n' + wake + 'Mochi: hello Sam', row['text']) at = row['text'].index(wake) self.assertTrue(all(b <= at or a >= at + len(wake) for a, b in row['spans'])) for slot in scene['slots']: for event in slot['events']: at = row['text'].index(event) self.assertTrue(all(b <= at or a >= at + len(event) for a, b in row['spans']))
def test_curriculum_catch_event_depends_on_answer(self): for answer in ('yes', 'no'): with self.subTest(answer=answer): st, scene = self.curriculum_fixture('care', 'catch-bored-awake') replies = ['hello', 'okay [' + answer + ']', 'a quiet moment', '[emote:happy]'] row = G.make_curriculum_chat(self.curriculum_teacher(replies), st, scene) event = f"* Sam played catch with Mochi ({scene['slots'][1]['catch_stars']} stars) *" self.assertEqual(event in row['text'], answer == 'yes') if answer == 'yes': at = row['text'].index(event) self.assertTrue(all(b <= at or a >= at + len(event) for a, b in row['spans']))
def test_repeated_light_states_are_not_app_possible(self): self.st['asleep'] = False for state in ('on', 'off'): repeated = f'* Sam turned {state} the lights *\nMochi: hi\n* Sam turned {state} the lights *\nMochi: hello' with self.assertRaisesRegex(ValueError, 'repeated lights state'): G.check_chat(self.st, repeated) st, scene = self.curriculum_fixture('sleep', 'light') events = [event for slot in scene['slots'] for event in slot['events'] if ' the lights *' in event] self.assertGreater(len(events), 2) self.assertTrue(all(('turned on' in first) != ('turned on' in second) for first, second in zip(events, events[1:])))
def test_curriculum_sampled_wake_positions(self): for kind in ('talk', 'touch', 'light'): with self.subTest(kind=kind), patch.object(W, 'wake_chance', return_value=1): st, scene = self.curriculum_fixture('sleep', kind) slots = scene['slots'] self.assertTrue(4 <= len(slots) <= 6) first = slots[0]['events'] self.assertEqual(first[1 if kind == 'light' else 0], '* Mochi woke up *') self.assertEqual(slots[0]['mode'], 'AWAKE') row = G.make_curriculum_chat(self.curriculum_teacher(['hello'] * len(slots)), st, scene) self.assertEqual(row['text'].count('* Mochi woke up *'), 1) G.recheck(row)
def test_curriculum_memory_followup_and_lookup_mask(self): st, scene = self.curriculum_fixture('memory', 'woken') fact, key, kw = (scene['target'][name] for name in ('fact', 'key', 'kw')) replies = [f'saved [remember:{fact}]', 'mmhm', f'[recall:{kw}] {key}!', '[emote:happy]'] row = G.make_curriculum_chat(self.curriculum_teacher(replies), st, scene) self.assertIn(f'* Mochi woke up *\nSam: ', row['text']) self.assertIn(f'[recall:{kw}]{{{fact}}} {key}!', row['text']) at = row['text'].index('{' + fact + '}') self.assertTrue(all(b <= at or a >= at + len(fact) + 2 for a, b in row['spans'])) bad = replies.copy() bad[0] = 'i will remember' with self.assertRaisesRegex(ValueError, 'memory fact'): G.make_curriculum_chat(self.curriculum_teacher(bad), st, scene) bad = replies.copy() bad[2] = f'[recall:{kw}] i remember' with self.assertRaisesRegex(ValueError, 'answer missing'): G.make_curriculum_chat(self.curriculum_teacher(bad), st, scene)
def test_curriculum_recall_requires_visible_answer(self): for place in ('old', 'header', 'absent'): with self.subTest(place=place): st, scene = self.curriculum_fixture('recall', place) target = scene['target'] key, kw, fact = (target[name] for name in ('key', 'kw', 'fact')) if place == 'absent': replies = [f'[recall:{kw}] i do not remember; can you tell me?', 'mmhm', f'saved [remember:{fact}]', key + '!'] else: replies = [(f'[recall:{kw}] ' if place == 'old' else '') + key + '!', 'mmhm', '[emote:happy]', 'thank you'] row = G.make_curriculum_chat(self.curriculum_teacher(replies), st, scene) G.recheck(row) bad = replies.copy() bad[0] = f'[recall:{kw}] {key}!' if place == 'absent' else f'[recall:{kw}] i remember' with self.assertRaises(ValueError): G.make_curriculum_chat(self.curriculum_teacher(bad), st, scene)
def test_recall_argument_and_result_are_not_visible_answers(self): target = dict(place='old', key='piano', kw='play') for reply in ('[recall:piano]{Sam plays piano}', '[recall:play]{Sam plays piano} [remember:Sam plays piano]'): with self.assertRaisesRegex(ValueError, 'answer missing'): G.check_recall(target, 'Mochi: ' + reply, 'Mochi') G.check_recall(target, 'Mochi: [recall:play]{Sam plays piano} you play piano!', 'Mochi') G.check_recall({**target, 'place':'absent'}, 'Mochi: [recall:piano]{nothing} i do not know\n' 'Mochi: saved [remember:Sam plays piano]\nMochi: piano!', 'Mochi')
def test_curriculum_lookups_are_app_written_untrained(self): st, scene = self.curriculum_fixture('lookup', 'no-diary') replies = ['[' + slot['lookup'] + '] a little answer' for slot in scene['slots'][:3]] + ['[emote:happy]'] row = G.make_curriculum_chat(self.curriculum_teacher(replies), st, scene) ids, labels, offsets = D.tokenize_row(ByteOffsetTokenizer(), row) for slot in scene['slots'][:3]: tool = slot['lookup'] result = '{' + W.tool_result(st, tool, None) + '}' call = '[' + tool + ']' at = row['text'].index(call + result) result_at = at + len(call) end = result_at + len(result) self.assertTrue(all(b <= result_at or a >= end for a, b in row['spans'])) for token, label, (a, b) in zip(ids, labels, offsets): if result_at <= a < end: self.assertEqual(label, -100) elif at <= a < result_at or end <= a < end + len(' a little answer'): self.assertEqual(label, token) G.recheck(row) bad = replies.copy() bad[0] += '{invented}' with self.assertRaisesRegex(ValueError, 'noncanonical reply'): G.make_curriculum_chat(self.curriculum_teacher(bad), st, scene)
def test_curriculum_rejects_invalid_without_repair(self): st, scene = self.curriculum_fixture('general', 'greeting') for invalid in ('[emote:surprised]', '*purrs* hello', 'Mochi: hello', '[invented]'): replies = [invalid, 'hello', '[emote:happy]', 'okay'] raw = json.dumps(replies) row, error, receipt = G.run_generation( lambda *args, **kwargs: raw, lambda teacher: G.make_curriculum_chat(teacher, st, scene), {'scenario_seed': 19}) self.assertIsNone(row) self.assertIsNotNone(error) self.assertFalse(receipt['accepted']) self.assertEqual(receipt['requests'][0]['raw'], raw) self.assertEqual(receipt['scenario_seed'], 19)
def test_corrected_continuation_replays_later_lookup_from_initial_state(self): st = copy.deepcopy(self.st) st.update(asleep=False, shelf=[]) scene = {'focus': 'general', 'variant': 'correction', 'slots': [ G.curriculum_slot(['Sam: choose a little present']), G.curriculum_slot(['Sam: look at the shelf'], lookup='look')]} original = G.replay_curriculum(st, scene, ['a flower! [gift:flower]', '[look] a flower']) corrected = G.replay_curriculum(st, scene, ['a shell! [gift:shell]', '[look] a shell']) self.assertIn('shelf: flower', original['text']) self.assertIn('shelf: shell', corrected['text']) self.assertNotIn('shelf: flower', corrected['text']) self.assertEqual(st['shelf'], []) producer = {'state': st, 'scenario': scene, 'requests': [ {'raw': json.dumps(['a flower! [gift:flower]', '[look] a flower'])}]} E.recover_producer(original, producer) with self.assertRaisesRegex(ValueError, 'replay text'): E.recover_producer(corrected, producer)
def test_corrected_offer_recomputes_app_events_not_stale_student_events(self): st, scene = self.curriculum_fixture('care', 'catch-bored-awake') replies = ['hello'] * len(scene['slots']) offer = next(i for i, slot in enumerate(scene['slots']) if slot['mode'] == 'OFFER') replies[offer] = 'let us play [yes]' accepted = G.replay_curriculum(st, scene, replies) replies[offer] = 'not now [no]' declined = G.replay_curriculum(st, scene, replies) event = f"* {st['owner']} played catch with {st['pet']} ({scene['slots'][offer]['catch_stars']} stars) *" self.assertIn(event, accepted['text']) self.assertNotIn(event, declined['text'])
def test_audit_rejects_wrong_snapshot_and_does_not_repair_masks(self): st, scene = self.curriculum_fixture('general', 'greeting') replies = ['hello', 'thank you', '[emote:happy]', 'a gentle day'] row = G.replay_curriculum(st, scene, replies) producer = {'state': st, 'scenario': scene, 'requests': [{'raw': json.dumps(replies)}]} E.recover_producer(row, producer) altered = copy.deepcopy(producer) altered['state']['owner'] = 'DifferentOwner' with self.assertRaisesRegex(ValueError, 'snapshot/header'): E.recover_producer(row, altered) altered_row = copy.deepcopy(row) altered_row['spans'][0][0] += 1 with self.assertRaisesRegex(ValueError, 'mask mismatch'): E.recover_producer(altered_row, producer) self.assertEqual(row['mask_version'], 2)
def test_grounding_review_requires_real_slot_provenance_and_complete_checks(self): producer = {'state': self.st} row = {'kind': 'diary'} review = {'decision': 'accept', 'reviewer': 'independent-grounding-review', 'evidence': 'source-linked physical-log review', 'state_sha256': E.value_hash(self.st), 'checks': {key: True for key in ('events', 'actors', 'gift_direction', 'game_counts', 'unsupported_claims')}} E.grounded_review(review, row, producer) for bad in ({**review, 'state_sha256': 'wrong'}, {**review, 'decision': 'uncertain'}, {**review, 'checks': {**review['checks'], 'gift_direction': False}}, {**review, 'reviewer': ''}): with self.assertRaises(ValueError): E.grounded_review(bad, row, producer) # Historical row-only review inputs are not certificates of grounding. with self.assertRaises(ValueError): E.grounded_review({'text_sha256': 'retained-row'}, row, producer)
def test_request_accounting_counts_crashed_calls_and_transitive_history(self): from pathlib import Path with tempfile.TemporaryDirectory() as directory: first, second = Path(directory) / 'first', Path(directory) / 'second' first.mkdir() second.mkdir() E.save_json(first / 'generation.json', {'status': 'running', 'prior_generations': []}) (first / 'api-calls.jsonl').write_text(json.dumps({'event': 'start', 'request_id': 'crashed', 'group_id': 'train/interrupted'}) + '\n') (first / 'requests.jsonl').write_text(json.dumps({'group_id': 'train/old'}) + '\n') E.save_json(second / 'generation.json', {'prior_generations': [E.source_identity(first / 'generation.json')]}) (second / 'api-calls.jsonl').write_text(json.dumps({'event': 'start', 'request_id': 'finished', 'group_id': 'train/completed'}) + '\n' + json.dumps({'event': 'finish', 'request_id': 'finished', 'raw': 'answer'}) + '\n') _, calls, groups = E.request_history([str(second), str(first)]) self.assertEqual(calls, {'crashed', 'finished'}) self.assertEqual(groups, {'train/old', 'train/interrupted', 'train/completed'}) E.save_json(first / 'generation.json', {'status': 'rewritten'}) with self.assertRaisesRegex(ValueError, 'changed'): E.request_history([str(second)])
def test_shared_replay_checks_duplicate_gift_commit_order(self): from pathlib import Path import semantic_data as S binary = Path(__file__).resolve().parents[1] / 'server/target/debug/tokenling-contract' if not binary.is_file(): self.skipTest('build existing tokenling-contract for real gift persistence') st = copy.deepcopy(self.st) st.update(asleep=False, shelf=['star']) st['stats']['energy'] = 20 scene = {'focus': 'sleep', 'variant': 'nap', 'slots': [ G.curriculum_slot(['* Sam turned off the lights *'], mode='NAP'), G.curriculum_slot(['Sam: rest well'], mode='DREAM')]} engine = S.EngineAdapter(binary) valid = G.replay_curriculum(st, scene, ['[gift:star] [sleep]', '[dream:star]']) evidence = E.shared_producer_replay(valid, {'state': st}, engine) self.assertEqual(evidence['turns'][0]['after']['shelf'], ['star', 'star']) skipped = G.replay_curriculum(st, scene, ['[sleep] [gift:star]', '[dream:star]']) with self.assertRaisesRegex(ValueError, 'multiplicity'): E.shared_producer_replay(skipped, {'state': st}, engine) st['shelf'] = ['star'] * 24 bounded = G.replay_curriculum(st, scene, ['[gift:flower] [sleep]', '[dream:star]']) result = E.shared_producer_replay(bounded, {'state': st}, engine) self.assertEqual(result['turns'][0]['after']['shelf'], ['star'] * 23 + ['flower'])
def test_legacy_started_request_requires_recoverable_scene_identity(self): from pathlib import Path with tempfile.TemporaryDirectory() as directory: root = Path(directory) E.save_json(root / 'generation.json', {'status': 'running'}) (root / 'api-calls.jsonl').write_text(json.dumps({ 'event': 'start', 'request_id': 'old-call', 'case_id': 'old-case'}) + '\n') with self.assertRaisesRegex(ValueError, 'group reservation'): E.request_history([root]) (root / 'requests.jsonl').write_text(json.dumps({ 'case_id': 'old-case', 'group_id': 'train/recovered'}) + '\n') _, calls, groups = E.request_history([root]) self.assertEqual(calls, {'old-call'}) self.assertEqual(groups, {'train/recovered'})
def test_semantic_workers_overlap_http_without_corrupting_request_accounting(self): from pathlib import Path from types import SimpleNamespace import threading import semantic_data as S barrier = threading.Barrier(4) active = {'current': 0, 'maximum': 0} lock = threading.Lock() raw = json.dumps(['hello there', 'a gentle day']) def teacher(prompt, **kwargs): with lock: active['current'] += 1 active['maximum'] = max(active['maximum'], active['current']) try: barrier.wait(timeout=5) return raw finally: with lock: active['current'] -= 1 cases = [] for index in range(8): st = copy.deepcopy(self.st) st['asleep'] = False cases.append({'case_id': f'train/{index}', 'group_id': f'train/group/{index}', 'template_id': 'train/general', 'family': 'capture', 'stage': 'kid', 'state': st, 'scene': {'focus': 'general', 'variant': 'greeting', 'slots': [ G.curriculum_slot(['Sam: hello']), G.curriculum_slot(['Sam: good day'])]}, 'gold': {}}) with tempfile.TemporaryDirectory() as directory: root = Path(directory) manifest = root / 'manifest.json' E.save_json(manifest, {}) args = SimpleNamespace(manifest=str(manifest), workers=4, per_cell=1, curriculum_chats=0, diaries=0, offset=0, prior_generation=[], seed=85710, out=str(root / 'generated'), teacher_url='http://teacher.invalid', teacher_model='stand-in', teacher_artifact=None) with patch.object(S, 'load_manifest', return_value={}), patch.object(S, 'training_cases', return_value=cases), \ patch.object(S, 'isolation_check', return_value=[]), patch.object(G, 'Teacher', return_value=teacher), \ patch.object(E, 'request_history', return_value=([], {f'prior/{i}' for i in range(4088)}, set())), \ patch('builtins.print'): E.semantic_generate(args) ledger = [item for _, item in E.jsonl_rows(root / 'generated/api-calls.jsonl')] starts = [item for item in ledger if item['event'] == 'start'] finishes = [item for item in ledger if item['event'] == 'finish'] requests = [item for _, item in E.jsonl_rows(root / 'generated/requests.jsonl')] generation = E.load_json(root / 'generated/generation.json') self.assertEqual(active['maximum'], 4) self.assertEqual(len(starts), 8) self.assertEqual(len({item['request_id'] for item in starts}), 8) self.assertEqual({item['request_id'] for item in starts}, {item['request_id'] for item in finishes}) self.assertTrue(all(item['raw'] == raw for item in finishes)) self.assertEqual({item['case_id'] for item in requests}, {case['case_id'] for case in cases}) self.assertEqual(generation['api_calls'], 8) self.assertEqual(generation['completed_scenes'], 8) self.assertEqual(generation['structurally_valid'], 8) self.assertEqual(generation['workers'], 4) self.assertEqual(generation['total_api_calls'], 4096)
def test_correction_workers_reject_unsafe_shared_student_session(self): from types import SimpleNamespace with self.assertRaisesRegex(ValueError, 'requires --workers 1'): E.semantic_generate(SimpleNamespace(workers=4, student_export='must-not-be-loaded'))
def test_published_pairing_does_not_assign_reference_training_hash(self): graph = 'a' * 64 config = {'seed': 5, 'device': 'cpu', 'threads': 4, 'generation': {'react': {'context': 1024}}, 'sampling': {'temperature': .85, 'top_k': 40, 'top_p': .92, 'repetition_penalty': 1.0, 'seed_strategy': 'paired', 'eos_token_ids': [0]}} reference = {'status': 'complete', 'onnx_artifact': {'graph_sha256': graph}, 'experiment_contract': {'version': 1}, 'scenario_sha256': 'scenes', 'scenario_suite_sha256': {'semantic': 'cases'}, 'scenario_counts': {'semantic': 16}, 'config': config, 'dataset_sha256': None, 'rates': {'correct': 1.0}, 'rate_counts': {'correct': {'numerator': 16, 'denominator': 16}}} candidate = copy.deepcopy(reference) candidate['dataset_sha256'] = 'C1-only' candidate['onnx_artifact']['graph_sha256'] = 'b' * 64 policy = {'margin': 0, 'minimum': ['correct'], 'maximum': []} result = E.published_inference_gate(reference, candidate, policy, graph) self.assertTrue(result['passed'], result['reasons']) self.assertIsNone(reference['dataset_sha256']) self.assertFalse(E.quality_gate(reference, candidate, policy)['passed']) candidate['config']['sampling']['temperature'] = .5 self.assertFalse(E.published_inference_gate(reference, candidate, policy, graph)['passed'])
def test_curriculum_diary_teacher_body_and_doodles(self): st = copy.deepcopy(self.st) body = 'I feel cozy with Sam. I hope we have a gentle day.' def teacher(prompt, **kwargs): return json.dumps({'body': body, 'doodles': ['heart', 'moon']}) row = G.make_curriculum_diary(teacher, st) opener = row['text'].index('dear diary,') + len('dear diary,') self.assertEqual(row['text'][opener:], '\n' + body + '\n[doodle:heart] [doodle:moon]') self.assertEqual(row['spans'], [[opener, len(row['text'])]]) G.recheck(row) for result in ({'body': '[invented]', 'doodles': []}, {'body': body, 'doodles': ['invented']}, {'body': body, 'doodles': [], 'signature': 'PET'}): with self.subTest(result=result), self.assertRaises(ValueError): G.make_curriculum_diary(lambda *args, result=result, **kwargs: json.dumps(result), st)
def test_migration_diary_masks(self): for raw in ['dear diary,\ni dreamed of cookies.\n- Mochi\n[doodle:star]', ' Dear diary,\n\ni dreamed of cookies.\n- Mochi\n[doodle:star]\n ']: text, spans = G.check_diary(self.st, raw) legacy_text = W.header('diary', self.st) + raw cases = [('empty', dict(spans=[])), ('header trained', dict(spans=[[0, len(legacy_text)]])), ('missing', {})] for label, mask in cases: with self.subTest(body=raw, mask=label): old = dict(kind='diary', text=legacy_text, **mask) original = copy.deepcopy(old) migrated = M.migrate(old) self.assertEqual(migrated['text'], text) self.assertEqual(migrated['spans'], spans) self.assertEqual(old, original) G.recheck(migrated) opener_end = text.index('dear diary,') + len('dear diary,') self.assertEqual(spans, [[opener_end, len(text)]])
def test_migration_no_wake(self): for action in ['Sam: hello', '* Sam patted Mochi *', '* Sam turned on the lights *']: with self.subTest(action=action): raw = action + '\nMochi: zzz [dream:moon]\nMochi: [dream:star]' text, spans = G.check_chat(self.st, raw) old = dict(kind='chat', text=text, spans=[]) migrated = M.migrate(old) self.assertEqual(migrated['text'], text) self.assertEqual(migrated['spans'], spans) self.assertNotIn('* Mochi woke up *', migrated['text']) G.recheck(migrated)
def test_migration_mixed_wakes(self): actions = ['Sam: hello', '* Sam patted Mochi *', '* Sam turned on the lights *'] for sleeping_action in actions + ['\n'.join(actions)]: for awake_action in actions: with self.subTest(sleeping=sleeping_action, awake=awake_action): between = '* Sam turned off the lights *\n' if 'lights' in sleeping_action and 'lights' in awake_action else '' raw = sleeping_action + '\nMochi: zzz [dream:moon]\n' + between + awake_action + '\n' raw += 'Mochi: hello Sam\nMochi: [emote:wave]' woke_action = (awake_action + '\n* Mochi woke up *' if 'lights' in awake_action else '* Mochi woke up *\n' + awake_action) expected = sleeping_action + '\nMochi: zzz [dream:moon]\n' + between + woke_action + '\n' expected += 'Mochi: hello Sam\nMochi: [emote:wave]' text, spans = G.check_chat(self.st, expected) old = dict(kind='chat', text=W.header('chat', self.st) + raw, spans=[]) migrated = M.migrate(old) self.assertEqual(migrated['text'], text) self.assertEqual(migrated['spans'], spans) self.assertEqual(migrated['text'].count('* Mochi woke up *'), 1) self.assertLess(migrated['text'].index('[dream:moon]'), migrated['text'].index('* Mochi woke up *')) G.recheck(migrated)
def test_migration_sleep_yes(self): self.st['stats']['tummy'] = 10 offer = '* Sam offered Mochi a cookie *\nMochi: cookie? [yes]\n' for prefix in ['', 'Sam: hello\nMochi: zzz [dream:moon]\n']: with self.subTest(prefix=prefix): raw = prefix + offer + 'Mochi: yummy [emote:happy]' expected = prefix + offer + '* Mochi woke up *\nMochi: yummy [emote:happy]' text, spans = G.check_chat(self.st, expected) old = dict(kind='chat', text=W.header('chat', self.st) + raw, spans=[]) migrated = M.migrate(old) self.assertEqual(migrated['text'], text) self.assertEqual(migrated['spans'], spans) self.assertIn('[yes]\n* Mochi woke up *\n', migrated['text']) G.recheck(migrated) # Already-current rows must not acquire a second wake either. self.assertEqual(M.migrate({**migrated, 'spans': []}), migrated)
def test_migration_standin(self): hdr = W.header('chat', self.st).replace('---\n', 'today: Sam said: hello; ate a cookie\n---\n') row = M.migrate(dict(kind='chat', text=hdr + 'Sam: hello\nMochi: [recall:cat]{nothing} hello\nMochi: hi\n', spans=[])) self.assertNotIn('Sam said:', row['text']) self.assertIn('* Mochi woke up *\nSam: hello', row['text']) start = row['text'].index('{nothing}') self.assertTrue(all(b <= start or a >= start + 9 for a, b in row['spans']))
def test_migration_cli(self): self.st['stage'] = 'kid' old = dict(kind='chat', stage='kid', text=W.header('chat', self.st) + 'Sam: hello\nMochi: hello Sam\nMochi: good morning\n', spans=[]) migrated = M.migrate(old) teen = {**migrated, 'stage': 'teen', 'text': migrated['text'].replace(' · kid · ', ' · teen · ')} # The stage is one character longer, so rebuild the spans for the stand-in. teen = M.migrate(teen) with tempfile.TemporaryDirectory() as d: def write(name, content): path = os.path.join(d, name) with open(path, 'w') as f: f.write(content) return path src = write('old.jsonl', json.dumps(old)+'\n'+json.dumps(teen)+'\n') new = write('new.jsonl', json.dumps(teen)+'\n') excluded = write('exclude.json', json.dumps({hashlib.sha256(migrated['text'].encode()).hexdigest(): 'stand-in'})) out = os.path.join(d, 'out.jsonl') subprocess.run([sys.executable, M.__file__, src, out, new, '--exclude', excluded], check=True, capture_output=True) with open(out) as f: rows = [json.loads(l) for l in f] self.assertEqual(rows, [teen])
if __name__ == '__main__': unittest.main()