# -*- coding: utf-8 -*- """Game probe -- trace, state and remote control of a running Millionaire game (development only). Shared by the headless harness (scripts/game_harness) and the real game modal in Kodi. In Kodi it is attached only while the dev harness is enabled (the flag file harness/ENABLED, created by scripts/dialog_harness/kodi_live.py); otherwise `attach_if_enabled` returns at once and nothing of the game changes. A probe records into /trace.jsonl: content.* every call of the content source (begin/welcome/question/feedback/tip) with its result or error llm.* every LLM call of the game (system prompt, chat messages, parsed answer, usage, provider/model, ms) http every HTTP request of the addon while the probe runs (body and answer, never headers / API keys) state each change of the game screen (screen, game state, round) and keeps /state.json up to date (see `snapshot`). Commands are JSON files in /cmd/ (see `Bridge`). """ import json import os import threading import time import traceback import uuid LETTERS = 'ABCD' MAX_BODY = 200000 POLL_S = 0.2 _ACTIVE = {'bridge': None} def _jsonable(value, depth=0): if depth > 8: return repr(value) if isinstance(value, (str, int, float, bool)) or value is None: return value if isinstance(value, dict): return {str(k): _jsonable(v, depth + 1) for k, v in value.items()} if isinstance(value, (list, tuple, set)): return [_jsonable(v, depth + 1) for v in value] if hasattr(value, '_asdict'): return _jsonable(value._asdict(), depth + 1) return repr(value) class Trace: """Thread safe event log, in memory and appended to a JSONL file.""" def __init__(self, path=None): self.path = path self.events = [] self.listeners = [] self._lock = threading.Lock() self._t0 = time.time() self._ids = 0 if path: folder = os.path.dirname(path) if folder and not os.path.isdir(folder): os.makedirs(folder) def next_id(self): with self._lock: self._ids += 1 return self._ids def emit(self, kind, **data): with self._lock: event = {'seq': len(self.events) + 1, 't': round(time.time() - self._t0, 3), 'kind': kind, 'thread': threading.current_thread().name} event.update(_jsonable(data)) self.events.append(event) if self.path: try: with open(self.path, 'a', encoding='utf-8') as handle: handle.write(json.dumps(event, ensure_ascii=False) + '\n') except OSError: pass for listener in list(self.listeners): try: listener(event) except Exception: pass return event # --------------------------------------------------------------------------- # instrumentation # --------------------------------------------------------------------------- class Instruments: """Wraps the content source, the LLM entry points and `requests` so every step lands in the trace.""" def __init__(self, trace): self.trace = trace self._undo = [] def install(self, content): self._wrap_content(content) self._wrap_llm() self._wrap_http() return self def uninstall(self): while self._undo: owner, name, original, is_instance = self._undo.pop() try: if is_instance: owner.__dict__.pop(name, None) else: setattr(owner, name, original) except Exception: pass # -- content source ------------------------------------------------------ def _wrap_content(self, content): trace = self.trace for name in ('begin', 'welcome', 'question', 'feedback', 'tip'): original = getattr(content, name, None) if original is None or getattr(original, '_probe', False): continue def make(name, original): def wrapper(*args, **kwargs): call = trace.next_id() trace.emit('content.call', call=call, method=name, source=getattr(content, 'name', ''), args=list(args)) started = time.time() try: result = original(*args, **kwargs) except Exception as exc: trace.emit('content.error', call=call, method=name, ms=_ms(started), error='{}: {}'.format(type(exc).__name__, exc), traceback=traceback.format_exc()) raise extra = {} if name == 'begin': extra['system_prompt'] = getattr(content, 'system_prompt', '') extra['session_id'] = getattr(content, 'session_id', '') history = getattr(content, 'chat_history', None) if history and name in ('welcome', 'question', 'feedback'): users = [m for m in history if m.get('role') == 'user'] extra['user_message'] = users[-1]['content'] if users else '' extra['history_len'] = len(history) trace.emit('content.result', call=call, method=name, ms=_ms(started), result=result, **extra) return result wrapper._probe = True return wrapper setattr(content, name, make(name, original)) self._undo.append((content, name, None, True)) # -- LLM entry points of the game ----------------------------------------- def _wrap_llm(self): try: from wcs.ai import AIHandler except Exception: return # the demo never loads the AI stack trace = self.trace def model_info(): try: from wcs.ai.AIProvider import ai_provider return ai_provider.get_current_provider(), ai_provider.get_current_model() except Exception: return '', '' chat = AIHandler.generate_millionaire_chat if not getattr(chat, '_probe', False): def chat_wrapper(system_prompt, messages, session_id=None, **kwargs): call = trace.next_id() provider, model = model_info() trace.emit('llm.request', call=call, purpose='chat', provider=provider, model=model, system_prompt=system_prompt, messages=[dict(m) for m in messages], options=kwargs) started = time.time() try: data, usage = chat(system_prompt, messages, session_id=session_id, **kwargs) except Exception as exc: trace.emit('llm.error', call=call, purpose='chat', provider=provider, model=model, ms=_ms(started), error='{}: {}'.format(type(exc).__name__, exc), traceback=traceback.format_exc()) raise trace.emit('llm.response', call=call, purpose='chat', provider=provider, model=model, ms=_ms(started), data=data, usage=usage) return data, usage chat_wrapper._probe = True AIHandler.generate_millionaire_chat = chat_wrapper self._undo.append((AIHandler, 'generate_millionaire_chat', chat, False)) tip = AIHandler.generate_director_tip if not getattr(tip, '_probe', False): def tip_wrapper(*args, **kwargs): call = trace.next_id() provider, model = model_info() trace.emit('llm.request', call=call, purpose='tip', provider=provider, model=model, args=list(args), kwargs=kwargs) started = time.time() try: text = tip(*args, **kwargs) except Exception as exc: trace.emit('llm.error', call=call, purpose='tip', provider=provider, model=model, ms=_ms(started), error='{}: {}'.format(type(exc).__name__, exc), traceback=traceback.format_exc()) raise trace.emit('llm.response', call=call, purpose='tip', provider=provider, model=model, ms=_ms(started), data=text) return text tip_wrapper._probe = True AIHandler.generate_director_tip = tip_wrapper self._undo.append((AIHandler, 'generate_director_tip', tip, False)) # -- raw HTTP ------------------------------------------------------------- def _wrap_http(self): try: import requests except Exception: return trace = self.trace for method in ('post', 'get'): original = getattr(requests, method) if getattr(original, '_probe', False): continue def make(method, original): def wrapper(url, *args, **kwargs): call = trace.next_id() body = kwargs.get('json') if body is None and isinstance(kwargs.get('data'), (str, bytes)): body = '<{} bytes>'.format(len(kwargs['data'])) started = time.time() try: response = original(url, *args, **kwargs) except Exception as exc: trace.emit('http', call=call, method=method.upper(), url=_clean_url(url), request=body, ms=_ms(started), error='{}: {}'.format(type(exc).__name__, exc)) raise trace.emit('http', call=call, method=method.upper(), url=_clean_url(url), request=body, ms=_ms(started), status=response.status_code, response=_response_body(response)) return response wrapper._probe = True return wrapper setattr(requests, method, make(method, original)) self._undo.append((requests, method, original, False)) def _ms(started): return int((time.time() - started) * 1000) def _clean_url(url): return str(url).split('?', 1)[0] def _response_body(response): kind = response.headers.get('Content-Type', '') if 'json' in kind: try: return response.json() except ValueError: pass if kind.startswith(('audio/', 'image/', 'application/octet-stream')): return '<{} {} bytes>'.format(kind, len(response.content)) return response.text[:MAX_BODY] # --------------------------------------------------------------------------- # state # --------------------------------------------------------------------------- def snapshot(modal): """Everything worth knowing about the game right now (JSON safe).""" adapter = getattr(modal, '_adapter', None) base = {'t': round(time.time(), 3), 'alive': adapter is not None, 'focus': _focus(modal), 'settings_open': bool(modal.getProperty('WCS.Game.Settings')), 'exit_prompt': bool(modal.getProperty('WCS.Game.Exit'))} if adapter is None: return base if getattr(modal, 'mode', 'millionaire') == 'story': from wcs.games.story.view import FIELDS engine = adapter.engine demo = adapter.source.name == 'demo' scene = engine.scene if engine else None question = ({'question': scene.question, 'options': list(scene.options), 'correct': scene.correct, 'correct_letter': LETTERS[scene.correct], 'explanation': scene.reveal} if scene and demo else None) base.update(mode='story', design='modern', content=adapter.source.name, title=modal.title, media_type=modal.media_type, screen=adapter.screen(), game_state='playing' if adapter.screen() == 'question' else adapter.screen(), round=engine.index + 1 if engine else 0, score=engine.correct if engine else 0, lives=engine.lives if engine else 3, answered=engine.answered if engine else 0, question_count=len(engine.pack.scenes) if engine and engine.pack else 0, question=question, selected=engine.selected if engine else None, countdown=0, autoplay=False, lifelines={}, removed=[], buttons_ready=adapter.screen() != 'loading', view={name: modal.getProperty('WCS.Story.' + name) for name in FIELDS} if demo else {}) base['settled'] = settled(base) return base from wcs.games import game_view game = adapter.controller question = game.current_question or None view = game_view.view(game) view['ButtonsReady'] = 'true' if adapter._buttons_ready else '' content = game.content base.update({ 'design': adapter.design.key, 'content': getattr(content, 'name', ''), 'session_id': getattr(content, 'session_id', ''), 'chat_len': len(getattr(content, 'chat_history', []) or []), 'title': game.title, 'media_type': game.media_type, 'screen': game_view.screen_of(game), 'game_state': game.game_state, 'round': game.current_round, 'score': game.score, 'question': _question(question), 'prefetched': game.prefetched_question is not None, 'prefetch_done': game.prefetch_ready.is_set(), 'selected': getattr(game, 'selected_answer_index', None), 'feedback': game.feedback_data, 'countdown': game.countdown, 'lifelines': {'5050': game.lifeline_5050_used, 'tip': game.lifeline_tip_used, 'audience': game.lifeline_audience_used}, 'removed': [i for i in range(4) if game.lifeline_5050_indices and i not in game.lifeline_5050_indices], 'tip': game.tip_view, 'poll': game.poll_view, 'autoplay': game._autoplay_enabled, 'buttons_ready': adapter._buttons_ready, 'settled': False, 'view': view, }) base['settled'] = settled(base) return base def _focus(modal): try: return modal.getFocusId() except Exception: return None def _question(question): if not question: return None data = dict(question) correct = data.get('correct', 0) data['correct_letter'] = LETTERS[correct] if isinstance(correct, int) and 0 <= correct < 4 else '?' return data def settled(snap): """True when the game waits for the player (nothing is being prepared, no countdown or animation runs).""" if not snap.get('alive'): return True screen, view = snap.get('screen'), snap.get('view', {}) if screen in ('title', 'rules', 'error', 'results'): return True if screen == 'intro': return snap.get('game_state') == 'welcome_screen' and bool(snap.get('prefetched') or snap.get('prefetch_done')) if screen == 'loading': return False if screen == 'question': return (snap.get('game_state') == 'playing' and view.get('TipState') != 'loading' and view.get('PollState') != 'running') if screen in ('feedback', 'gameover', 'win'): return bool(snap.get('buttons_ready')) return False def _brief(snap): return {k: snap.get(k) for k in ('screen', 'game_state', 'round', 'settled', 'focus', 'exit_prompt')} # --------------------------------------------------------------------------- # remote control # --------------------------------------------------------------------------- class _BackAction: def getId(self): return 92 def getButtonCode(self): return 0 def control_of(modal, intent): """Control id of an intent in the modal's design (answer0-3 / A-D, life5050, start, next ...).""" return control_for(modal._design, intent, getattr(modal, 'mode', ''), modal._adapter.screen()) def control_for(design, intent, mode='', screen=None): """The same intent resolution for an attached modal and native clicks from a snapshot.""" intent = normalize_intent(intent) if mode == 'story': if intent.startswith('answer'): return 9790 + int(intent[6:]) if intent == 'start': return 9601 if screen == 'title' else 9781 if intent == 'back': return {'rules': 9782, 'loading': 9783, 'error': 9785}.get(screen, 9642) story_controls = {'next': 9794, 'again': 9794, 'title': 9796, 'home': 9797, 'retry': 9784} if intent in story_controls: return story_controls[intent] for control, name in design.clicks.items(): if name == intent: return control return None def normalize_intent(intent): intent = str(intent).strip() aliases = {'a': 'answer0', 'b': 'answer1', 'c': 'answer2', 'd': 'answer3', '5050': 'life5050', 'tip': 'lifetip', 'audience': 'lifeaudience', 'poll': 'lifeaudience', 'stay': 'exit_stay', 'leave': 'exit_confirm'} return aliases.get(intent.lower(), intent) class Bridge: """Keeps /state.json current, records the trace and runs commands from /cmd/*.json. Command {"id": str, "op": ...}; the answer goes to /res-.json {"id", "ok", "result" | "error"}. act {"intent": "start" | "answer1" | "B" | "life5050" | "next" ...} -> onClick of its control click {"control": 9500} -> onClick(control) back {} -> Back as from the remote close {} -> close the game window props {"names": ["WCS.Game.Screen", ...]} window properties as the XML reads them seed {"value": 1} random.seed (repeatable lifelines) snapshot {} chat {} system prompt + chat history of the AI content call {"method": "show_generation_error", "args": [...]} a controller method (debug) set {"attrs": {"current_round": 5}} controller attributes (debug) """ def __init__(self, modal, folder, trace=None): self.modal = modal self.folder = folder self.cmd_dir = os.path.join(folder, 'cmd') self.trace = trace or Trace(os.path.join(folder, 'trace.jsonl')) self.instruments = None self._stop = threading.Event() self._thread = None self._last_brief = None self._content = None self.id = uuid.uuid4().hex[:12] self._write_lock = threading.Lock() self.owner_path = os.path.join(folder, 'OWNER') for path in (folder, self.cmd_dir): if not os.path.isdir(path): os.makedirs(path) # -- lifecycle ------------------------------------------------------------- def start(self): with open(self.owner_path, 'w') as handle: # one game owns the folder; an older probe stops by itself handle.write(self.id) self.trace.emit('probe.start', folder=self.folder, probe=self.id) self.instrument() self._thread = threading.Thread(target=self._loop, name='game-probe', daemon=True) self._thread.start() return self def instrument(self): """(Re)wrap the content of the current adapter (a new adapter comes with every return to the title).""" adapter = getattr(self.modal, '_adapter', None) if getattr(self.modal, 'mode', '') == 'story': if self.instruments: self.instruments.uninstall() self.instruments = None self._content = None return # Story AI has no persisted content/HTTP archive content = adapter.controller.content if adapter else None if content is None or content is self._content: return if self.instruments: self.instruments.uninstall() self._content = content self.instruments = Instruments(self.trace).install(content) def owns_folder(self): try: with open(self.owner_path) as handle: return handle.read().strip() == self.id except OSError: return True def stop(self): self._stop.set() if self.instruments: self.instruments.uninstall() self.instruments = None if self.owns_folder(): try: self._write_state(snapshot(self.modal)) except Exception: pass self.trace.emit('probe.stop', probe=self.id) # -- loop ------------------------------------------------------------------ def _loop(self): while not self._stop.is_set(): if not self.owns_folder(): self.stop() return try: self.instrument() self.publish() self._commands() except Exception as exc: self.trace.emit('probe.error', error=str(exc), traceback=traceback.format_exc()) self._stop.wait(POLL_S) def publish(self): snap = snapshot(self.modal) brief = _brief(snap) if brief != self._last_brief: self._last_brief = brief self.trace.emit('state', **brief) self._write_state(snap) return snap def _write_state(self, snap): snap['probe'] = self.id path = os.path.join(self.folder, 'state.json') with self._write_lock: # the loop and a command may publish at the same moment tmp = path + '.tmp' with open(tmp, 'w', encoding='utf-8') as handle: json.dump(_jsonable(snap), handle, ensure_ascii=False, indent=1) os.replace(tmp, path) def _commands(self): for name in sorted(os.listdir(self.cmd_dir)): if not name.endswith('.json'): continue path = os.path.join(self.cmd_dir, name) try: with open(path, encoding='utf-8') as handle: command = json.load(handle) except (OSError, ValueError): continue os.remove(path) self.execute(command) def execute(self, command): command_id = str(command.get('id') or uuid.uuid4().hex[:10]) self.trace.emit('command', **command) try: result = {'ok': True, 'result': self.run(command)} except Exception as exc: result = {'ok': False, 'error': '{}: {}'.format(type(exc).__name__, exc), 'traceback': traceback.format_exc()} try: self.publish() # state.json reflects the command before its answer is written except Exception: pass result['id'] = command_id path = os.path.join(self.folder, 'res-{}.json'.format(command_id)) with open(path + '.tmp', 'w', encoding='utf-8') as handle: json.dump(_jsonable(result), handle, ensure_ascii=False, indent=1) os.replace(path + '.tmp', path) return result def run(self, command): op = command.get('op') modal = self.modal if op == 'act': control = control_of(modal, command['intent']) if control is None: raise ValueError('intent {!r} has no control in design {}'.format( command['intent'], modal._design.key)) modal.onClick(control) self.instrument() # a mode change retires HTTP/content recording synchronously return {'control': control} if op == 'click': modal.onClick(int(command['control'])) self.instrument() return {'control': int(command['control'])} if op == 'back': modal.onAction(_BackAction()) return {} if op == 'snapshot': return snapshot(modal) if op == 'props': return {name: modal.getProperty(name) for name in command.get('names', [])} if op == 'seed': import random random.seed(command.get('value')) return {'seed': command.get('value')} if op == 'close': modal.close() return {} adapter = getattr(modal, '_adapter', None) if adapter is None: raise RuntimeError('the game is not running') if getattr(modal, 'mode', '') == 'story': if op == 'chat': return {'system_prompt': '', 'messages': [], 'session_id': ''} raise ValueError('Story debug access uses snapshot and props, not controller mutation') game = adapter.controller if op == 'chat': content = game.content return {'system_prompt': getattr(content, 'system_prompt', ''), 'messages': getattr(content, 'chat_history', []), 'session_id': getattr(content, 'session_id', '')} if op == 'call': value = getattr(game, command['method'])(*command.get('args', []), **command.get('kwargs', {})) return {'value': value} if op == 'set': for key, value in command.get('attrs', {}).items(): setattr(game, key, value) return {'set': sorted(command.get('attrs', {}))} raise ValueError('unknown op: {}'.format(op)) # --------------------------------------------------------------------------- # Kodi side: attach to the real game modal while the dev harness is enabled # --------------------------------------------------------------------------- def _kodi_folder(): import xbmcvfs base = xbmcvfs.translatePath('special://profile/addon_data/plugin.video.milionar/harness/') return base, os.path.join(base, 'game') def attach_if_enabled(modal): """Start a Bridge for the game modal when harness/ENABLED exists (no-op otherwise, never raises).""" try: if _ACTIVE['bridge'] is not None and _ACTIVE['bridge'].modal is modal: _ACTIVE['bridge'].instrument() return _ACTIVE['bridge'] base, folder = _kodi_folder() if not os.path.exists(os.path.join(base, 'ENABLED')): return None detach() if os.path.isdir(folder): for name in os.listdir(folder): path = os.path.join(folder, name) if os.path.isfile(path) and (name.startswith('res-') or name in ('trace.jsonl', 'state.json')): os.remove(path) _ACTIVE['bridge'] = Bridge(modal, folder).start() return _ACTIVE['bridge'] except Exception as exc: try: import xbmc xbmc.log('[Milionář Harness] game probe not attached: {}'.format(exc), xbmc.LOGWARNING) except Exception: pass return None def detach(modal=None): bridge = _ACTIVE['bridge'] if bridge is None or (modal is not None and bridge.modal is not modal): return _ACTIVE['bridge'] = None try: bridge.stop() except Exception: pass