From 84c9e4d315642e166c38a6bd6c9932608e4723bf Mon Sep 17 00:00:00 2001
From: IAnMove <216241348+IAnMove@users.noreply.github.com>
Date: Tue, 1 Sep 2026 02:04:08 +0200
Subject: [PATCH] refactor: bind the live WanGP instance behind a generation
 wall

Launch still imports wgp once after the argv patch, then bind_wgp(wgp).
HocusPocus consumers now read that singleton through get_wgp(), ModelCatalog
or RuntimeConfig. Video concat stays in mix_concat. The architecture allowlist
shrinks to the bootstrap import; the wall does not reimport WanGP.
---
 app/_launch_runtime.py                      |  4 +-
 app/services/alternative_songs.py           |  6 +--
 app/services/director/prompt_polish.py      |  8 +--
 app/services/director_pipeline.py           |  4 +-
 app/services/enhance_guides.py              |  7 ++-
 app/services/generation/__init__.py         | 17 +++++++
 app/services/generation/runtime.py          | 55 ++++++++++++++++++++
 app/services/llm_service.py                 |  6 +--
 app/services/mix_concat.py                  | 26 ++++++++++
 app/services/model3d_service.py             | 12 ++---
 app/shared/api.py                           | 10 +++-
 app/shared/magic_mask.py                    | 16 +++---
 docs/development/ARCHITECTURE_FOUNDATION.md | 14 ++++--
 tests/test_architecture_contracts.py        | 51 +++++++++++++------
 tests/test_generation_runtime.py            | 56 +++++++++++++++++++++
 tests/test_mix_concat.py                    | 37 ++++++++++++++
 16 files changed, 275 insertions(+), 54 deletions(-)
 create mode 100644 app/services/generation/__init__.py
 create mode 100644 app/services/generation/runtime.py
 create mode 100644 tests/test_generation_runtime.py

diff --git a/app/_launch_runtime.py b/app/_launch_runtime.py
index 7cdb8990..1c7305a5 100644
--- a/app/_launch_runtime.py
+++ b/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")
diff --git a/app/services/alternative_songs.py b/app/services/alternative_songs.py
index a6831263..c208ac0f 100644
--- a/app/services/alternative_songs.py
+++ b/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,
diff --git a/app/services/director/prompt_polish.py b/app/services/director/prompt_polish.py
index 2e543f3d..0d4c4522 100644
--- a/app/services/director/prompt_polish.py
+++ b/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")
diff --git a/app/services/director_pipeline.py b/app/services/director_pipeline.py
index bae203dd..695a4f28 100644
--- a/app/services/director_pipeline.py
+++ b/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
diff --git a/app/services/enhance_guides.py b/app/services/enhance_guides.py
index 8b2a24b2..ac4c6c97 100644
--- a/app/services/enhance_guides.py
+++ b/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
diff --git a/app/services/generation/__init__.py b/app/services/generation/__init__.py
new file mode 100644
index 00000000..5a8fc1e1
--- /dev/null
+++ b/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",
+]
diff --git a/app/services/generation/runtime.py b/app/services/generation/runtime.py
new file mode 100644
index 00000000..0aeb699d
--- /dev/null
+++ b/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)
diff --git a/app/services/llm_service.py b/app/services/llm_service.py
index 275ad3ed..d7687805 100644
--- a/app/services/llm_service.py
+++ b/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")
diff --git a/app/services/mix_concat.py b/app/services/mix_concat.py
index e0252d0f..e5e98af8 100644
--- a/app/services/mix_concat.py
+++ b/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,
+    )
diff --git a/app/services/model3d_service.py b/app/services/model3d_service.py
index 455f1ac6..5cd1f455 100644
--- a/app/services/model3d_service.py
+++ b/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 {
diff --git a/app/shared/api.py b/app/shared/api.py
index 506ae5a3..be830e86 100644
--- a/app/shared/api.py
+++ b/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}")
diff --git a/app/shared/magic_mask.py b/app/shared/magic_mask.py
index 2b57a4a3..e0b24836 100644
--- a/app/shared/magic_mask.py
+++ b/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(
diff --git a/docs/development/ARCHITECTURE_FOUNDATION.md b/docs/development/ARCHITECTURE_FOUNDATION.md
index 3899f482..57ae185c 100644
--- a/docs/development/ARCHITECTURE_FOUNDATION.md
+++ b/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:
 
diff --git a/tests/test_architecture_contracts.py b/tests/test_architecture_contracts.py
index 01342ad5..cca8fd55 100644
--- a/tests/test_architecture_contracts.py
+++ b/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"]
diff --git a/tests/test_generation_runtime.py b/tests/test_generation_runtime.py
new file mode 100644
index 00000000..0455e8cd
--- /dev/null
+++ b/tests/test_generation_runtime.py
@@ -0,0 +1,56 @@
+"""WanGP generation wall: bind once, never reimport."""
+from __future__ import annotations
+
+import sys
+from types import SimpleNamespace
+
+import pytest
+
+from services.generation import (
+    ModelCatalog,
+    RuntimeConfig,
+    bind_wgp,
+    get_model_def,
+    get_wgp,
+)
+from services.generation import runtime as generation_runtime
+
+
+@pytest.fixture
+def isolated_wgp():
+    previous = generation_runtime._wgp
+    bind_wgp(None)
+    try:
+        yield
+    finally:
+        generation_runtime._wgp = previous
+
+
+def test_get_wgp_raises_before_bind(isolated_wgp) -> None:
+    with pytest.raises(RuntimeError, match="generation.bind_wgp\\(\\) was not called"):
+        get_wgp()
+
+
+def test_get_wgp_is_the_bound_sys_modules_singleton(isolated_wgp, monkeypatch) -> None:
+    fake = SimpleNamespace(server_config={"vram_safety_coefficient": 0.5})
+    monkeypatch.setitem(sys.modules, "wgp", fake)
+    bind_wgp(fake)
+    assert get_wgp() is sys.modules["wgp"]
+    assert get_wgp() is fake
+
+
+def test_model_catalog_and_runtime_config_read_the_bound_instance(isolated_wgp) -> None:
+    fake = SimpleNamespace(
+        get_model_def=lambda model_type: {"family": "ltx2", "name": model_type},
+        server_config={
+            "services": {"nsfw_mode": True},
+            "maestro_production_profile": {"image": {"provider": "local"}},
+            "vram_safety_coefficient": 0.72,
+        },
+    )
+    bind_wgp(fake)
+    assert get_model_def("ltx2_22B") == {"family": "ltx2", "name": "ltx2_22B"}
+    assert ModelCatalog.get_model_def("ltx2_22B")["family"] == "ltx2"
+    assert RuntimeConfig.get("vram_safety_coefficient", 0.80) == 0.72
+    assert RuntimeConfig.services() == {"nsfw_mode": True}
+    assert RuntimeConfig.get("maestro_production_profile")["image"]["provider"] == "local"
diff --git a/tests/test_mix_concat.py b/tests/test_mix_concat.py
index 51b82c5f..e8bc60db 100644
--- a/tests/test_mix_concat.py
+++ b/tests/test_mix_concat.py
@@ -4,9 +4,14 @@
 
 import pytest
 
+from types import SimpleNamespace
+
+from services.generation import bind_wgp, get_wgp
+from services.generation import runtime as generation_runtime
 from app.services.mix_concat import (
     build_hard_concat_filter,
     build_hold_crossfade_filter,
+    concatenate_multi_clip_videos,
     hold_crossfade_output_seconds,
     probe_audio_flags,
     probe_has_audio,
@@ -14,6 +19,38 @@
 )
 
 
+def test_concatenate_port_delegates_to_bound_wgp():
+    calls = []
+    previous = generation_runtime._wgp
+
+    def fake_concat(*args, **kwargs):
+        calls.append((args, kwargs))
+        return True
+
+    bind_wgp(SimpleNamespace(concatenate_multi_clip_videos=fake_concat))
+    try:
+        assert get_wgp().concatenate_multi_clip_videos is fake_concat
+        assert concatenate_multi_clip_videos(
+            ["a.mp4", "b.mp4"],
+            "out.mp4",
+            "song.wav",
+            abort_callback=None,
+        ) is True
+    finally:
+        generation_runtime._wgp = previous
+    assert calls == [
+        (
+            (["a.mp4", "b.mp4"], "out.mp4", "song.wav"),
+            {
+                "audio_start_sec": 0.0,
+                "abort_callback": None,
+                "pad_audio": False,
+                "audio_duration_sec": None,
+            },
+        )
+    ]
+
+
 def test_hold_crossfade_filter_covers_every_clip_and_xfade():
     filter_str, video, audio = build_hold_crossfade_filter([5.0, 5.0, 5.0])
     assert "[0:v]tpad=stop_mode=clone:stop_duration=0.500[v0]" in filter_str