fix: preserve standalone WanGP API bootstrap

app/docs/API.md

@@ -53,7 +53,10 @@ else:
 ## Main Entry Points
 
 - `init(...) -> WanGPSession`
-  - Creates a reusable session and eagerly loads the runtime.
+  - Creates a reusable session and eagerly loads the runtime. If HocusPocus
+    already bound WanGP, it reuses that exact instance; in a standalone Python
+    process it performs the one authorized import and binds it for all later
+    calls.
 - `WanGPSession.submit(source) -> SessionJob`
   - Starts a job from a settings dict, a manifest list, or a saved `.json` / `.zip` file.
 - `WanGPSession.submit_task(settings) -> SessionJob`

app/services/generation/bootstrap.py

@@ -0,0 +1,76 @@
+"""Explicit bootstrap for the documented standalone WanGP Python API.
+
+Normal HocusPocus consumers must use :func:`get_wgp` and never load WanGP.
+This module is the one alternate entrypoint for processes that start through
+``shared.api.init()`` instead of ``_launch_runtime.py``.
+"""
+from __future__ import annotations
+
+import importlib
+import sys
+import threading
+from pathlib import Path
+from typing import Any, Callable
+
+from .runtime import bind_wgp, get_wgp
+
+
+_BOOTSTRAP_LOCK = threading.RLock()
+
+
+def _import_wgp() -> Any:
+    """Keep the exceptional dynamic import named and visible to the gate."""
+    return importlib.import_module("wgp")
+
+
+def _module_root(module: Any) -> Path:
+    module_file = getattr(module, "__file__", None)
+    if not module_file:
+        raise RuntimeError("The loaded WanGP module has no __file__; its root cannot be verified")
+    return Path(module_file).resolve().parent
+
+
+def _require_root(module: Any, expected_root: Path) -> None:
+    actual_root = _module_root(module)
+    if actual_root != expected_root:
+        raise RuntimeError(f"WanGP module already loaded from {actual_root}, expected {expected_root}")
+
+
+def get_or_bootstrap_wgp(
+    *,
+    expected_root: str | Path,
+    importer: Callable[[str], Any] | None = None,
+) -> Any:
+    """Return the singleton, importing it once for a standalone API process.
+
+    The caller owns cwd and argv setup.  ``importer`` exists for tests; the
+    production path uses the single, architecture-gated import above.
+    """
+    expected = Path(expected_root).resolve()
+    with _BOOTSTRAP_LOCK:
+        try:
+            bound = get_wgp()
+        except RuntimeError:
+            bound = None
+
+        registered = sys.modules.get("wgp")
+        if bound is not None:
+            if registered is not bound:
+                raise RuntimeError(
+                    "The bound WanGP instance does not match sys.modules['wgp']; "
+                    "refusing to create a second runtime"
+                )
+            _require_root(bound, expected)
+            return bound
+
+        module = registered
+        if module is None:
+            module = importer("wgp") if importer is not None else _import_wgp()
+            if sys.modules.get("wgp") is not module:
+                raise RuntimeError(
+                    "The WanGP importer did not register the returned module in sys.modules"
+                )
+
+        _require_root(module, expected)
+        bind_wgp(module)
+        return module

app/services/generation/runtime.py

@@ -5,24 +5,33 @@
 """
 from __future__ import annotations
 
+import threading
 from typing import Any
 
 _UNBOUND_MESSAGE = (
-    "generation.bind_wgp() was not called; launch bootstrap must import wgp first"
+    "generation.bind_wgp() was not called; bootstrap through launch or "
+    "shared.api.init() first"
 )
 
 _wgp = None
+_wgp_lock = threading.RLock()
 
 
 def bind_wgp(module) -> None:
     global _wgp
-    _wgp = module
+    if module is None:
+        raise ValueError("Cannot bind an empty WanGP runtime")
+    with _wgp_lock:
+        if _wgp is not None and _wgp is not module:
+            raise RuntimeError("A different WanGP runtime is already bound")
+        _wgp = module
 
 
 def get_wgp():
-    if _wgp is None:
-        raise RuntimeError(_UNBOUND_MESSAGE)
-    return _wgp
+    with _wgp_lock:
+        if _wgp is None:
+            raise RuntimeError(_UNBOUND_MESSAGE)
+        return _wgp
 
 
 class ModelCatalog:

app/shared/api.py

@@ -870,17 +870,9 @@ def _ensure_runtime(self) -> _WanGPRuntime:
                 sys.path.insert(0, str(self._root))
 
             with _pushd(self._root), _temporary_argv(argv):
-                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}")
+                from services.generation.bootstrap import get_or_bootstrap_wgp
+
+                module = get_or_bootstrap_wgp(expected_root=self._root)
                 if not hasattr(module, "app"):
                     module.app = module.WAN2GPApplication()
                 module.download_ffmpeg()

docs/development/ARCHITECTURE_FOUNDATION.md

@@ -41,12 +41,15 @@ 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.
+does not import WanGP for consumers. The documented standalone Python API is
+the single exception: `generation/bootstrap.py` may import once when a process
+starts through `shared.api.init()` instead of launch, then binds that exact
+`sys.modules["wgp"]` instance. Video concat stays in `services.mix_concat` and
+only uses `get_wgp()` to reach the already-bound helper.
+
+The remaining first-party allowlist contains exactly the launch bootstrap and
+the standalone API bootstrap. 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,6 +15,11 @@
 
 WGP_ALLOWLIST = {
     ("app/_launch_runtime.py", "<module>", "import wgp"),
+    (
+        "app/services/generation/bootstrap.py",
+        "_import_wgp",
+        "importlib.import_module('wgp')",
+    ),
 }
 
 IGNORED_WGP_TREES = (
@@ -133,12 +138,18 @@ 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 generation_imports == set(), (
-        "The WanGP wall must bind the live bootstrap instance and must not "
-        f"reimport wgp. found={sorted(generation_imports)!r}"
+    assert generation_imports == {
+        (
+            "app/services/generation/bootstrap.py",
+            "_import_wgp",
+            "importlib.import_module('wgp')",
+        ),
+    }, (
+        "Only the explicit standalone bootstrap may import WanGP inside the "
+        f"generation boundary. found={sorted(generation_imports)!r}"
     )
     assert found == WGP_ALLOWLIST, (
-        "First-party WanGP imports may only be the launch bootstrap. "
+        "First-party WanGP imports may only be the two explicit bootstraps. "
         f"Added={sorted(found - WGP_ALLOWLIST)!r}, "
         f"removed={sorted(WGP_ALLOWLIST - found)!r}"
     )

tests/test_generation_runtime.py

@@ -14,12 +14,13 @@
     get_wgp,
 )
 from services.generation import runtime as generation_runtime
+from services.generation.bootstrap import get_or_bootstrap_wgp
 
 
 @pytest.fixture
 def isolated_wgp():
     previous = generation_runtime._wgp
-    bind_wgp(None)
+    generation_runtime._wgp = None
     try:
         yield
     finally:
@@ -54,3 +55,87 @@ def test_model_catalog_and_runtime_config_read_the_bound_instance(isolated_wgp)
     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"
+
+
+def _fake_wgp(root):
+    return SimpleNamespace(__file__=str(root / "wgp.py"), server_config={})
+
+
+def test_standalone_bootstrap_imports_once_and_reuses_the_singleton(
+    isolated_wgp,
+    monkeypatch,
+    tmp_path,
+) -> None:
+    fake = _fake_wgp(tmp_path)
+    calls = []
+
+    def importer(name: str):
+        calls.append(name)
+        monkeypatch.setitem(sys.modules, name, fake)
+        return fake
+
+    monkeypatch.delitem(sys.modules, "wgp", raising=False)
+    assert get_or_bootstrap_wgp(expected_root=tmp_path, importer=importer) is fake
+    assert get_or_bootstrap_wgp(expected_root=tmp_path, importer=importer) is fake
+    assert get_wgp() is sys.modules["wgp"]
+    assert calls == ["wgp"]
+
+
+def test_standalone_bootstrap_binds_an_existing_sys_modules_instance(
+    isolated_wgp,
+    monkeypatch,
+    tmp_path,
+) -> None:
+    fake = _fake_wgp(tmp_path)
+    monkeypatch.setitem(sys.modules, "wgp", fake)
+
+    def unexpected_import(_name: str):
+        raise AssertionError("the registered singleton must be reused")
+
+    assert get_or_bootstrap_wgp(expected_root=tmp_path, importer=unexpected_import) is fake
+    assert get_wgp() is fake
+
+
+def test_standalone_bootstrap_rejects_a_different_root_without_binding(
+    isolated_wgp,
+    monkeypatch,
+    tmp_path,
+) -> None:
+    other_root = tmp_path / "other"
+    fake = _fake_wgp(other_root)
+    monkeypatch.setitem(sys.modules, "wgp", fake)
+
+    with pytest.raises(RuntimeError, match="already loaded from"):
+        get_or_bootstrap_wgp(expected_root=tmp_path)
+    with pytest.raises(RuntimeError, match="generation.bind_wgp"):
+        get_wgp()
+
+
+def test_standalone_bootstrap_does_not_bind_a_failed_or_unregistered_import(
+    isolated_wgp,
+    monkeypatch,
+    tmp_path,
+) -> None:
+    monkeypatch.delitem(sys.modules, "wgp", raising=False)
+
+    def failed_import(_name: str):
+        raise ImportError("broken WanGP import")
+
+    with pytest.raises(ImportError, match="broken WanGP import"):
+        get_or_bootstrap_wgp(expected_root=tmp_path, importer=failed_import)
+    with pytest.raises(RuntimeError, match="generation.bind_wgp"):
+        get_wgp()
+
+    fake = _fake_wgp(tmp_path)
+    with pytest.raises(RuntimeError, match="did not register"):
+        get_or_bootstrap_wgp(expected_root=tmp_path, importer=lambda _name: fake)
+    with pytest.raises(RuntimeError, match="generation.bind_wgp"):
+        get_wgp()
+
+
+def test_bind_wgp_is_idempotent_but_rejects_a_second_runtime(isolated_wgp) -> None:
+    first = SimpleNamespace()
+    bind_wgp(first)
+    bind_wgp(first)
+    with pytest.raises(RuntimeError, match="different WanGP runtime"):
+        bind_wgp(SimpleNamespace())

tests/test_shared_api_bootstrap.py

@@ -0,0 +1,44 @@
+"""The documented Python API remains a real standalone WanGP bootstrap."""
+from __future__ import annotations
+
+import sys
+from types import SimpleNamespace
+
+from services.generation import bootstrap, get_wgp
+from services.generation import runtime as generation_runtime
+from shared import api as shared_api
+
+
+def test_init_bootstraps_and_binds_wgp_without_launch(monkeypatch, tmp_path) -> None:
+    previous_bound = generation_runtime._wgp
+    previous_runtime = shared_api._RUNTIME
+    previous_banner = shared_api._BANNER_PRINTED
+    generation_runtime._wgp = None
+    shared_api._RUNTIME = None
+    shared_api._BANNER_PRINTED = False
+    monkeypatch.delitem(sys.modules, "wgp", raising=False)
+
+    application = object()
+    downloads = []
+    fake = SimpleNamespace(
+        __file__=str(tmp_path / "wgp.py"),
+        WanGP_version="test",
+        WAN2GPApplication=lambda: application,
+        download_ffmpeg=lambda: downloads.append(True),
+    )
+
+    def import_once():
+        monkeypatch.setitem(sys.modules, "wgp", fake)
+        return fake
+
+    monkeypatch.setattr(bootstrap, "_import_wgp", import_once)
+    try:
+        session = shared_api.init(root=tmp_path, console_output=False)
+        assert session._ensure_runtime().module is fake
+        assert get_wgp() is sys.modules["wgp"] is fake
+        assert fake.app is application
+        assert downloads == [True]
+    finally:
+        shared_api._RUNTIME = previous_runtime
+        shared_api._BANNER_PRINTED = previous_banner
+        generation_runtime._wgp = previous_bound