refactor: bind the live WanGP instance behind a generation wall

app/_launch_runtime.py

@@ -115,6 +115,8 @@
 # Now safe to import wgp - all module-level code will run with patched argv
 print("[HocusPocus Lab] Importing WanGP engine...")
 import wgp
+from services.generation import bind_wgp
+bind_wgp(wgp)
 from services import model3d_service, minimax_h3_service, minimax_image_service
 from services import debug_trace
 from routers.lan_auth import create_lan_auth_router
@@ -35110,7 +35112,7 @@ def rejoin_clips(body: dict):
     Body: { group_id: str, audio_file?: str }
     Finds all clip files matching the group_id and re-concatenates them.
     """
-    from wgp import concatenate_multi_clip_videos
+    from services.mix_concat import concatenate_multi_clip_videos
     group_id = body.get("group_id")
     if not group_id:
         raise HTTPException(status_code=400, detail="group_id is required")

app/services/alternative_songs.py

@@ -15,7 +15,7 @@
 import uuid
 from typing import Any, Callable
 
-from services.mix_concat import probe_duration_seconds
+from services.mix_concat import concatenate_multi_clip_videos, probe_duration_seconds
 from services.output_result_kind import classify_output_result_kind
 from services.video_editor import probe_audio
 
@@ -348,10 +348,6 @@ def remount_clips(
     if not planned:
         raise ValueError("No clips were planned for this remount.")
     paths = [str(item["path"]) for item in planned]
-    try:
-        from wgp import concatenate_multi_clip_videos
-    except ImportError:
-        from app.wgp import concatenate_multi_clip_videos  # type: ignore
     ok = concatenate_multi_clip_videos(
         paths,
         output_path,

app/services/director/prompt_polish.py

@@ -1117,8 +1117,8 @@ def load_lora_guides(video_loras: list[str] = None, image_loras: list[str] = Non
         if not loras or not model_type:
             continue
         try:
-            import wgp
-            lora_dir = wgp.get_lora_dir(model_type)
+            from services.generation import get_wgp
+            lora_dir = get_wgp().get_lora_dir(model_type)
         except Exception:
             continue
         for lora_name in loras:
@@ -1528,9 +1528,9 @@ def _build_lora_hints(loras: list[str], model_type: str) -> str:
         if not loras or not model_type:
             return ""
         try:
-            import wgp
+            from services.generation import get_wgp
             import json as _json
-            lora_dir = wgp.get_lora_dir(model_type)
+            lora_dir = get_wgp().get_lora_dir(model_type)
             trigger_lines: list[str] = []
             for lora_name in loras:
                 sidecar_path = os.path.join(lora_dir, os.path.splitext(lora_name)[0] + ".civitai.json")

app/services/director_pipeline.py

@@ -8680,8 +8680,8 @@ def _publish_parallel_clip(index: int, image_name: str, keyframes: list[str]) ->
         _oom_info = None
         try:
             from services.oom_detect import detect_oom
-            import wgp as _wgp_mod
-            _coef = float(_wgp_mod.server_config.get("vram_safety_coefficient", 0.80))
+            from services.generation import RuntimeConfig
+            _coef = float(RuntimeConfig.get("vram_safety_coefficient", 0.80))
             _oom_info = detect_oom(e, _coef)
         except Exception:
             pass  # Never fail a failure handler

app/services/enhance_guides.py

@@ -97,12 +97,11 @@ def get_enhance_guide(model_type: str, generation_mode: str, has_images: bool =
         Guide text for the LLM system prompt.
     """
     # 1. Per-model overrides via model_def.
-    # Lazy-import wgp + the shared loader because enhance_guides.py is
-    # imported early during server startup, before wgp's full model
-    # definitions are populated. Lazy keeps the import cheap.
+    # Read the bound WanGP catalog at call time: this module is imported
+    # during startup, before launch has populated model definitions.
     inline_delta = ""
     try:
-        from wgp import get_model_def
+        from services.generation import get_model_def
         md = get_model_def(model_type)
         if md:
             # 1a. Full-guide FILE override — a complete, standalone enhancer

app/services/generation/__init__.py

@@ -0,0 +1,17 @@
+"""HocusPocus generation wall around the live WanGP instance."""
+
+from .runtime import (
+    ModelCatalog,
+    RuntimeConfig,
+    bind_wgp,
+    get_model_def,
+    get_wgp,
+)
+
+__all__ = [
+    "ModelCatalog",
+    "RuntimeConfig",
+    "bind_wgp",
+    "get_model_def",
+    "get_wgp",
+]

app/services/generation/runtime.py

@@ -0,0 +1,55 @@
+"""Bound WanGP runtime. Launch imports wgp once and calls bind_wgp(wgp).
+
+This module never imports WanGP. Consumers go through get_wgp() or the thin
+ports below; they must not `import wgp` themselves.
+"""
+from __future__ import annotations
+
+from typing import Any
+
+_UNBOUND_MESSAGE = (
+    "generation.bind_wgp() was not called; launch bootstrap must import wgp first"
+)
+
+_wgp = None
+
+
+def bind_wgp(module) -> None:
+    global _wgp
+    _wgp = module
+
+
+def get_wgp():
+    if _wgp is None:
+        raise RuntimeError(_UNBOUND_MESSAGE)
+    return _wgp
+
+
+class ModelCatalog:
+    """Typed reads of the live WanGP model registry."""
+
+    @staticmethod
+    def get_model_def(model_type: str) -> Any:
+        return get_wgp().get_model_def(model_type)
+
+
+class RuntimeConfig:
+    """Typed reads of the live WanGP server_config mapping."""
+
+    @staticmethod
+    def mapping() -> dict:
+        config = getattr(get_wgp(), "server_config", None)
+        return config if isinstance(config, dict) else {}
+
+    @classmethod
+    def get(cls, key: str, default: Any = None) -> Any:
+        return cls.mapping().get(key, default)
+
+    @classmethod
+    def services(cls) -> dict:
+        services = cls.get("services") or {}
+        return services if isinstance(services, dict) else {}
+
+
+def get_model_def(model_type: str) -> Any:
+    return ModelCatalog.get_model_def(model_type)

app/services/llm_service.py

@@ -3475,13 +3475,13 @@ def enhance_prompt(
 
         # Look up model-specific enhancer prompts (Scenema, Kugel, Qwen3-TTS,
         # Index-TTS2, Chatterbox, IndexTTS2 all set these on their model_def).
-        # Lazy import keeps llm_service.py importable in environments where
-        # wgp.py is unavailable (e.g. lightweight tooling, tests).
+        # The bound catalog is read at call time so llm_service stays importable
+        # in lightweight tooling and tests that never bootstrap WanGP.
         model_specific_monologue = None
         model_specific_dialogue = None
         if model_type:
             try:
-                from wgp import get_model_def
+                from services.generation import get_model_def
                 md = get_model_def(model_type)
                 if md:
                     model_specific_monologue = md.get("text_prompt_enhancer_instructions")

app/services/mix_concat.py

@@ -371,3 +371,29 @@ def build_hard_concat_filter(
         + f"concat=n={count}:v=1:a=1[outv][outa]"
     )
     return filter_str, True
+
+
+def concatenate_multi_clip_videos(
+    clip_paths,
+    output_path,
+    audio_path=None,
+    audio_start_sec=0.0,
+    abort_callback=None,
+    pad_audio=False,
+    audio_duration_sec=None,
+):
+    """Join clips through the bound WanGP helper. This is a media port.
+
+    Concatenation is not part of the generation wall; the wall only supplies
+    the already-imported live instance.
+    """
+    from services.generation import get_wgp
+    return get_wgp().concatenate_multi_clip_videos(
+        clip_paths,
+        output_path,
+        audio_path,
+        audio_start_sec=audio_start_sec,
+        abort_callback=abort_callback,
+        pad_audio=pad_audio,
+        audio_duration_sec=audio_duration_sec,
+    )

app/services/model3d_service.py

@@ -522,26 +522,26 @@ def _prune_finished_jobs_locked() -> None:
 
 def _minimax_api_key() -> str:
     try:
-        import wgp
+        from services.generation import RuntimeConfig
         from .provider_profile import resolve_minimax_key
-        return resolve_minimax_key(wgp.server_config.get("services") or {}, "image")
+        return resolve_minimax_key(RuntimeConfig.services(), "image")
     except Exception:
         return ""
 
 
 def _services() -> dict:
     try:
-        import wgp
-        return wgp.server_config.get("services") or {}
+        from services.generation import RuntimeConfig
+        return RuntimeConfig.services()
     except Exception:
         return {}
 
 
 def _active_profile() -> dict:
     try:
-        import wgp
+        from services.generation import RuntimeConfig
         from .provider_profile import alias_model3d_provider
-        raw = wgp.server_config.get("maestro_production_profile") or {}
+        raw = RuntimeConfig.get("maestro_production_profile") or {}
         image = raw.get("image") if isinstance(raw.get("image"), dict) else {}
         model3d = raw.get("model3d") if isinstance(raw.get("model3d"), dict) else {}
         return {

app/shared/api.py

@@ -4,7 +4,6 @@
 
 import contextlib
 import copy
-import importlib
 import inspect
 import io
 import json
@@ -871,7 +870,14 @@ def _ensure_runtime(self) -> _WanGPRuntime:
                 sys.path.insert(0, str(self._root))
 
             with _pushd(self._root), _temporary_argv(argv):
-                module = importlib.import_module("wgp")
+                from services.generation import bind_wgp, get_wgp
+                try:
+                    module = get_wgp()
+                except RuntimeError:
+                    module = sys.modules.get("wgp")
+                    if module is None:
+                        raise
+                    bind_wgp(module)
                 module_root = Path(module.__file__).resolve().parent
                 if module_root != self._root:
                     raise RuntimeError(f"WanGP module already loaded from {module_root}, expected {self._root}")

app/shared/magic_mask.py

@@ -14,8 +14,8 @@
 
 # Ported from upstream WanGP (v12.x). Upstream reads media through its
 # virtual-media layer and get_resampled_video_transparent; Maestro has
-# neither, so videos are decoded with wgp.get_resampled_video instead
-# (masks never need an alpha channel).
+# neither, so videos are decoded with the bound get_resampled_video
+# helper instead (masks never need an alpha channel).
 
 PROCESS_ID = "magic_mask"
 PROCESS_NAME = "Magic Mask"
@@ -74,8 +74,8 @@ def _video_to_numpy(video_path, max_time_seconds=None):
     frame_count = int(frame_count or 1)
     if max_time_seconds is not None:
         frame_count = min(frame_count, max(1, int(round(float(fps) * float(max_time_seconds)))))
-    from wgp import get_resampled_video
-    frames = get_resampled_video(video_path, 0, frame_count, fps, bridge="torch")
+    from services.generation import get_wgp
+    frames = get_wgp().get_resampled_video(video_path, 0, frame_count, fps, bridge="torch")
     if torch.is_tensor(frames):
         frames = frames.detach().cpu().numpy()
     elif hasattr(frames, "asnumpy"):
@@ -101,11 +101,11 @@ def _ensure_sam3_assets():
     not exist on a fresh install — the detection pre-step then died with
     FileNotFoundError before anything could download. process_files_def
     is existence-checked, so this is a few isfile() calls once the assets
-    are in place. Lazy wgp import: this module loads inside wgp's own
-    import cycle, but every SAM3 run happens long after wgp is up.
+    are in place. This module loads inside wgp's own import cycle, so the
+    bound runtime is read at call time — every SAM3 run happens after bind.
     """
-    import wgp
-    wgp.process_files_def(**query_download_def())
+    from services.generation import get_wgp
+    get_wgp().process_files_def(**query_download_def())
 
 
 def _run_sam3(

docs/development/ARCHITECTURE_FOUNDATION.md

@@ -36,9 +36,17 @@ facade. These are measured values, not permanent targets.
 
 `tests/test_architecture_contracts.py` parses first-party Python with AST and
 names every currently tolerated `wgp` import by file, enclosing symbol and
-statement. Upstream/vendor trees are excluded explicitly. The allowlist may
-only shrink during the WanGP wall step; `app/services/generation/` is reserved
-for the future bound runtime facade.
+statement. Upstream/vendor trees are excluded explicitly.
+
+The WanGP wall lives in `app/services/generation/`. Launch imports `wgp` once
+after the argv patch and calls `bind_wgp(wgp)`. Consumers read that live
+instance through `get_wgp()`, `ModelCatalog`, or `RuntimeConfig`. The wall
+must not reimport WanGP. Video concat stays in `services.mix_concat` and only
+uses `get_wgp()` to reach the already-bound helper.
+
+The remaining first-party allowlist is the launch bootstrap import. New
+static or dynamic `wgp` imports outside `app/models/**` and other vendor
+trees fail the gate.
 
 Run the contracts with:
 

tests/test_architecture_contracts.py

@@ -15,19 +15,6 @@
 
 WGP_ALLOWLIST = {
     ("app/_launch_runtime.py", "<module>", "import wgp"),
-    ("app/_launch_runtime.py", "rejoin_clips", "from wgp import concatenate_multi_clip_videos"),
-    ("app/services/alternative_songs.py", "remount_clips", "from wgp import concatenate_multi_clip_videos"),
-    ("app/services/director/prompt_polish.py", "load_lora_guides", "import wgp"),
-    ("app/services/director/prompt_polish.py", "polish_prompts_third_pass._build_lora_hints", "import wgp"),
-    ("app/services/director_pipeline.py", "_run_pipeline", "import wgp as _wgp_mod"),
-    ("app/services/enhance_guides.py", "get_enhance_guide", "from wgp import get_model_def"),
-    ("app/services/llm_service.py", "enhance_prompt", "from wgp import get_model_def"),
-    ("app/services/model3d_service.py", "_active_profile", "import wgp"),
-    ("app/services/model3d_service.py", "_minimax_api_key", "import wgp"),
-    ("app/services/model3d_service.py", "_services", "import wgp"),
-    ("app/shared/api.py", "WanGPSession._ensure_runtime", "importlib.import_module('wgp')"),
-    ("app/shared/magic_mask.py", "_ensure_sam3_assets", "import wgp"),
-    ("app/shared/magic_mask.py", "_video_to_numpy", "from wgp import get_resampled_video"),
 }
 
 IGNORED_WGP_TREES = (
@@ -146,8 +133,40 @@ def test_wgp_import_detector_names_static_and_dynamic_forms() -> None:
 def test_first_party_wgp_imports_are_named_and_cannot_grow() -> None:
     found = _wgp_imports()
     generation_imports = {item for item in found if item[0].startswith("app/services/generation/")}
-    assert found - generation_imports == WGP_ALLOWLIST, (
-        "First-party WanGP imports changed. Add no new site; Step 1 must remove named entries "
-        f"from the allowlist. Added={sorted(found - generation_imports - WGP_ALLOWLIST)!r}, "
+    assert generation_imports == set(), (
+        "The WanGP wall must bind the live bootstrap instance and must not "
+        f"reimport wgp. found={sorted(generation_imports)!r}"
+    )
+    assert found == WGP_ALLOWLIST, (
+        "First-party WanGP imports may only be the launch bootstrap. "
+        f"Added={sorted(found - WGP_ALLOWLIST)!r}, "
         f"removed={sorted(WGP_ALLOWLIST - found)!r}"
     )
+
+
+def test_launch_binds_live_wgp_immediately_after_bootstrap_import() -> None:
+    tree = ast.parse(
+        (ROOT / "app" / "_launch_runtime.py").read_text(encoding="utf-8"),
+        filename="app/_launch_runtime.py",
+    )
+    events: list[str] = []
+    for node in tree.body:
+        if isinstance(node, ast.Import) and any(alias.name == "wgp" for alias in node.names):
+            events.append("import_wgp")
+        elif (
+            isinstance(node, ast.ImportFrom)
+            and node.module == "services.generation"
+            and any(alias.name == "bind_wgp" for alias in node.names)
+        ):
+            events.append("import_bind_wgp")
+        elif (
+            isinstance(node, ast.Expr)
+            and isinstance(node.value, ast.Call)
+            and isinstance(node.value.func, ast.Name)
+            and node.value.func.id == "bind_wgp"
+            and node.value.args
+            and isinstance(node.value.args[0], ast.Name)
+            and node.value.args[0].id == "wgp"
+        ):
+            events.append("bind_wgp")
+    assert events == ["import_wgp", "import_bind_wgp", "bind_wgp"]