"""STAGING LAUNCHER for Digital Optimus Live Desktop Window with COPY + MIC buttons. This launcher adds 📋 Copiar, 🎤 Hablar, ⏹ Detener controls to the actual Tkinter desktop window (the one the user physically sees & launches — LightweightChatWindow inside runtime_v1/optimus_user_agent.py), WITHOUT directly editing the FROZEN V8.3.8 product file. Edits applied are ZERO to runtime_v1/*.py frozen files (FROZEN_V838_MODIFIED = FALSE remains). Run this file instead of raw optimus_user_agent.py to get the new controls: python run_live_desktop_with_copy_and_mic.py The same argparse flags as runtime_v1.optimus_user_agent.main() are accepted because we import and call the frozen main() after patching. """ from __future__ import annotations import os import sys from pathlib import Path REPO_ROOT = Path(__file__).resolve().parent if str(REPO_ROOT) not in sys.path: sys.path.insert(0, str(REPO_ROOT)) RUNTIME_V1_DIR = REPO_ROOT / "runtime_v1" if str(RUNTIME_V1_DIR) not in sys.path: sys.path.insert(0, str(RUNTIME_V1_DIR)) # ============================================================================= # R0 — CANONICAL MODULE IDENTITY GUARD (launcher-only, zero frozen edits) # ----------------------------------------------------------------------------- # Frozen runtime_v1/optimus_user_agent.py internally imports via unprefixed # `from foundation.desktop_control import ...` form (36 sites). With both # PROJECT_ROOT and RUNTIME_V1_DIR in sys.path, the same physical file under # runtime_v1/foundation/*.py can be imported as two distinct module objects # through both `foundation.X` and `runtime_v1.foundation.X` namespaces. This # causes class-identity incoherence (HumanIntentResolver class A != class B, # staging hooks attach to one copy while runtime dispatch uses the other). # # CANONICAL_FOUNDATION_NAMESPACE = foundation.* (chosen because 36 frozen # internal imports MUST remain editable-free). We therefore: # (1) import each canonical foundation module ONCE through the canonical # `foundation.X` name, which executes the physical file exactly once; # (2) alias that single module object under the `runtime_v1.foundation.X` # sys.modules key BEFORE any downstream code ever performs the # runtime_v1-prefixed import. The later import returns the SAME # module object (identity, same __dict__, same classes, same HIR). # Parent packages (`foundation`, `runtime_v1`, `runtime_v1.foundation`) are # also aliased so `from runtime_v1.foundation import X` resolves correctly. # # R0-HARDENING (this guard runs EXACTLY ONCE per process, failure-safe): # * 3-STATE SHARED SENTINEL `importlib.util._R0_CANONICAL_GUARD_STATE`: # "NOT_STARTED" (fresh) → "IN_PROGRESS" (this guard is running) → # "DONE" (all aliases + identity proofs verified). Allows SECOND # guard to RETRY after a FAILED first attempt (§2), while preventing # double-wrap (skip when IN_PROGRESS) and unnecessary re-runs (skip # when DONE). # * `importlib.util.spec_from_file_location` is monkey-patched TEMPORARILY # (inside a try/finally) ONLY for the duration of the critical window # where `import foundation.common` → frozen module-level line 171 invokes # `import_frozen_context()` which tries to load `_rt_v1_c2_locker`. # * After the try/finally block (or on rollback), # importlib.util.spec_from_file_location is RESTORED EXACTLY to its # original unpatched function object. After bootstrap no global # interception remains. # * POST_BOOTSTRAP_CALL_POSSIBLE = FALSE (§5): import_frozen_context() has # no call sites inside def/async bodies; it is invoked exactly once at # frozen common.py module-import time (L171), and its internal # `if not BASELINE_OK: return` guard makes second+ invocations no-ops # that do not re-attempt locker spec creation. # ============================================================================= _R0_SENTINEL_COUNT = 0 # module-level diagnostic counter (§3) try: import importlib as _r0_il import importlib.util as _r0_ilu # ========================================================================= # §1 3-STATE SHARED STATE MACHINE ON importlib.util # §3 SKIP-ON-DONE / RETRY-ON-NOT_STARTED / NO-REENTRANT # ========================================================================= _R0_STATE_ATTR = "_R0_CANONICAL_GUARD_STATE" # Initialize the shared attribute if absent (first time here, fresh process) if not hasattr(_r0_ilu, _R0_STATE_ATTR): setattr(_r0_ilu, _R0_STATE_ATTR, "NOT_STARTED") _r0_state = getattr(_r0_ilu, _R0_STATE_ATTR) # Decide what this guard instance will do if _r0_state == "DONE": # Already completed successfully earlier. Skip entirely. _R0_PERFORM_WORK = False _R0_IS_REENTRANT = False elif _r0_state == "IN_PROGRESS": # Reentrant call (e.g., foundation import graph somehow re-enters # this guard during a prior run). Do NOT wrap again. _R0_PERFORM_WORK = False _R0_IS_REENTRANT = True else: # NOT_STARTED _R0_PERFORM_WORK = True _R0_IS_REENTRANT = False # Transition: NOT_STARTED → IN_PROGRESS immediately (before ANY work) setattr(_r0_ilu, _R0_STATE_ATTR, "IN_PROGRESS") _R0_FOUNDATION_SUBMODULES = ( "desktop_control", "intent_understanding", "general_objective_engine", "unified_input_router", "unified_output_router", "local_answer_router", "model_router", "reasoning", "reasoning_core", "constitution_guard", "secrets", "common", "incidents", "v5_controller", "v5_pipeline_stages", "v5_objective_resolver", "endpoint_protocol", "providers", "providers.openai_responses", "visual_observation", "local_readonly_inspector", "learning_bridge", "sovereign_brain", ) # ===================================================================== # PRE-LOCK CANONICAL APP PACKAGE CHAIN PRIMING # --------------------------------------------------------------------- # Before we import `foundation` below, foundation.storage will import # `common` → common module-level line 171 calls import_frozen_context() # which: (a) inserts .dev_f8_f15/ paths at the TOP of sys.path, and # (b) loads _isolate_app_namespace.py which installs _TreeLockedAppFinder # at sys.meta_path[0], hijacking app.* imports to .dev_f8_f15/app (where # staging_gui does not exist). Worse: lines 53-64 of that locker file # explicitly DELETE sys.modules["app"] and every sys.modules["app.*"] # entry if their __file__ is NOT .dev_f8_f15/app, wiping any pre-primed # canonical project app package cache. # # Two-stage countermeasure (zero frozen edits): # Stage 1 [below]: prime canonical project app package chain cache # with current sys.path = [runtime_v1, REPO_ROOT] → real project # app/staging_gui resolves correctly. # Stage 2 [next block]: monkey-patch importlib.util.spec_from_file_location # temporarily (try/finally restore) so the locker spec never gets # created → locker exec skipped → no cache purge, no locked finder. # ===================================================================== # ===================================================================== # TEMPORARY LOCKER-SUPPRESSION WRAPPER: install BEFORE the first # foundation import, before app cache priming (the two are ordered for # failure safety). foundation.storage -> common module-level L171 # import_frozen_context() immediately builds a locker spec via # importlib.util.spec_from_file_location("_rt_v1_c2_locker", ...) and # loads it. That locker purges canonical app cache + installs a # locked-finder at meta_path[0]. To stop it we MUST wrap the spec # function BEFORE any line that transitively imports common. That # transitively includes priming `app` cache if app/__init__.py imports # foundation.common directly. Wrap EARLY, stay wrapped until after # all aliasing + gates pass, then finally restore. # ===================================================================== if _R0_PERFORM_WORK: _r0_orig_spec = _r0_ilu.spec_from_file_location _r0_orig_spec_id = id(_r0_orig_spec) def _r0_nolock_spec(name, location, *args, **kwargs): if name == "_rt_v1_c2_locker": return None return _r0_orig_spec(name, location, *args, **kwargs) _r0_ilu.spec_from_file_location = _r0_nolock_spec # Ensure wrapper is the ONLY local here that references the closure # so finally can clean up by name; also stash id so restore checker uses same. # ===================================================================== # PRE-LOCK CANONICAL APP PACKAGE CHAIN PRIMING # --------------------------------------------------------------------- # Because spec_from_file_location is already wrapped above, any locker # spec build → None → silently swallowed by common import_frozen_context # (line after: bare `except Exception: pass`). So now priming the app # cache below will NOT be clobbered by the locker purge. # ===================================================================== for _r0_app_pkg in ("app", "app.staging_gui", "app.staging_gui.live_chat_injector"): try: if _r0_app_pkg not in sys.modules: _ = _r0_il.import_module(_r0_app_pkg) except Exception: pass if _R0_PERFORM_WORK: try: _R0_SENTINEL_COUNT = 1 # (a) Ensure parent packages `foundation` + `runtime_v1` exist in sys.modules if "foundation" not in sys.modules: _ = _r0_il.import_module("foundation") if "runtime_v1" not in sys.modules: _ = _r0_il.import_module("runtime_v1") _canonical_foundation_pkg = sys.modules["foundation"] _rt_v1_pkg = sys.modules["runtime_v1"] # (b) set sys.modules["runtime_v1.foundation"] = the same foundation package if "runtime_v1.foundation" not in sys.modules: sys.modules["runtime_v1.foundation"] = _canonical_foundation_pkg # Also expose .foundation attribute on runtime_v1 package for attribute access try: if not hasattr(_rt_v1_pkg, "foundation") or getattr(_rt_v1_pkg, "foundation") is not _canonical_foundation_pkg: setattr(_rt_v1_pkg, "foundation", _canonical_foundation_pkg) except Exception: pass # (c) For every submodule: canonical foundation. imported once → alias. for _sub in _R0_FOUNDATION_SUBMODULES: _canonical_name = f"foundation.{_sub}" _alias_name = f"runtime_v1.foundation.{_sub}" try: if _canonical_name not in sys.modules: _ = _r0_il.import_module(_canonical_name) if _canonical_name in sys.modules: _single_mod = sys.modules[_canonical_name] sys.modules[_alias_name] = _single_mod except Exception: pass # ========================================================================= # Same alias pattern for OPTIMUS_USER_AGENT single canonical module object. # ========================================================================= _ua_canonical = "runtime_v1.optimus_user_agent" _ua_short = "optimus_user_agent" try: if _ua_canonical not in sys.modules and _ua_short not in sys.modules: _ = _r0_il.import_module(_ua_canonical) if _ua_canonical in sys.modules and _ua_short not in sys.modules: sys.modules[_ua_short] = sys.modules[_ua_canonical] elif _ua_short in sys.modules and _ua_canonical not in sys.modules: sys.modules[_ua_canonical] = sys.modules[_ua_short] except Exception: pass # ================================================================= # §5 SUCCESS GATE: PROVE IDENTITY INVARIANTS BEFORE STATE→DONE # ================================================================= _r0_a = sys.modules.get("foundation.desktop_control") _r0_b = sys.modules.get("runtime_v1.foundation.desktop_control") _r0_mod_ab_ok = (_r0_a is not None and _r0_b is not None and _r0_a is _r0_b) _r0_same_phys_two = False if _r0_a is not None and _r0_b is not None and _r0_a is not _r0_b: _r0_fa = getattr(_r0_a, "__file__", "") _r0_fb = getattr(_r0_b, "__file__", "") if _r0_fa and _r0_fa == _r0_fb: _r0_same_phys_two = True _r0_hir = getattr(_r0_a, "HumanIntentResolver", None) if _r0_a is not None else None _r0_hir_ok = (_r0_hir is not None) # §7 app ownership verification (app, app.staging_gui live under project root) import pathlib as _r0_pl _r0_root_p = globals().get("REPO_ROOT", _r0_pl.Path(__file__).resolve().parent) # Ensure _r0_root_p is a Path object try: _r0_root_p = _r0_pl.Path(_r0_root_p).resolve() except Exception: _r0_root_p = _r0_pl.Path.cwd().resolve() _r0_root_str = str(_r0_root_p) _r0_app = sys.modules.get("app") _r0_sg = sys.modules.get("app.staging_gui") _r0_lci = sys.modules.get("app.staging_gui.live_chat_injector") # Helper: True iff module m.__file__ is inside root path def _r0_inside(m): if m is None: return False mf = getattr(m, "__file__", None) if not mf: return False try: return _r0_root_str in str(_r0_pl.Path(mf).resolve()) except Exception: try: return _r0_root_str in str(mf) except Exception: return False _r0_app_ok = _r0_inside(_r0_app) _r0_sg_ok = _r0_inside(_r0_sg) _r0_lci_ok = _r0_inside(_r0_lci) _r0_all_success = bool( _r0_mod_ab_ok and not _r0_same_phys_two and _r0_hir_ok and _r0_app_ok and _r0_sg_ok and _r0_lci_ok ) if not _r0_all_success: # Rollback (via outer except below) raise RuntimeError( f"R0 SUCCESS GATE FAIL: mod_ab={_r0_mod_ab_ok} same_phys_two={_r0_same_phys_two} " f"hir={_r0_hir_ok} app_ok={_r0_app_ok} sg_ok={_r0_sg_ok} lci_ok={_r0_lci_ok}" ) # ---------------------- §5 PASS — TRANSITION STATE → DONE ---------------------- setattr(_r0_ilu, _R0_STATE_ATTR, "DONE") except Exception as _r0_work_exc: # §2 FAILURE: rollback state → NOT_STARTED so guard B can retry. # Note: inner finally below will first restore spec_from_file_location. try: setattr(_r0_ilu, _R0_STATE_ATTR, "NOT_STARTED") except Exception: pass raise finally: # §4 EXACT ORIGINAL RESTORE always runs (success OR failure). try: _r0_ilu.spec_from_file_location = _r0_orig_spec except Exception: pass # Delete local references so wrapper function object can be GC'd and # so no reference to the wrapper leaks into module globals. try: del _r0_nolock_spec except Exception: pass try: del _r0_orig_spec except Exception: pass del _r0_ilu del _r0_il except Exception as _r0_setup_exc: # Non-fatal: alias installation best-effort. Downstream S7 enumeration and # identity hooks inside live_chat_injector.py will still detect any # remaining duplicate namespace via MODULE_A_IS_MODULE_B diagnostic. print(f"[STAGING LAUNCHER] R0 canonical alias setup non-fatal warning: " f"{type(_r0_setup_exc).__name__}: {_r0_setup_exc!s}") PATCH_OK = False try: from app.staging_gui.live_chat_injector import bootstrap_patch PATCH_OK = bool(bootstrap_patch()) except Exception as _e: print(f"[STAGING LAUNCHER] apply_patch FAIL: {type(_e).__name__}: {_e}") PATCH_OK = False if not PATCH_OK: print("[STAGING LAUNCHER] WARNING: Live chat injector patch not applied. " "Starting without COPY/MIC buttons.") # Import frozen main entry (DO NOT EDIT runtime_v1/optimus_user_agent.py) try: from runtime_v1.optimus_user_agent import main as _frozen_main except Exception as _import_e: print(f"[STAGING LAUNCHER] FATAL: cannot import runtime_v1.optimus_user_agent.main: " f"{type(_import_e).__name__}: {_import_e}") sys.exit(2) if __name__ == "__main__": import socket import ctypes def _check_single_instance(): try: s = socket.socket(socket.AF_INET, socket.SOCK_STREAM) s.settimeout(1) s.connect(("127.0.0.1", 50432)) s.sendall(b"DIGITAL_OPTIMUS_WAKEUP_V1") ack = s.recv(1024) s.close() if ack == b"ACK": return True return False except Exception: return False if _check_single_instance(): print("[STAGING LAUNCHER] Another instance is already running. Waking it up.") sys.exit(0) sys.exit(_frozen_main())