"""Eager module loading for immutable runtime immunity (Issue #5 Phase 4). Ensures all dependencies (loop, delegation, agents, external packages) are loaded into sys.modules at loop startup, preventing mid-flight disk modifications from triggering lazy-import syntax/EOF errors (B-13). """ from __future__ import annotations import importlib import os import sys from typing import List, Sequence # Framework version locked with framework constants FRAMEWORK_VERSION = "4.8.8" CORE_MODULES: Sequence[str] = ( # Standard library modules "atexit", "argparse", "copy", "glob", "hashlib", "json", "os", "pathlib", "re", "shutil", "signal", "sqlite3", "subprocess", "sys", "tempfile", "time", "typing", # Framework loop modules "lib_py.loop.parser", "lib_py.loop.checkpoint", "lib_py.loop.runner", "lib_py.loop.attribution", "lib_py.loop.autocomplete", "lib_py.loop.evidence", "lib_py.loop.policy", "lib_py.loop.shadow", "lib_py.loop.store", "lib_py.loop.watcher", # Delegation and common modules "lib_py.ints", "lib_py.paths", "lib_py.state", "lib_py.atomic_yaml", "lib_py.workspace_uuid", "lib_py.verify_session", "lib_py.layout", "lib_py.delegation.engine", "lib_py.delegation.finalize", "lib_py.delegation.job_subscriber", "lib_py.delegation.mqtt_common", "lib_py.delegation.notifier", "lib_py.delegation.publish_event", "lib_py.delegation.publisher", "lib_py.delegation.registry", "lib_py.delegation.service", "lib_py.delegation.cli", # Agent adapter registry "lib_py.agents.base", "lib_py.agents.registry", "lib_py.agents.input_region", "lib_py.agents.sanitize", ) OPTIONAL_MODULES: Sequence[str] = ( "yaml", "paho.mqtt", "paho.mqtt.client", ) def verify_framework_version(expected_version: str = FRAMEWORK_VERSION) -> None: """Verifies that MAM_VERSION matches framework version if set.""" env_version = os.environ.get("MAM_VERSION") if env_version and env_version != expected_version: raise RuntimeError( f"MAM_VERSION mismatch: expected {expected_version}, got {env_version} in environment." ) def eager_load( modules: Sequence[str] = CORE_MODULES, optional_modules: Sequence[str] = OPTIONAL_MODULES, ) -> List[str]: """Eagerly imports all modules into sys.modules. Returns the list of successfully loaded module names. """ verify_framework_version() # Ensure PYTHONDONTWRITEBYTECODE is set for isolation sys.dont_write_bytecode = True os.environ["PYTHONDONTWRITEBYTECODE"] = "1" loaded: List[str] = [] for mod_name in modules: if mod_name not in sys.modules: try: importlib.import_module(mod_name) except Exception as exc: raise RuntimeError(f"Failed to eager load {mod_name}: {exc}") from exc loaded.append(mod_name) for opt_mod in optional_modules: if opt_mod not in sys.modules: try: importlib.import_module(opt_mod) loaded.append(opt_mod) except (ImportError, ModuleNotFoundError): pass return loaded def is_eager_loaded(modules: Sequence[str] = CORE_MODULES) -> bool: """Returns True if all required core modules are present in sys.modules.""" return all(mod in sys.modules for mod in modules)