refactor(megaprompt-v3): guard seed+generate with a shared GENERATION_LOCK

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Isulion
2026-10-20 20:36:45 +02:00
co-authored by Claude Opus 5.5
parent 6b901cefc1
commit 8a30471e60
2 changed files with 31 additions and 0 deletions
+14
View File
@@ -1,6 +1,7 @@
import json
import os
import random
import threading
from functools import lru_cache
from typing import Dict, List, Tuple, Optional
@@ -160,6 +161,12 @@ class ThemeRegistry:
return available_display_themes
# Every handler shares ConfigManager.random, so seeding, the random theme pick
# and handler.generate must not interleave between threads (workflow
# execution vs. the Mega Prompt V4 preview routes).
GENERATION_LOCK = threading.Lock()
@lru_cache(maxsize=1)
def get_theme_registry() -> "ThemeRegistry":
"""Shared ThemeRegistry singleton.
@@ -247,6 +254,13 @@ class IsulionMegaPromptV3:
lora_key: str = "",
include_environment: str = "yes", include_style: str = "yes",
include_effects: str = "yes", debug_mode: str = "off") -> Tuple[str, str, str, str, str, str, int]:
with GENERATION_LOCK:
return self._generate_unlocked(
theme, complexity, randomize, seed, custom_subject, custom_location, lora_key,
include_environment, include_style, include_effects, debug_mode)
def _generate_unlocked(self, theme, complexity, randomize, seed, custom_subject, custom_location,
lora_key, include_environment, include_style, include_effects, debug_mode):
self.debug_mode = (debug_mode == "on")
if self.debug_mode:
print(f"[DEBUG] generate called with theme={theme}, seed={seed}")
+17
View File
@@ -58,3 +58,20 @@ def test_generate_with_error_theme_returns_error_tuple_without_raising():
assert isinstance(prompt, str)
assert "error" in prompt.lower() or "failed" in prompt.lower()
assert selected_theme == "--- Error ---"
def test_v3_generate_runs_under_the_shared_generation_lock():
import_package()
from isulion_pkg.Core_Nodes import mega_prompt_V3 as v3
class RecordingLock:
def __init__(self): self.entries = 0
def __enter__(self): self.entries += 1
def __exit__(self, *a): return False
original, v3.GENERATION_LOCK = v3.GENERATION_LOCK, RecordingLock()
try:
v3.IsulionMegaPromptV3().generate(theme=ANIME_THEME, seed=1)
assert v3.GENERATION_LOCK.entries == 1
finally:
v3.GENERATION_LOCK = original