#!/usr/bin/env python3 # -*- coding: utf-8 -*- """Millionaire game harness: play, watch, control and debug the real game -- headless or in the running Kodi. Headless runs the real game modal, adapter, controller, content source and AI clients in this process (fake Kodi modules, real HTTP to the LLM). Live attaches to the game modal in Kodi through the game probe (needs the dev harness listener: scripts/dialog_harness/kodi_live.py start). game.py run --meta matrix AI game about a demo title (live LLM, costs credit) game.py run --demo matrix offline demo content (no AI) game.py run --title "Pelíšky" --plot ... AI game about any title game.py run ... --live [--input click] [--shots] the same game in the running Kodi game.py suite --titles all --jobs 4 one game per demo title, overview suite.md game.py serve --meta matrix & headless game kept running for state/act/... commands game.py state | act start | act B | back | wait | chat | trace | last | call | set | quit [--live] game.py open --demo matrix | --meta matrix open a game in Kodi (live) game.py repl [--meta matrix | --live] interactive play with state, prompts and LLM calls game.py prompts --meta matrix the exact prompts the game would send (files) game.py resend RUN_DIR --call 7 [--n 3] [--system-append ...] send a recorded LLM request again game.py report RUN_DIR rebuild report.md from trace + steps game.py compare RUN_A RUN_B metrics side by side (runs or suite folders) game.py models | selftest Settings come from Kodi (read only); --provider/--model/--set key=value override them for headless runs only. Outputs: scripts/game_harness/out/ (runs/, serve/, prompts/). """ import argparse import json import os import subprocess import sys import time HERE = os.path.dirname(os.path.abspath(__file__)) sys.path.insert(0, HERE) import env # noqa: E402 SERVE_DIR = os.path.join(HERE, 'out', 'serve') SETTINGS = {} # --------------------------------------------------------------------------- # setup # --------------------------------------------------------------------------- def add_subject(parser): group = parser.add_argument_group('subject') group.add_argument('--demo', help='offline demo title id (demo content, no AI)') group.add_argument('--mode', choices=('millionaire', 'story'), default='millionaire') group.add_argument('--difficulty', choices=('easy', 'medium', 'hard'), default=None) group.add_argument('--mistakes', default='', help='Story demo: wrong answers at these scene numbers, e.g. 1,2,3') group.add_argument('--meta', help='AI game about a demo title (its title, plot, genre)') group.add_argument('--title') group.add_argument('--plot', default='') group.add_argument('--genre', default='') group.add_argument('--media', default='movie', choices=('movie', 'series', 'channel', 'actor', 'director', 'writer', 'creator')) group.add_argument('--year', default='') group.add_argument('--tmdb', default='', help='TMDb id (live: the router loads plot/genre itself)') def add_settings(parser): group = parser.add_argument_group('settings (headless)') group.add_argument('--provider', choices=('openai', 'anthropic', 'google', 'mistral', 'groq')) group.add_argument('--model') group.add_argument('--set', action='append', default=[], metavar='KEY=VALUE', help='any addon setting') group.add_argument('--design', choices=('classic', 'modern')) group.add_argument('--realtime', action='store_true', help='real countdowns/pauses (default: instant)') group.add_argument('--time-scale', type=float) group.add_argument('--pace', type=float, default=0.0, help='demo content pauses (0: none)') prompt = parser.add_argument_group('prompt overrides (headless)') prompt.add_argument('--system-file', help='whole system prompt; {{BASE}} = the production one') prompt.add_argument('--system-append', help='text or @file added to the system prompt') prompt.add_argument('--system-replace', action='append', default=[], metavar='OLD=>NEW') prompt.add_argument('--msg-append', action='append', default=[], metavar='KIND=TEXT', help='welcome|question|feedback=text or @file added to that user message') def setup(args, headless=True): """Install the fake Kodi with the setting overrides of the command line.""" overrides = {} if headless: if getattr(args, 'difficulty', None) and getattr(args, 'mode', None) == 'story': overrides['story_difficulty'] = str(('easy', 'medium', 'hard').index(args.difficulty)) for item in getattr(args, 'set', []) or []: key, sep, value = item.partition('=') if not sep: raise SystemExit('--set needs KEY=VALUE: ' + item) overrides[key.strip()] = value if getattr(args, 'provider', None): overrides['ai_provider'] = args.provider if getattr(args, 'model', None): provider = overrides.get('ai_provider') or env._user_settings().get('ai_provider') or 'openai' overrides['{}_model'.format(provider)] = args.model if getattr(args, 'design', None): overrides['milionar_game_design'] = {'classic': '0', 'modern': '1'}[args.design] scale = 1.0 if getattr(args, 'realtime', False) else (getattr(args, 'time_scale', None) or 0.0) SETTINGS.update(env.install(overrides, time_scale=scale)) if headless and getattr(args, 'model', None): from wcs.ai import model_catalog provider = SETTINGS.get('ai_provider', 'openai') if args.model not in model_catalog.MODELS.get(provider, ()): model_catalog.MODELS[provider] = tuple(model_catalog.MODELS.get(provider, ())) + (args.model,) print('note: {} is not in the model catalog of {} (no price, allowed for this run)'.format( args.model, provider)) return SETTINGS def prompt_overrides(args): import session return session.apply_prompt_overrides(args.system_file, args.system_append, args.system_replace, args.msg_append) def subject_of(args): import session if args.demo: return session.subject_from_args(demo_meta=args.demo), 'demo' subject = session.subject_from_args(args.meta, args.title, args.plot, args.genre, args.media, args.year) subject['tmdb_id'] = args.tmdb return subject, 'ai' def model_meta(): try: from wcs.ai.AIProvider import AIProvider provider = AIProvider() return provider.get_current_provider(), provider.get_current_model() except Exception as exc: return 'unknown', str(exc) def driver_for(args): """A Channel to the running game: live Kodi, or the headless `serve` daemon.""" import session if args.live: return session.LiveGame() return session.Channel(args.dir or SERVE_DIR, 'headless') # --------------------------------------------------------------------------- # rendering # --------------------------------------------------------------------------- def render(snap, reveal=True): if not snap.get('alive'): return '[closed] the game is not running' + (' (no state yet)' if snap.get('missing') else '') view = snap.get('view') or {} head = '[{}] {} · round {} · {} · focus {}{}{}'.format( snap.get('screen'), snap.get('game_state'), snap.get('round'), view.get('Money') or view.get('Earned'), snap.get('focus'), '' if snap.get('settled') else ' · busy', ' · EXIT?' if snap.get('exit_prompt') else '') lines = [head] screen = snap.get('screen') if screen == 'intro': lines.append('moderator: ' + (view.get('Moderator') or '…')) question = snap.get('question') or {} if screen in ('question', 'feedback', 'gameover') and question: if question.get('moderator_intro') and screen == 'question': lines.append('intro: ' + question['moderator_intro']) lines.append('Q: ' + question.get('question', '')) removed = set(snap.get('removed') or []) for index, letter in enumerate('ABCD'): options = question.get('options') or [] text = options[index] if index < len(options) else '' marks = '' if reveal and index == question.get('correct'): marks += ' *' if index in removed: marks += ' (50:50 removed)' if snap.get('selected') == index and screen != 'question': marks += ' <- player' lines.append(' {}: {}{}'.format(letter, text, marks)) if screen in ('feedback', 'gameover', 'win'): lines.append('result: ' + (view.get('Result') or '')) if view.get('Transition'): lines.append('next: ' + view['Transition']) if view.get('Explanation'): lines.append('why: ' + view['Explanation']) if view.get('TipState'): lines.append('tip [{}]: {}'.format(view['TipState'], view.get('TipText', ''))) if view.get('PollState'): lines.append('poll [{}]: A {PollA}% B {PollB}% C {PollC}% D {PollD}%'.format(view['PollState'], **view)) lifelines = snap.get('lifelines') or {} lines.append('lifelines: ' + ' '.join('{}{}'.format(k, '✗' if v else '✓') for k, v in lifelines.items()) + ' · score {}'.format(view.get('Score', ''))) return '\n'.join(lines) def print_step(record): if record['step'] == 'abort': print(' ABORT {}'.format(record.get('error'))) return snap = record.get('snapshot') or {} question = snap.get('question') or {} detail = '' if record['step'].startswith('question-'): detail = question.get('question', '')[:90] elif record['step'].startswith('answer-'): detail = (snap.get('view') or {}).get('Result', '')[:90] print(' {:<14} {:>6} ms {:<9} {}'.format(record['step'], record['wait_ms'], record.get('screen') or '', detail)) # --------------------------------------------------------------------------- # run / suite # --------------------------------------------------------------------------- def cmd_run(args): import checks import report import scenario import session settings = setup(args, headless=not args.live) overrides = prompt_overrides(args) if not args.live else {} subject, source = subject_of(args) if args.mode == 'story' and source != 'demo': raise SystemExit('Use --demo ID for offline Story checks or story-review for explicitly requested AI review.') run_dir = args.out or session.new_run_dir('{}-{}{}'.format(subject['id'], source, '-live' if args.live else '')) os.makedirs(run_dir, exist_ok=True) provider, model = model_meta() if source == 'ai' else ('-', '-') meta = {'title': subject['title'], 'subject': subject, 'source': source, 'target': 'live' if args.live else 'headless', 'provider': provider, 'model': model, 'policy': args.policy, 'lifelines': args.lifelines, 'design': None, 'overrides': overrides, 'label': args.label, 'started': time.strftime('%Y-%m-%d %H:%M:%S'), 'settings': env.public_settings(settings), 'time_scale': 1.0 if args.realtime else (args.time_scale or 0.0)} if args.mode == 'story': meta.update(mode='story', difficulty=args.difficulty, mistakes=args.mistakes) if not args.quiet: print('{} {} [{}] {}/{} -> {}'.format('LIVE' if args.live else 'headless', subject['title'], source, provider, model, run_dir)) on_step = None if args.quiet else print_step if args.live: driver = session.LiveGame() driver.open(session.live_url(args.demo, subject, mode=args.mode)) if args.shots: shots = os.path.join(run_dir, 'shots') os.makedirs(shots, exist_ok=True) previous = on_step def on_step(record, previous=previous): if previous: previous(record) env.real_sleep(0.7) # property settlement is immediate; Kodi's screen animations are not driver.screenshot(os.path.join(shots, '{:02d}-{}.png'.format(len(os.listdir(shots)) + 1, record['step']))) on_step = on_step else: driver = session.HeadlessGame(run_dir, subject, source, design=settings.get('milionar_game_design'), pace=args.pace, mode=args.mode) meta['design'] = driver.snapshot().get('design') original_difficulty = None if args.mode == 'story': import story as story_runner if args.live and args.difficulty: original_difficulty = story_runner.set_live_difficulty(driver, args.difficulty) try: runner = story_runner if args.mode == 'story' else scenario extra = {'mistakes': tuple(int(n) for n in args.mistakes.split(',') if n), 'difficulty': args.difficulty} if args.mode == 'story' else {} steps, outcome = runner.play(driver, policy=args.policy, lifelines=scenario.parse_lifelines(args.lifelines), rules=args.rules, seed=args.seed, timeout=args.timeout, input_mode=args.input, on_step=on_step, **extra) finally: events = driver.events() if args.live: if original_difficulty is not None: driver.back() if driver.snapshot().get('exit_prompt'): driver.act('exit_confirm') driver.wait_settled({'title'}, timeout=args.timeout) story_runner.set_live_difficulty(driver, {'10': 'easy', '15': 'medium', '20': 'hard'}[original_difficulty]) session.copy_live_trace(run_dir) if not args.keep_open: driver.close() else: driver.close() if args.mode == 'story': status = story_runner.write(run_dir, meta, steps, outcome) print(json.dumps({'run': run_dir, 'verdict': 'PASS' if status == 0 else 'FAIL', 'outcome': outcome})) return status analysis = checks.analyze(events, steps, settings, source, realtime=meta['time_scale'] >= 1) data = report.write(run_dir, meta, analysis, steps, outcome) if args.quiet: print(json.dumps({'run': run_dir, 'verdict': data['verdict'], 'outcome': outcome})) else: report.console(data, run_dir) return 1 if data['totals']['errors'] else 0 def cmd_story_review(args): """Explicit one-pack AI benchmark, separate from the silent authored demo runner.""" import session import story as story_runner if args.mode != 'story' or args.demo: raise SystemExit('story-review needs --meta ID or --title TITLE and --mode story (default).') if args.system_file or args.system_append or args.system_replace or args.msg_append: raise SystemExit('story-review uses the production Story prompt; Millionaire overrides do not apply.') settings = setup(args) # Headless review never requests speech, including when the user's normal game voice is enabled. for key in settings: if key.startswith('milionar_tts_') and key.rsplit('_', 1)[-1] in ( 'welcome', 'intro', 'question', 'options', 'result', 'transition', 'explanation'): settings[key] = 'false' subject, _source = subject_of(args) from wcs.games.millionaire.subject import Subject value = Subject(title=subject['title'], plot=subject.get('plot', ''), year=subject.get('year', ''), genre=subject.get('genre', ''), media_type=subject.get('media', 'movie')) folder = args.out or session.new_run_dir(subject['id'] + '-story-review') if any(value is not None and value <= 0 for value in (args.output_tokens, args.request_timeout)): raise SystemExit('Review output/timeout overrides must be positive.') result = story_runner.review(value, args.difficulty or 'medium', folder, args.capture_content, args.output_tokens, args.request_timeout) print(json.dumps(dict(result, run=folder), ensure_ascii=False)) return 0 if result['verdict'] == 'PASS' else 1 def cmd_suite(args): import report import session setup(args, headless=True) from wcs.games.demo import catalog ids = [t.id for t in catalog.titles()] if args.titles == 'all' else [t.strip() for t in args.titles.split(',')] out_dir = session.new_run_dir('suite-{}-{}{}'.format(args.source, len(ids), '-' + args.label if args.label else '')) passthrough = [] for flag in ('provider', 'model', 'design', 'system_file', 'system_append', 'policy', 'lifelines', 'seed', 'time_scale'): value = getattr(args, flag, None) if value not in (None, ''): passthrough += ['--' + flag.replace('_', '-'), str(value)] for item in args.set: passthrough += ['--set', item] for item in args.system_replace: passthrough += ['--system-replace', item] for item in args.msg_append: passthrough += ['--msg-append', item] if args.realtime: passthrough.append('--realtime') jobs = [] print('suite {} games, {} parallel -> {}'.format(len(ids), args.jobs, out_dir)) pending = list(ids) running = [] results = [] while pending or running: while pending and len(running) < args.jobs: title_id = pending.pop(0) run_dir = os.path.join(out_dir, title_id) command = [sys.executable, os.path.abspath(__file__), 'run', '--' + ('demo' if args.source == 'demo' else 'meta'), title_id, '--out', run_dir, '--quiet', '--label', args.label or ''] + passthrough running.append((title_id, run_dir, subprocess.Popen(command, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True))) for item in list(running): title_id, run_dir, process = item if process.poll() is None: continue running.remove(item) output = process.stdout.read() if os.path.exists(os.path.join(run_dir, 'summary.json')): data = report.load(run_dir) results.append((run_dir, data)) print(' {:<16} {:<5} {:<12} err {} warn {}'.format(title_id, data['verdict'], data['outcome'], data['totals']['errors'], data['totals']['warnings'])) else: print(' {:<16} CRASH\n{}'.format(title_id, output[-2000:])) env.real_sleep(0.2) jobs.append(out_dir) path = report.suite(sorted(results, key=lambda r: r[0]), out_dir) print('suite: ' + path) return 1 if any(d['totals']['errors'] for _, d in results) or len(results) < len(ids) else 0 # --------------------------------------------------------------------------- # serve + channel commands # --------------------------------------------------------------------------- def cmd_serve(args): import session settings = setup(args, headless=True) overrides = prompt_overrides(args) subject, source = subject_of(args) folder = args.dir or SERVE_DIR os.makedirs(folder, exist_ok=True) for name in os.listdir(folder): path = os.path.join(folder, name) if os.path.isfile(path): os.remove(path) if os.path.isdir(os.path.join(folder, 'cmd')): for name in os.listdir(os.path.join(folder, 'cmd')): os.remove(os.path.join(folder, 'cmd', name)) game = session.HeadlessGame(folder, subject, source, design=settings.get('milionar_game_design'), pace=args.pace, mode=args.mode) provider, model = model_meta() if source == 'ai' else ('-', '-') with open(os.path.join(folder, 'meta.json'), 'w', encoding='utf-8') as handle: json.dump({'subject': subject, 'source': source, 'provider': provider, 'model': model, 'pid': os.getpid(), 'overrides': overrides}, handle, ensure_ascii=False, indent=1) print('serving {} [{}] {}/{} in {} (pid {})'.format(subject['title'], source, provider, model, folder, os.getpid()), flush=True) quit_flag = os.path.join(folder, 'QUIT') idle_deadline = time.time() + args.idle try: while not os.path.exists(quit_flag) and time.time() < idle_deadline: if os.listdir(os.path.join(folder, 'cmd')): idle_deadline = time.time() + args.idle env.real_sleep(0.3) finally: game.close() if os.path.exists(quit_flag): os.remove(quit_flag) print('serve stopped', flush=True) return 0 def cmd_channel(args): setup(args, headless=False) driver = driver_for(args) name = args.command if name == 'state': snap = driver.snapshot() print(json.dumps(snap, ensure_ascii=False, indent=1) if args.json else render(snap, reveal=not args.hide)) return 0 if name in ('act', 'click'): for intent in args.intents: (driver.click if name == 'click' else driver.act)(intent) if not args.no_wait: env.real_sleep(0.15) driver.wait_settled(timeout=args.timeout) print(render(driver.snapshot())) return 0 if name == 'back': driver.back() if not args.no_wait: env.real_sleep(0.15) driver.wait_settled(timeout=args.timeout) print(render(driver.snapshot())) return 0 if name == 'wait': snap = driver.wait_settled(set(args.screen) if args.screen else None, timeout=args.timeout) print(render(snap)) return 0 if name == 'chat': chat = driver.chat() if args.json: print(json.dumps(chat, ensure_ascii=False, indent=1)) else: print('=== SYSTEM ({} chars) ===\n{}'.format(len(chat.get('system_prompt', '')), chat.get('system_prompt'))) for message in chat.get('messages', []): print('=== {} ===\n{}'.format(message['role'].upper(), message['content'])) return 0 if name == 'trace': events = driver.events() if args.kind: events = [e for e in events if e['kind'].startswith(args.kind)] for event in events[-args.last:]: print(brief_event(event) if not args.json else json.dumps(event, ensure_ascii=False)) return 0 if name == 'last': events = [e for e in driver.events() if e['kind'].startswith('llm.')] if not events: print('no LLM call yet') return 0 call = events[-1]['call'] for event in (e for e in events if e['call'] == call): print(json.dumps(event, ensure_ascii=False, indent=1)) return 0 if name == 'call': values = [json.loads(v) if v[:1] in '[{0123456789-tfn"' else v for v in args.args] print(json.dumps(driver.command('call', method=args.method, args=values), ensure_ascii=False, indent=1)) return 0 if name == 'set': attrs = {} for item in args.pairs: key, _, value = item.partition('=') try: attrs[key] = json.loads(value) except ValueError: attrs[key] = value print(json.dumps(driver.command('set', attrs=attrs), ensure_ascii=False)) return 0 if name == 'quit': if args.live: driver.close() else: open(os.path.join(args.dir or SERVE_DIR, 'QUIT'), 'w').close() print('closing') return 0 raise SystemExit('unknown command ' + name) def brief_event(event): kind = event['kind'] head = '{:>4} {:>8.2f}s {:<15}'.format(event['seq'], event['t'], kind) if kind == 'state': return head + ' {screen} {game_state} r{round} settled={settled}'.format(**event) if kind.startswith('content.'): return head + ' {} {}{}'.format(event.get('method'), event.get('args', '') if kind == 'content.call' else '{} ms'.format(event.get('ms')), ' ' + event.get('error', '') if kind == 'content.error' else '') if kind == 'llm.request': messages = event.get('messages') or [] last = messages[-1]['content'] if messages else str(event.get('kwargs', '')) return head + ' #{} {} {}/{} {}'.format(event['call'], event.get('purpose'), event.get('provider'), event.get('model'), last[:100].replace('\n', ' ')) if kind == 'llm.response': return head + ' #{} {} ms {}'.format(event['call'], event.get('ms'), json.dumps(event.get('data'), ensure_ascii=False)[:120]) if kind == 'llm.error': return head + ' #{} {} ms {}'.format(event['call'], event.get('ms'), event.get('error')) if kind == 'http': return head + ' {} {} -> {} {} ms'.format(event.get('method'), event.get('url'), event.get('status', event.get('error')), event.get('ms')) if kind == 'command': return head + ' {} {}'.format(event.get('op'), event.get('intent', event.get('method', ''))) return head + ' ' + json.dumps({k: v for k, v in event.items() if k not in ('seq', 't', 'kind', 'thread')}, ensure_ascii=False)[:140] def cmd_open(args): import session setup(args, headless=False) subject, source = subject_of(args) game = session.LiveGame() snap = game.open(session.live_url(args.demo, subject, mode=args.mode)) print(render(snap)) return 0 # --------------------------------------------------------------------------- # repl # --------------------------------------------------------------------------- REPL_HELP = """commands: a b c d answer 5050 tip poll lifelines start next rules again results retry title home menu buttons (act) click live: real Kodi click back Back key s | state screen json full snapshot chat system prompt + chat sys system prompt only last last LLM request/response trace [n] last n events resend [text] send the last chat request again (text: appended to the system prompt) call [json args] set = hide/reveal correct answer q | quit""" def cmd_repl(args): import session settings = setup(args, headless=not args.live) if args.live: driver = session.LiveGame() if args.demo or args.meta or args.title: subject, _ = subject_of(args) driver.open(session.live_url(args.demo, subject, mode=args.mode)) else: prompt_overrides(args) subject, source = subject_of(args) run_dir = session.new_run_dir('{}-{}-repl'.format(subject['id'], source)) driver = session.HeadlessGame(run_dir, subject, source, design=settings.get('milionar_game_design'), pace=args.pace, mode=args.mode) print('headless {} [{}] {}/{} trace: {}'.format(subject['title'], source, *model_meta(), run_dir)) reveal = True print(REPL_HELP) print(render(driver.snapshot(), reveal)) aliases = {'a': 'answer0', 'b': 'answer1', 'c': 'answer2', 'd': 'answer3', '5050': 'life5050', 'tip': 'lifetip', 'poll': 'lifeaudience'} menu = ('start', 'next', 'rules', 'again', 'results', 'retry', 'title', 'home', 'exit_stay', 'exit_confirm') try: while True: try: line = input('milionar> ').strip() except EOFError: break if not line: continue word, _, rest = line.partition(' ') try: if word in ('q', 'quit', 'exit'): break elif word in aliases or word in menu or word.startswith('answer') or word.startswith('life'): driver.act(aliases.get(word, word)) _repl_wait(driver, args.timeout) print(render(driver.snapshot(), reveal)) elif word == 'click': driver.click(rest.strip()) _repl_wait(driver, args.timeout) print(render(driver.snapshot(), reveal)) elif word == 'back': driver.back() _repl_wait(driver, args.timeout) print(render(driver.snapshot(), reveal)) elif word in ('s', 'state'): print(render(driver.snapshot(), reveal)) elif word == 'json': print(json.dumps(driver.snapshot(), ensure_ascii=False, indent=1)) elif word in ('chat', 'sys'): chat = driver.chat() print(chat.get('system_prompt')) if word == 'chat': for message in chat.get('messages', []): print('--- {} ---\n{}'.format(message['role'], message['content'])) elif word == 'last': events = [e for e in driver.events() if e['kind'].startswith('llm.')] for event in (e for e in events if events and e['call'] == events[-1]['call']): print(json.dumps(event, ensure_ascii=False, indent=1)) elif word == 'trace': for event in driver.events()[-int(rest or 20):]: print(brief_event(event)) elif word == 'resend': requests = [e for e in driver.events() if e['kind'] == 'llm.request' and e.get('purpose') == 'chat'] if not requests: print('no chat request yet') else: resend_one(requests[-1], system_append=rest or None) elif word == 'call': method, _, raw = rest.partition(' ') values = json.loads(raw) if raw.strip() else [] print(driver.command('call', method=method, args=values if isinstance(values, list) else [values])) elif word == 'set': key, _, raw = rest.partition('=') print(driver.command('set', attrs={key.strip(): json.loads(raw)})) elif word == 'hide': reveal = False elif word == 'reveal': reveal = True else: print(REPL_HELP) except Exception as exc: print('error: {}'.format(exc)) finally: if not args.live: driver.close() return 0 def _repl_wait(driver, timeout): env.real_sleep(0.15) try: driver.wait_settled(timeout=timeout) except Exception as exc: print('(still busy: {})'.format(exc)) # --------------------------------------------------------------------------- # prompts / resend / report / compare / models / selftest # --------------------------------------------------------------------------- def cmd_prompts(args): setup(args, headless=True) overrides = prompt_overrides(args) subject, _ = subject_of(args) from wcs.games import millionaire_prompts as prompts folder = args.out or os.path.join(HERE, 'out', 'prompts', subject['id']) os.makedirs(folder, exist_ok=True) if args.mode == 'story': from wcs.games.story import prompts as story_prompts from wcs.games.millionaire.subject import Subject system, request = story_prompts.build(Subject(title=subject['title'], media_type=subject['media'], plot=subject['plot'], year=subject['year']), args.difficulty or 'medium') for name, value in (('story-system.txt', system), ('story-request.txt', request)): with open(os.path.join(folder, name), 'w', encoding='utf-8') as handle: handle.write(value) print('Story prompt rendered offline: ' + folder) if args.show: print(system + '\n\n' + request) return 0 system = prompts.build_system_prompt(media_type=subject['media'], title=subject['title'], plot=subject['plot'], genre=subject['genre']) files = {'system.txt': system, 'welcome.txt': prompts.build_welcome_message()} for number in range(1, 10): files['question_{}.txt'.format(number)] = prompts.build_question_message(number) files['feedback_correct_3.txt'] = prompts.build_feedback_message(3, 'B', 'Možnost B', 'B', 'Možnost B', True) files['feedback_wrong_3.txt'] = prompts.build_feedback_message(3, 'A', 'Možnost A', 'B', 'Možnost B', False) files['feedback_win_9.txt'] = prompts.build_feedback_message(9, 'C', 'Možnost C', 'C', 'Možnost C', True, True) for name, text in files.items(): with open(os.path.join(folder, name), 'w', encoding='utf-8') as handle: handle.write(text) print('system prompt: {} chars (~{} tokens), provider/model {}/{}{}'.format( len(system), len(system) // 4, *model_meta(), ' overrides: {}'.format(overrides) if overrides else '')) print('files: ' + folder) if args.show: print(system) return 0 def resend_one(request, system_append=None, system_replace=(), times=1): """Send a recorded llm.request again; prints the parsed answer, time and checks.""" import checks from wcs.ai import AIHandler system = request.get('system_prompt', '') for item in system_replace or (): old, _, new = item.partition('=>') if old not in system: raise SystemExit('--system-replace: not found: ' + old[:80]) system = system.replace(old, new) if system_append: system = system.rstrip() + '\n\n' + system_append.strip() results = [] for attempt in range(times): started = time.time() if request.get('purpose') == 'tip': data, usage = AIHandler.generate_director_tip(*request.get('args', []), **request.get('kwargs', {})), {} else: data, usage = AIHandler.generate_millionaire_chat(system, request.get('messages', [])) ms = int((time.time() - started) * 1000) tokens_in, tokens_out, cached = checks._usage_tokens(usage) print('--- attempt {} · {} ms · tokens {}/{} (cache {})'.format(attempt + 1, ms, tokens_in, tokens_out, cached)) print(json.dumps(data, ensure_ascii=False, indent=1)) last = (request.get('messages') or [{}])[-1].get('content', '') if isinstance(data, dict) and 'question' in data: import re match = re.search(r'Kolo (\d+)/9', last) number = int(match.group(1)) if match else 1 events = [{'kind': 'content.call', 'call': 1, 'method': 'question', 'args': [number]}, {'kind': 'content.result', 'call': 1, 'method': 'question', 'result': data, 'ms': ms}] for item in checks.analyze(events, [], SETTINGS, 'ai')['issues']: print(' {} {}: {}'.format(item['severity'], item['code'], item['message'])) results.append(data) return results def cmd_resend(args): setup(args, headless=True) path = os.path.join(args.run, 'trace.jsonl') if os.path.isdir(args.run) else args.run with open(path, encoding='utf-8') as handle: events = [json.loads(line) for line in handle if line.strip()] requests = [e for e in events if e['kind'] == 'llm.request'] if args.call: chosen = [e for e in requests if e['call'] == args.call] else: chosen = requests[-1:] if not chosen: for event in requests: print(brief_event(event)) raise SystemExit('pick one of the calls above with --call N') resend_one(chosen[0], args.system_append and _text(args.system_append), args.system_replace, args.n) return 0 def _text(value): import session return session._text(value) def cmd_report(args): import checks import report setup(args, headless=False) with open(os.path.join(args.run, 'trace.jsonl'), encoding='utf-8') as handle: events = [json.loads(line) for line in handle if line.strip()] with open(os.path.join(args.run, 'steps.json'), encoding='utf-8') as handle: steps = json.load(handle) old = report.load(args.run) analysis = checks.analyze(events, steps, SETTINGS, old['meta'].get('source', 'ai'), realtime=(old['meta'].get('time_scale') or 0) >= 1) data = report.write(args.run, old['meta'], analysis, steps, old['outcome']) report.console(data, args.run) return 0 def cmd_questions(args): """All questions of a run or suite, compact (for reading them through).""" import report rounds_wanted = set(range(args.from_round, 10)) for run_dir in report._run_dirs(args.run): title = report.load(run_dir)['meta'].get('title') calls = {} with open(os.path.join(run_dir, 'trace.jsonl'), encoding='utf-8') as handle: for line in handle: event = json.loads(line) if event['kind'] == 'content.call': calls[event['call']] = event elif event['kind'] == 'content.result' and event.get('method') == 'question': number = (calls.get(event['call'], {}).get('args') or [0])[0] if number not in rounds_wanted: continue data = event.get('result') or {} options = data.get('options') or {} right = str(data.get('correct', '')).upper() wrong = ' / '.join(v for k, v in sorted(options.items()) if k != right) print('{} r{} | {} | ✔ {} | ✗ {}'.format(title, number, data.get('question'), options.get(right), wrong)) if args.explanations: print(' → ' + ' '.join(str(data.get('explanation', '')).split())) return 0 def cmd_golden(args): """Record or verify the golden runs (scripts/game_harness/golden.py).""" import golden import session setup(args, headless=not args.live) names = args.scenario or sorted(golden.SCENARIOS) kind = 'live' if args.live else 'headless' failed = 0 for name in names: demo = golden.SCENARIOS[name][0] out_dir = session.new_run_dir('golden-{}-{}'.format(name, kind)) if args.live: driver = session.LiveGame() driver.open(session.live_url(demo)) else: subject = session.subject_from_args(demo_meta=demo) driver = session.HeadlessGame(out_dir, subject, 'demo', design='1', pace=0.0) frames = (golden.frames_dir(name) if args.mode == 'record' else os.path.join(out_dir, 'frames')) \ if args.live else None if frames: os.makedirs(frames, exist_ok=True) try: steps = golden.play(driver, name, frames_dir=frames) finally: driver.close() if args.mode == 'record': golden.save(name, kind, steps) print('recorded {} ({}, {} steps){}'.format(name, kind, len(steps), ' + frames' if frames else '')) continue diffs = golden.compare_steps(golden.load(name, kind), steps) frame_issues = [] if frames: for record in steps: if record.get('frame'): share, peak = golden.frame_diff(os.path.join(golden.frames_dir(name), record['frame']), os.path.join(frames, record['frame'])) if share > golden.FRAME_TOLERANCE: frame_issues.append((record['frame'], share, peak)) ok = not diffs and not frame_issues failed += 0 if ok else 1 print('{} {} ({}, {} steps)'.format('OK ' if ok else 'FAIL', name, kind, len(steps))) for step, what, want, got in diffs[:args.limit]: print(' {:<14} {:<28} {!r} -> {!r}'.format(step, what, str(want)[:60], str(got)[:60])) for frame, share, peak in frame_issues: print(' frame {} differs on {:.2%} of the card (max {})'.format(frame, share, peak)) if frame_issues: print(' frames: ' + frames) return 1 if failed else 0 def cmd_compare(args): import report report.compare(args.a, args.b) return 0 def cmd_models(args): setup(args, headless=False) from wcs.ai import model_catalog from wcs.ai.UsageTracker import get_pricing current = SETTINGS.get('ai_provider') for provider, models in model_catalog.MODELS.items(): if provider in ('openai_tts', 'elevenlabs'): continue chosen = SETTINGS.get('{}_model'.format(provider)) print('{}{}'.format(provider, ' (active)' if provider == current else '')) for model in models: price = get_pricing(provider, model) print(' {}{:<32} in ${:<6} out ${:<6}'.format('*' if model == chosen else ' ', model, price.get('input', '-'), price.get('output', '-'))) return 0 def cmd_selftest(args): """Offline: the demo game end to end headless, plus a lost game with all lifelines.""" code = 0 for extra in ([], ['--policy', 'fail-at-5', '--lifelines', '5050:2,tip:3,audience:4', '--rules']): command = [sys.executable, os.path.abspath(__file__), 'run', '--demo', args.demo, '--label', 'selftest'] + extra print('$ game.py ' + ' '.join(command[2:])) code |= subprocess.call(command) print('selftest', 'OK' if code == 0 else 'FAILED') return code # --------------------------------------------------------------------------- def main(argv=None): parser = argparse.ArgumentParser(description=__doc__.split('\n\n')[0]) sub = parser.add_subparsers(dest='command', required=True) run = sub.add_parser('run', help='play one game and check it') add_subject(run) add_settings(run) run.add_argument('--policy', default='correct', help='correct | fail-at-N | random | script:ABCD...') run.add_argument('--lifelines', default='', help='5050:N,tip:N,audience:N') run.add_argument('--rules', action='store_true', help='visit the rules screen first') run.add_argument('--seed', type=int) run.add_argument('--timeout', type=float, default=180.0, help='per step, seconds') run.add_argument('--live', action='store_true', help='the game in the running Kodi') run.add_argument('--input', choices=('act', 'click'), default='act', help='live: click = real Kodi SendClick') run.add_argument('--shots', action='store_true', help='live: screenshot after every step') run.add_argument('--keep-open', action='store_true', help='live: leave the game open') run.add_argument('--out') run.add_argument('--label', default='') run.add_argument('--quiet', action='store_true') run.set_defaults(func=cmd_run) review = sub.add_parser('story-review', help='one real Story AI preparation and local play, no voice') add_subject(review) add_settings(review) review.add_argument('--out') review.add_argument('--capture-content', action='store_true', help='explicitly save this test prompt/response for manual fact review; never gameplay history') review.add_argument('--output-tokens', type=int, help='explicit benchmark output budget; production default otherwise') review.add_argument('--request-timeout', type=float, help='explicit benchmark request timeout in seconds') review.set_defaults(func=cmd_story_review, mode='story') suite = sub.add_parser('suite', help='one headless game per demo title') add_settings(suite) suite.add_argument('--titles', default='all', help='all | id,id,...') suite.add_argument('--source', choices=('ai', 'demo'), default='ai') suite.add_argument('--jobs', type=int, default=3) suite.add_argument('--policy', default='correct') suite.add_argument('--lifelines', default='') suite.add_argument('--seed', type=int) suite.add_argument('--label', default='') suite.set_defaults(func=cmd_suite) serve = sub.add_parser('serve', help='keep a headless game running for state/act/... commands') add_subject(serve) add_settings(serve) serve.add_argument('--dir') serve.add_argument('--idle', type=float, default=1800.0, help='stop after so many seconds without commands') serve.set_defaults(func=cmd_serve) for name, help_text in (('state', 'show the game state'), ('act', 'act on intents (start, B, 5050, next ...)'), ('click', 'live: real Kodi click on intents'), ('back', 'Back key'), ('wait', 'wait until the game waits for the player'), ('chat', 'system prompt + chat'), ('trace', 'trace events'), ('last', 'last LLM call'), ('call', 'call a controller method'), ('set', 'set controller attributes'), ('quit', 'stop serve / close the live game')): channel = sub.add_parser(name, help=help_text) channel.add_argument('--live', action='store_true') channel.add_argument('--dir', help='serve folder (default out/serve)') channel.add_argument('--timeout', type=float, default=180.0) channel.add_argument('--json', action='store_true') if name == 'state': channel.add_argument('--hide', action='store_true', help='do not mark the correct answer') if name in ('act', 'click'): channel.add_argument('intents', nargs='+') if name in ('act', 'click', 'back'): channel.add_argument('--no-wait', action='store_true') if name == 'wait': channel.add_argument('--screen', action='append') if name == 'trace': channel.add_argument('--kind', help='prefix: llm, http, content, state, command') channel.add_argument('--last', type=int, default=40) if name == 'call': channel.add_argument('method') channel.add_argument('args', nargs='*') if name == 'set': channel.add_argument('pairs', nargs='+', metavar='ATTR=JSON') channel.set_defaults(func=cmd_channel) opener = sub.add_parser('open', help='open a game in the running Kodi') add_subject(opener) opener.set_defaults(func=cmd_open) repl = sub.add_parser('repl', help='interactive play') add_subject(repl) add_settings(repl) repl.add_argument('--live', action='store_true') repl.add_argument('--timeout', type=float, default=180.0) repl.set_defaults(func=cmd_repl) prompts = sub.add_parser('prompts', help='write the prompts the game would send') add_subject(prompts) add_settings(prompts) prompts.add_argument('--out') prompts.add_argument('--show', action='store_true') prompts.set_defaults(func=cmd_prompts) resend = sub.add_parser('resend', help='send a recorded LLM request again') resend.add_argument('run', help='run folder or trace.jsonl') resend.add_argument('--call', type=int) resend.add_argument('--n', type=int, default=1) add_settings(resend) resend.set_defaults(func=cmd_resend) rep = sub.add_parser('report', help='rebuild report.md of a run') rep.add_argument('run') rep.set_defaults(func=cmd_report) cmp_parser = sub.add_parser('compare', help='compare two runs or suites') cmp_parser.add_argument('a') cmp_parser.add_argument('b') cmp_parser.set_defaults(func=cmd_compare) questions = sub.add_parser('questions', help='list the questions of a run or suite') questions.add_argument('run') questions.add_argument('--from-round', type=int, default=1) questions.add_argument('--explanations', action='store_true') questions.set_defaults(func=cmd_questions) gold = sub.add_parser('golden', help='record / verify the golden runs of the game (refactor safety net)') gold.add_argument('mode', choices=('record', 'verify')) gold.add_argument('--scenario', action='append') gold.add_argument('--live', action='store_true', help='in the running Kodi, with frames') gold.add_argument('--limit', type=int, default=30, help='differences printed per scenario') gold.set_defaults(func=cmd_golden) models = sub.add_parser('models', help='selectable models and prices') models.set_defaults(func=cmd_models) selftest = sub.add_parser('selftest', help='offline check of the harness (demo content)') selftest.add_argument('--demo', default='matrix') selftest.set_defaults(func=cmd_selftest) args = parser.parse_args(argv) if args.command in ('run', 'serve', 'repl', 'prompts', 'open', 'story-review') and not ( args.demo or args.meta or args.title) and not (args.command == 'repl' and args.live): parser.error('give --demo ID, --meta ID or --title TITLE') return args.func(args) if __name__ == '__main__': sys.exit(main())