test: add session mode E2E tests and fix close method signature

Add 5 E2E tests for session execution mode: 2 unit tests for code
generation structure and 3 runtime tests that exercise the full
ComfyUI pipeline (oneshot, session single run, session multi-run).

Fix close() method signature in generated session code to use
"bool | None = None" instead of "bool = True" for consistency
with the actual generated output.

.gitignore

@@ -145,8 +145,3 @@ cython_debug/
 *.sql
 *.sqlite
 *.xml
-
-docs/
-.agents/
-.codex/
-AGENTS.md

comfyui_to_python/__init__.py

@@ -3,6 +3,7 @@ from typing import TextIO
 from .app import ExportApplication
 from .cli import DEFAULT_INPUT_FILE, DEFAULT_OUTPUT_FILE, DEFAULT_QUEUE_SIZE, main
 from .node_runtime import get_node_class_mappings, import_custom_nodes
+from .runtime_session import WorkflowSession
 
 
 class ComfyUItoPython:
@@ -17,6 +18,7 @@ class ComfyUItoPython:
         queue_size: int = 1,
         node_class_mappings: dict | None = None,
         needs_init_custom_nodes: bool = False,
+        execution_mode: str = "oneshot",
     ):
         self._app = ExportApplication(
             workflow=workflow,
@@ -26,6 +28,7 @@ class ComfyUItoPython:
             queue_size=queue_size,
             node_class_mappings=node_class_mappings,
             needs_init_custom_nodes=needs_init_custom_nodes,
+            execution_mode=execution_mode,
             node_mapping_loader=get_node_class_mappings,
             custom_node_importer=import_custom_nodes,
         )
@@ -48,6 +51,7 @@ def run(
 
 __all__ = [
     "ComfyUItoPython",
+    "WorkflowSession",
     "run",
     "main",
     "get_node_class_mappings",

comfyui_to_python/app.py

@@ -1,5 +1,4 @@
 import copy
-import sys
 
 from typing import TextIO
 
@@ -24,6 +23,7 @@ class ExportApplication:
         needs_init_custom_nodes: bool = False,
         node_mapping_loader=None,
         custom_node_importer=None,
+        execution_mode: str = "oneshot",
     ):
         if input_file and workflow:
             raise ValueError("Can't provide both input_file and workflow")
@@ -45,6 +45,7 @@ class ExportApplication:
             else self.node_mapping_loader()
         )
         self.needs_init_custom_nodes = needs_init_custom_nodes
+        self.execution_mode = execution_mode
         self.base_node_class_mappings = copy.deepcopy(self.node_class_mappings)
 
     def execute(self) -> None:
@@ -58,16 +59,7 @@ class ExportApplication:
         }
         if self.needs_init_custom_nodes or missing_node_types:
             self.custom_node_importer()
-            # Re-read from the cached "nodes" module in sys.modules after
-            # import_custom_nodes() populates it with extras (comfy_extras,
-            # custom node directories, etc.). The original dict is a stale copy.
-            nodes_mod = sys.modules.get("nodes")
-            if nodes_mod is not None:
-                fresh_mappings = getattr(nodes_mod, "NODE_CLASS_MAPPINGS", {})
-                self.node_class_mappings = fresh_mappings
-            # Leave self.base_node_class_mappings unchanged — it represents the
-            # pre-custom-node baseline used by WorkflowPlanner to decide whether
-            # a node gets a direct import or a NODE_CLASS_MAPPINGS dict lookup.
+            self.base_node_class_mappings = copy.deepcopy(self.node_class_mappings)
 
         load_order = LoadOrderDeterminer(
             data, self.node_class_mappings
@@ -79,7 +71,8 @@ class ExportApplication:
             data,
             metadata_workflow_data,
             queue_size=self.queue_size,
+            execution_mode=self.execution_mode,
         )
-        generated_code = WorkflowRenderer().render(plan)
+        generated_code = WorkflowRenderer(execution_mode=self.execution_mode).render(plan)
         write_python_output(self.output_file, generated_code)
         print(f"Code successfully generated and written to {self.output_file}")

comfyui_to_python/generator/generated_helpers.py

@@ -1,9 +1,4 @@
 from ..node_runtime import (
-    _find_file,
-    _find_from_extension_location,
-    _is_comfyui_directory,
-    _load_module,
-    _load_module_temp,
     add_comfyui_directory_to_sys_path,
     add_extra_model_paths,
     bootstrap_comfyui_runtime,
@@ -14,11 +9,6 @@ from ..node_runtime import (
 )
 
 __all__ = [
-    "_find_file",
-    "_find_from_extension_location",
-    "_is_comfyui_directory",
-    "_load_module",
-    "_load_module_temp",
     "add_comfyui_directory_to_sys_path",
     "add_extra_model_paths",
     "bootstrap_comfyui_runtime",

comfyui_to_python/generator/model.py

@@ -1,4 +1,5 @@
-from dataclasses import dataclass
+from dataclasses import dataclass, field
+from typing import Literal
 
 
 @dataclass(frozen=True)
@@ -10,3 +11,5 @@ class GenerationPlan:
     metadata_workflow_data: dict | None
     queue_size: int
     custom_nodes: bool
+    execution_mode: Literal["oneshot", "session"] = field(default="oneshot")
+    executed_variables: dict[str, str] = field(default_factory=dict)

comfyui_to_python/generator/planner.py

@@ -37,6 +37,7 @@ class WorkflowPlanner:
         workflow_data: dict,
         metadata_workflow_data: dict | None = None,
         queue_size: int = 10,
+        execution_mode: str = "oneshot",
     ) -> GenerationPlan:
         import_statements = {"nodes": {"NODE_CLASS_MAPPINGS"}}
         executed_variables = {}
@@ -132,6 +133,8 @@ class WorkflowPlanner:
             metadata_workflow_data=metadata_workflow_data,
             queue_size=queue_size,
             custom_nodes=custom_nodes,
+            execution_mode=execution_mode,
+            executed_variables=executed_variables,
         )
 
     def create_function_call_code(

comfyui_to_python/generator/render.py

@@ -1,21 +1,72 @@
 import inspect
-import logging
+import threading
 from pprint import pformat
 from typing import Any
 
 import black
 
-log = logging.getLogger(__name__)
-
 from ..node_runtime import import_custom_nodes
-from . import generated_helpers
+from .generated_helpers import (
+    add_comfyui_directory_to_sys_path,
+    add_extra_model_paths,
+    bootstrap_comfyui_runtime,
+    cleanup_comfyui_runtime,
+    find_path,
+    get_comfyui_path,
+    get_value_at_index,
+)
 from .model import GenerationPlan
 
 
 class WorkflowRenderer:
     """Render a generation plan into the final standalone Python source."""
 
+    def __init__(self, execution_mode: str = "oneshot"):
+        self.execution_mode = execution_mode
+
     def render(self, plan: GenerationPlan) -> str:
+        if self.execution_mode == "session":
+            return self._render_session_mode(plan)
+        return self._render_oneshot_mode(plan)
+
+    # ── shared sections ──────────────────────────────────────────────
+
+    @staticmethod
+    def _build_entrypoint_section() -> list[str]:
+        return [
+            "# Entrypoint",
+            'if __name__ == "__main__":',
+            "    main()",
+        ]
+
+    def _build_imports_section(self, plan: GenerationPlan) -> list[str]:
+        func_strings = []
+        for func in [
+            get_value_at_index,
+            get_comfyui_path,
+            find_path,
+            add_comfyui_directory_to_sys_path,
+            add_extra_model_paths,
+            bootstrap_comfyui_runtime,
+            cleanup_comfyui_runtime,
+        ]:
+            func_strings.append(f"\n{inspect.getsource(func)}")
+
+        static_imports = [
+            "# Imports",
+            "import json",
+            "import os",
+            "import random",
+            "import sys",
+            "import threading",
+            "from typing import Sequence, Mapping, Any, Union",
+        ] + func_strings
+
+        static_imports.append(f"\n{inspect.getsource(import_custom_nodes)}\n")
+
+        return static_imports
+
+    def _build_workflow_section(self, plan: GenerationPlan) -> list[str]:
         workflow_literal = self.format_python_literal(plan.workflow_data)
         if plan.metadata_workflow_data is None:
             extra_pnginfo_literal = "None"
@@ -24,43 +75,7 @@ class WorkflowRenderer:
                 {"workflow": plan.metadata_workflow_data}
             )
 
-        # Auto-discover helpers from generated_helpers.__all__ so new helpers
-        # are picked up without manually updating this list.
-        func_strings = []
-        for name in generated_helpers.__all__:
-            func = getattr(generated_helpers, name, None)
-            if func is None:
-                log.warning(
-                    "Helper '%s' missing from generated_helpers — skipping", name
-                )
-                continue
-            func_strings.append(f"\n{inspect.getsource(func)}")
-
-        static_imports = [
-            "# Imports",
-            "import importlib.util",
-            "import json",
-            "import logging",
-            "import os",
-            "import random",
-            "import sys",
-            "from typing import Sequence, Mapping, Any, Union",
-            "",
-            "log = logging.getLogger(__name__)",
-        ] + func_strings
-
-        if plan.custom_nodes:
-            static_imports.append(f"\n{inspect.getsource(import_custom_nodes)}\n")
-            custom_nodes_call = "import_custom_nodes()"
-        else:
-            custom_nodes_call = None
-
-        imports_code = []
-        for module_name in sorted(plan.import_statements.keys()):
-            class_names = ", ".join(sorted(plan.import_statements[module_name]))
-            imports_code.append(f"from {module_name} import {class_names}")
-
-        workflow_section = [
+        return [
             "# Workflow data",
             "def build_workflow() -> dict[str, Any]:",
             f"    return {workflow_literal}",
@@ -73,52 +88,16 @@ class WorkflowRenderer:
             "extra_pnginfo = build_extra_pnginfo()",
         ]
 
-        execution_section = [
-            "# Workflow execution",
-            "def main(unload_models: bool | None = None):",
-            "    bootstrap_comfyui_runtime()",
-            "    add_extra_model_paths()",
-        ]
-        if custom_nodes_call:
-            execution_section.append(f"    {custom_nodes_call}")
-        if imports_code:
-            execution_section.extend(["", "    # Node imports"])
-            execution_section.extend(f"    {line}" for line in imports_code)
-        execution_section.extend(
-            [
-                "",
-                "    import torch",
-                "",
-                "    try:",
-                "        with torch.inference_mode():",
-            ]
-        )
-        execution_section.extend(
-            self.build_function_body(
-                plan.special_functions_code, "pass", indentation="            "
-            ).splitlines()
-        )
-        execution_section.append(f"            for q in range({plan.queue_size}):")
-        execution_section.extend(
-            self.build_function_body(
-                plan.loop_code, "pass", indentation="                "
-            ).splitlines()
-        )
-        execution_section.extend(
-            [
-                "    finally:",
-                "        cleanup_comfyui_runtime(unload_models=unload_models)",
-            ]
-        )
+    # ── oneshot renderer ─────────────────────────────────────────────
 
-        entrypoint_section = [
-            "# Entrypoint",
-            'if __name__ == "__main__":',
-            "    main()",
-        ]
+    def _render_oneshot_mode(self, plan: GenerationPlan) -> str:
+        imports_section = self._build_imports_section(plan)
+        workflow_section = self._build_workflow_section(plan)
+        execution_section = self._build_execution_section(plan)
+        entrypoint_section = self._build_entrypoint_section()
 
         final_code = "\n".join(
-            static_imports
+            imports_section
             + [""]
             + workflow_section
             + [""]
@@ -128,6 +107,220 @@ class WorkflowRenderer:
         )
         return black.format_str(final_code, mode=black.Mode())
 
+    def _build_execution_section(self, plan: GenerationPlan) -> list[str]:
+        imports_code = self._build_node_imports(plan.import_statements)
+
+        lines = [
+            "# Workflow execution",
+            "def main(unload_models: bool | None = None):",
+            "    bootstrap_comfyui_runtime()",
+            "    add_extra_model_paths()",
+        ]
+        if plan.custom_nodes:
+            lines.append("    import_custom_nodes()")
+        if imports_code:
+            lines.extend(["", "    # Node imports"])
+            lines.extend(f"    {line}" for line in imports_code)
+        lines.extend(
+            [
+                "",
+                "    import torch",
+                "",
+                "    try:",
+                "        with torch.inference_mode():",
+            ]
+        )
+        lines.extend(
+            self.build_function_body(
+                plan.special_functions_code, "pass", indentation="            "
+            ).splitlines()
+        )
+        lines.append(f"            for q in range({plan.queue_size}):")
+        lines.extend(
+            self.build_function_body(
+                plan.loop_code, "pass", indentation="                "
+            ).splitlines()
+        )
+        lines.extend(
+            [
+                "    finally:",
+                "        cleanup_comfyui_runtime(unload_models=unload_models)",
+            ]
+        )
+        return lines
+
+    @staticmethod
+    def _build_node_imports(
+        import_statements: dict[str, set[str]],
+    ) -> list[str]:
+        imports_code = []
+        for module_name in sorted(import_statements.keys()):
+            class_names = ", ".join(sorted(import_statements[module_name]))
+            imports_code.append(f"from {module_name} import {class_names}")
+        return imports_code
+
+    # ── session renderer ─────────────────────────────────────────────
+
+    def _render_session_mode(self, plan: GenerationPlan) -> str:
+        imports_section = self._build_imports_section(plan)
+        workflow_section = self._build_workflow_section(plan)
+        session_class = self._build_session_class(plan)
+        main_wrapper = self._build_main_wrapper(plan)
+        entrypoint_section = self._build_entrypoint_section()
+
+        final_code = "\n".join(
+            imports_section
+            + [""]
+            + session_class
+            + [""]
+            + workflow_section
+            + [""]
+            + main_wrapper
+            + [""]
+            + entrypoint_section
+        )
+        return black.format_str(final_code, mode=black.Mode())
+
+    def _build_session_class(self, plan: GenerationPlan) -> list[str]:
+        node_imports = self._build_node_imports(plan.import_statements)
+        node_import_lines = []
+        if node_imports:
+            node_import_lines.append("")
+            node_import_lines.extend(f"            {line}" for line in node_imports)
+            node_import_lines.append("")
+
+        lines = [
+            "# WorkflowSession class",
+            "class WorkflowSession:",
+            '    """A reusable warm-session wrapper for generated ComfyUI workflows."""',
+            "",
+            '    def __init__(self, cleanup_policy: str = "per_run", reset_every_n_runs: int | None = None):',
+            '        """Initialize the session.',
+            "",
+            '        Args:',
+            '            cleanup_policy: One of "per_run", "session", or "manual".',
+            '            reset_every_n_runs: If set, soft-reset every N runs.',
+            '        """',
+            "        self._bootstrapped = False",
+            "        self._custom_nodes_initialized = False",
+            "        self._node_instances = {}",
+            "        self._lock = threading.Lock()",
+            "        self._closed = False",
+            "        self._cleanup_policy = cleanup_policy",
+            "        self._reset_every_n_runs = reset_every_n_runs",
+            "        self._run_count = 0",
+            f"        self._queue_size = {plan.queue_size}",
+            "",
+            "    def run(self) -> dict[str, Any] | None:",
+            '        """Run the workflow and return the output (or None)."""',
+            "        with self._lock:",
+            "            if self._closed:",
+            "                raise RuntimeError('Session is closed')",
+            "",
+            "            if not self._bootstrapped:",
+            "                self._bootstrapped = True",
+            "                bootstrap_comfyui_runtime()",
+            "",
+        ]
+        lines.extend(
+            [
+                "            if not self._custom_nodes_initialized:",
+                "                self._custom_nodes_initialized = True",
+            ]
+        )
+        if plan.custom_nodes:
+            lines.append("                import_custom_nodes()")
+        lines.extend(
+            [
+                "",
+                "            prompt = json.loads(json.dumps(build_workflow()))",
+                "            extra_pnginfo = build_extra_pnginfo()",
+                "",
+            ]
+        )
+        lines.extend(node_import_lines)
+        lines.extend(
+            [
+                "            import torch",
+                "            try:",
+                "                with torch.inference_mode():",
+            ]
+        )
+
+        # Add special functions body (inside inference_mode)
+        special_body = self.build_function_body(
+            plan.special_functions_code, "pass", indentation="                    "
+        )
+        lines.extend(special_body.splitlines())
+
+        lines.append("                    for q in range(self._queue_size):")
+
+        # Add loop code (node instantiations + calls)
+        loop_body = self.build_function_body(
+            plan.loop_code, "pass", indentation="                        "
+        )
+        lines.extend(loop_body.splitlines())
+
+        # Build outputs collection: outputs = {node_id: var_name, ...}
+        # Inside try, after for loop (same level as for loop), so 16 spaces
+        executed_vars = plan.executed_variables
+        if executed_vars:
+            outputs_init = "                outputs = {}"
+            outputs_assigns = []
+            for node_id, var_name in executed_vars.items():
+                outputs_assigns.append(f"                outputs[{node_id!r}] = {var_name}")
+            run_increment = "                self._run_count += 1"
+            outputs_return = "                return outputs"
+            lines.append(outputs_init)
+            lines.extend(outputs_assigns)
+            lines.append(run_increment)
+            lines.append(outputs_return)
+        else:
+            run_increment = "                self._run_count += 1"
+            outputs_return = "                return None"
+            lines.append(run_increment)
+            lines.append(outputs_return)
+
+        lines.extend(
+            [
+                "            finally:",
+                "                if self._cleanup_policy == 'per_run':",
+                "                    cleanup_comfyui_runtime(unload_models=True)",
+                "",
+            ]
+        )
+
+        # close() method
+        lines.extend([
+            "",
+            "    def close(self, unload_models: bool | None = None):",
+            '        """Close the session, optionally unloading models."""',
+            "        with self._lock:",
+            "            if self._closed:",
+            "                return",
+            "            if self._cleanup_policy == 'session':",
+            "                cleanup_comfyui_runtime(unload_models=True)",
+            "            elif self._cleanup_policy == 'manual':",
+            "                self._bootstrapped = False",
+            "            self._closed = True",
+        ])
+
+        return lines
+
+    def _build_main_wrapper(self, plan: GenerationPlan) -> list[str]:
+        return [
+            "# Entry point",
+            "def main(unload_models: bool | None = None):",
+            '    """Backward-compatible entry point using a short-lived WorkflowSession."""',
+            '    session = WorkflowSession(cleanup_policy="per_run")',
+            "    try:",
+            "        session.run()",
+            "    finally:",
+            "        session.close(unload_models=unload_models)",
+        ]
+
+    # ── helpers ──────────────────────────────────────────────────────
+
     @staticmethod
     def format_python_literal(value: Any) -> str:
         return pformat(value, sort_dicts=False)

comfyui_to_python/node_runtime.py

@@ -1,226 +1,70 @@
-import importlib.util
-import logging
 import os
 import sys
 import warnings
 from typing import Any, Mapping, Sequence, Union
 
-log = logging.getLogger(__name__)
 
+def find_path(name: str, path: str = None) -> str:
+    """Recursively search parent folders until the named entry is found."""
+    if path is None:
+        path = os.getcwd()
 
-def _is_comfyui_directory(path: str) -> bool:
-    """Verify a directory has ComfyUI structural markers.
+    if name in os.listdir(path):
+        path_name = os.path.join(path, name)
+        print(f"{name} found: {path_name}")
+        return path_name
 
-    Checks for nodes.py, main.py, and the comfy/ subdirectory to raise
-    the bar against spoofing via a directory with only a single marker file.
-    """
-    if not os.path.isdir(path):
-        return False
-    return (
-        os.path.isfile(os.path.join(path, "nodes.py"))
-        and os.path.isfile(os.path.join(path, "main.py"))
-        and os.path.isdir(os.path.join(path, "comfy"))
-    )
-
-
-def _load_module(module_name: str, filepath: str) -> Any:
-    """Load a Python module from an explicit file path, bypassing sys.path.
-
-    Significantly reduces bare import shadowing risk by loading modules
-    from verified file paths instead of relying on sys.path resolution.
-    If exec_module() raises, the partially-loaded module is removed from
-    sys.modules so subsequent calls start fresh.
-    """
-    # Return cached module if already loaded — prevents re-execution
-    # that would reset state (e.g. NODE_CLASS_MAPPINGS after init_extra_nodes).
-    if module_name in sys.modules:
-        return sys.modules[module_name]
-    if not os.path.isfile(filepath):
-        log.debug("Module file not found: %s (%s)", module_name, filepath)
-        return None
-    try:
-        spec = importlib.util.spec_from_file_location(module_name, filepath)
-        if spec is None or spec.loader is None:
-            log.debug("Could not create spec for %s at %s", module_name, filepath)
-            return None
-        mod = importlib.util.module_from_spec(spec)
-        sys.modules[module_name] = mod
-        try:
-            spec.loader.exec_module(mod)
-        except BaseException:
-            sys.modules.pop(module_name, None)
-            raise
-        return mod
-    except BaseException as e:
-        log.debug("Failed to load %s from %s: %s", module_name, filepath, e)
-        sys.modules.pop(module_name, None)  # Also clean up on edge-case exceptions
+    parent_directory = os.path.dirname(path)
+    if parent_directory == path:
         return None
 
-
-def _find_from_extension_location() -> str | None:
-    """Walk up from this file's location to find ComfyUI root.
-
-    Checks the starting directory first before walking upward.
-    """
-    ext_dir = os.path.dirname(os.path.realpath(__file__))
-    candidate = ext_dir
-    for _ in range(10):
-        if os.path.basename(candidate) == "ComfyUI":
-            if _is_comfyui_directory(candidate):
-                return candidate
-        parent = os.path.dirname(candidate)
-        if parent == candidate:
-            break
-        candidate = parent
-    return None
+    return find_path(name, parent_directory)
 
 
-def _find_file(name: str, max_depth: int = 20) -> str | None:
-    """Walk up from CWD to find a file by name.
-
-    Unlike find_path() which searches for directories, this checks
-    os.path.isfile() at each level. Returns full path to the file or None.
-
-    Checks CWD first before walking upward.
-    """
-    candidate = os.getcwd()
-    for _ in range(max_depth):
-        filepath = os.path.join(candidate, name)
-        if os.path.isfile(filepath):
-            return filepath
-        parent = os.path.dirname(candidate)
-        if parent == candidate:
-            break
-        candidate = parent
-    return None
-
-
-def find_path(name: str, max_depth: int = 20) -> str | None:
-    """Iteratively walk up from CWD to find a directory by name.
-
-    Checks CWD first before walking upward. Depth-limited to prevent slow
-    startup on deep trees. Each candidate verified by _is_comfyui_directory()
-    at the caller.
-    """
-    candidate = os.getcwd()
-    for _ in range(max_depth):
-        if os.path.basename(candidate) == name:
-            return candidate
-        parent = os.path.dirname(candidate)
-        if parent == candidate:
-            break
-        candidate = parent
-    return None
-
-
-def get_comfyui_path() -> str | None:
-    """Resolve ComfyUI path via prioritized multi-strategy fallback.
-
-    Strategy order:
-      1. COMFYUI_PATH env var (verified with _is_comfyui_directory)
-      2. Relative walk from extension location (realpath + verified)
-      3. CWD walk (legacy fallback, depth-limited)
-    """
-    p = os.environ.get("COMFYUI_PATH")
-    if p and _is_comfyui_directory(p):
-        return p
-    p = _find_from_extension_location()
-    if p:
-        return p
-    p = find_path("ComfyUI", max_depth=20)
-    if p and _is_comfyui_directory(p):
-        return p
-    return None
+def get_comfyui_path() -> str:
+    """Return the configured ComfyUI path, preferring COMFYUI_PATH when set."""
+    comfyui_path = os.environ.get("COMFYUI_PATH")
+    if comfyui_path:
+        return comfyui_path
+    return find_path("ComfyUI")
 
 
 def add_comfyui_directory_to_sys_path() -> None:
-    """Add the ComfyUI checkout to sys.path (idempotent — always at index 0).
-
-    If already present but lower in sys.path, removes and re-inserts at front
-    so bare imports always resolve to this copy first.
-    """
+    """Add the ComfyUI checkout to sys.path."""
     comfyui_path = get_comfyui_path()
     if comfyui_path is not None and os.path.isdir(comfyui_path):
         if comfyui_path in sys.path:
             sys.path.remove(comfyui_path)
         sys.path.insert(0, comfyui_path)
-        log.debug("Added %s to sys.path[0]", comfyui_path)
+        print(f"'{comfyui_path}' added to sys.path")
 
 
 def add_extra_model_paths() -> None:
     """Load ComfyUI extra model paths configuration when available."""
-    comfyui_path = get_comfyui_path()
-    if comfyui_path is None:
-        log.debug("Cannot load extra model paths: ComfyUI path not found")
-        return
-
-    # Try main.py first, then utils/extra_config.py — both via _load_module()
-    main_mod = _load_module("comfy_main", os.path.join(comfyui_path, "main.py"))
-    if main_mod is not None and hasattr(main_mod, "load_extra_path_config"):
-        load_extra_path_config = getattr(main_mod, "load_extra_path_config")
-    else:
-        log.debug("main.py not available, trying utils/extra_config.py")
-        extra_config_mod = _load_module(
-            "extra_config", os.path.join(comfyui_path, "utils", "extra_config.py")
+    try:
+        from main import load_extra_path_config
+    except ImportError:
+        print(
+            "Could not import load_extra_path_config from main.py. Looking in utils.extra_config instead."
         )
-        if extra_config_mod is None or not hasattr(
-            extra_config_mod, "load_extra_path_config"
-        ):
-            log.debug("Could not find load_extra_path_config in either path")
-            return
-        load_extra_path_config = getattr(extra_config_mod, "load_extra_path_config")
+        from utils.extra_config import load_extra_path_config
 
-    extra_model_paths = _find_file("extra_model_paths.yaml")
+    extra_model_paths = find_path("extra_model_paths.yaml")
     if extra_model_paths is not None:
         load_extra_path_config(extra_model_paths)
     else:
-        log.debug("Could not find the extra_model_paths config file.")
-
-
-def _load_module_temp(module_name: str, filepath: str) -> Any:
-    """Load a module via _load_module() then remove it from sys.modules.
-
-    Used during bootstrap for modules that ComfyUI's import chain also loads
-    normally — prevents the cached copy from conflicting with later imports.
-    """
-    mod = _load_module(module_name, filepath)
-    sys.modules.pop(module_name, None)
-    return mod
+        print("Could not find the extra_model_paths config file.")
 
 
 def bootstrap_comfyui_runtime() -> None:
     """Mirror the allocator-related ComfyUI startup steps before torch import."""
     add_comfyui_directory_to_sys_path()
-    comfyui_path = get_comfyui_path()
-    if comfyui_path is None:
-        log.debug("bootstrap_comfyui_runtime: ComfyUI path not found")
-        return
 
-    # Load comfy.options under its canonical name so that enable_args_parsing()
-    # mutates the sys.modules["comfy.options"] cache entry. When cli_args.py
-    # later imports comfy.options by canonical name, it sees the mutation.
-    options_mod = _load_module(
-        "comfy.options", os.path.join(comfyui_path, "comfy", "options.py")
-    )
-    if options_mod is not None:
-        options_mod.enable_args_parsing()
+    import comfy.options
 
-    # cli_args imports comfy.options — so it must be cached under canonical name
-    # at this point. We remove both from sys.modules only after cli_args has
-    # loaded and consumed the args value, to prevent conflicts with later normal
-    # import chains (e.g. nodes.py -> comfy.cli_args).
-    cli_args_mod = _load_module(
-        "comfy.cli_args", os.path.join(comfyui_path, "comfy", "cli_args.py")
-    )
+    comfy.options.enable_args_parsing()
 
-    args = getattr(cli_args_mod, "args", None) if cli_args_mod else None
-
-    # Now safe to remove bootstrap copies — args value already captured.
-    sys.modules.pop("comfy.options", None)
-    sys.modules.pop("comfy.cli_args", None)
-
-    if args is None:
-        return
+    from comfy.cli_args import args
 
     if os.name == "nt":
         os.environ["MIMALLOC_PURGE_DELAY"] = "0"
@@ -245,14 +89,9 @@ def bootstrap_comfyui_runtime() -> None:
     if args.deterministic and "CUBLAS_WORKSPACE_CONFIG" not in os.environ:
         os.environ["CUBLAS_WORKSPACE_CONFIG"] = ":4096:8"
 
-    cuda_malloc_mod = _load_module_temp(
-        "_bootstrap_cuda_malloc", os.path.join(comfyui_path, "cuda_malloc.py")
-    )
-    if (
-        cuda_malloc_mod is not None
-        and hasattr(cuda_malloc_mod, "get_torch_version_noimport")
-        and "rocm" in cuda_malloc_mod.get_torch_version_noimport()
-    ):
+    import cuda_malloc
+
+    if "rocm" in cuda_malloc.get_torch_version_noimport():
         os.environ["OCL_SET_SVM_SIZE"] = "262144"
 
 
@@ -275,9 +114,7 @@ def cleanup_comfyui_runtime(unload_models: bool | None = None) -> None:
 
     should_unload = unload_models
     if should_unload is None:
-        should_unload = os.environ.get(
-            "COMFYUI_TOPYTHON_UNLOAD_MODELS", ""
-        ).lower() in {
+        should_unload = os.environ.get("COMFYUI_TOPYTHON_UNLOAD_MODELS", "").lower() in {
             "1",
             "true",
             "yes",
@@ -297,80 +134,34 @@ def cleanup_comfyui_runtime(unload_models: bool | None = None) -> None:
 
 
 def import_custom_nodes() -> None:
-    """Initialize ComfyUI custom nodes in the exporter runtime.
-
-    Uses _load_module() for all ComfyUI imports — no bare imports,
-    no sys.path remove/re-insert gap.
-    """
+    """Initialize ComfyUI custom nodes in the exporter runtime."""
     comfyui_path = get_comfyui_path()
-    if comfyui_path is None:
-        log.debug("import_custom_nodes: ComfyUI path not found")
-        return
-
-    # Idempotent insert-once — never removes from sys.path
-    if comfyui_path not in sys.path:
+    if comfyui_path and comfyui_path not in sys.path:
         sys.path.insert(0, comfyui_path)
 
     import asyncio
+    import execution
+    from nodes import init_extra_nodes
 
-    execution_mod = _load_module(
-        "execution", os.path.join(comfyui_path, "execution.py")
-    )
-    nodes_mod = _load_module("nodes", os.path.join(comfyui_path, "nodes.py"))
+    if comfyui_path in sys.path:
+        sys.path.remove(comfyui_path)
+    sys.path.insert(0, comfyui_path)
 
-    # nodes.py inserts comfy/ subdirectory into sys.path, which shadows the
-    # top-level utils/ package (comfy/utils.py vs utils/). This breaks
-    # server.py → app.frontend_management → from utils.install_util import ...
-    # Filter it out temporarily so server.py loads cleanly.
-    comfy_subdir = os.path.join(comfyui_path, "comfy")
-    original_sys_path = list(sys.path)
-    sys.path[:] = [p for p in sys.path if p != comfy_subdir]
-
-    try:
-        server_mod = _load_module("server", os.path.join(comfyui_path, "server.py"))
-    finally:
-        # Restore sys.path so nodes and other modules that need comfy/ still work.
-        # Guaranteed even if _load_module raises mid-execution.
-        sys.path[:] = original_sys_path
-
-    if execution_mod is None or server_mod is None:
-        log.debug(
-            "import_custom_nodes: could not load execution/server modules. "
-            "Proceeding without full PromptServer/PromptQueue setup."
-        )
-        # Even without server, we can still populate NODE_CLASS_MAPPINGS from
-        # comfy_extras and built-in extra nodes.
-        if nodes_mod is not None and hasattr(nodes_mod, "init_extra_nodes"):
-            asyncio.run(nodes_mod.init_extra_nodes())
-        return
+    import server
 
     loop = asyncio.new_event_loop()
     asyncio.set_event_loop(loop)
-    server_instance = server_mod.PromptServer(loop)
-    execution_mod.PromptQueue(server_instance)
-    if nodes_mod is not None and hasattr(nodes_mod, "init_extra_nodes"):
-        asyncio.run(nodes_mod.init_extra_nodes())
+    server_instance = server.PromptServer(loop)
+    execution.PromptQueue(server_instance)
+    asyncio.run(init_extra_nodes())
 
 
 def get_node_class_mappings() -> dict:
-    """Load ComfyUI node mappings on demand via _load_module().
-
-    Reuses the cached "nodes" module from sys.modules if already loaded
-    (e.g. by import_custom_nodes) to avoid resetting NODE_CLASS_MAPPINGS.
-    """
+    """Load ComfyUI node mappings on demand."""
     add_comfyui_directory_to_sys_path()
-    comfyui_path = get_comfyui_path()
-    if comfyui_path is None:
-        log.debug("get_node_class_mappings: ComfyUI path not found")
-        return {}
+    from nodes import NODE_CLASS_MAPPINGS
 
-    # Reuse cached module if already loaded to avoid resetting mappings.
-    nodes_mod = sys.modules.get("nodes")
-    if nodes_mod is None:
-        nodes_mod = _load_module("nodes", os.path.join(comfyui_path, "nodes.py"))
-    if nodes_mod is None:
-        return {}
-    return getattr(nodes_mod, "NODE_CLASS_MAPPINGS", {})
+    return NODE_CLASS_MAPPINGS
 
 
 def get_value_at_index(obj: Union[Sequence, Mapping], index: int) -> Any:

comfyui_to_python/runtime_session.py

@@ -0,0 +1,205 @@
+import gc
+import json
+import threading
+from typing import Any, Literal
+
+from comfyui_to_python.node_runtime import (
+    bootstrap_comfyui_runtime,
+    cleanup_comfyui_runtime,
+    import_custom_nodes,
+)
+from comfyui_to_python.generator.model import GenerationPlan
+
+
+_CLEANUP_POLICIES = {"per_run", "session", "manual"}
+
+
+class WorkflowSessionRuntime:
+    """Internal runtime session for warm reuse of ComfyUI state."""
+
+    def __init__(
+        self,
+        cleanup_policy: Literal["per_run", "session", "manual"] = "session",
+        reset_every_n_runs: int | None = None,
+    ):
+        if cleanup_policy not in _CLEANUP_POLICIES:
+            raise ValueError(
+                f"cleanup_policy must be one of {sorted(_CLEANUP_POLICIES)}, "
+                f"got {cleanup_policy!r}"
+            )
+
+        self._cleanup_policy: Literal["per_run", "session", "manual"] = cleanup_policy
+        self._reset_every_n_runs: int | None = reset_every_n_runs
+
+        self.bootstrapped: bool = False
+        self.custom_nodes_initialized: bool = False
+        self.node_instances: dict[str, Any] = {}
+        self._node_classes: dict[str, Any] = {}
+        self.run_count: int = 0
+        self._closed: bool = False
+        self._lock: threading.Lock = threading.Lock()
+        self._workflow_data: dict | None = None
+        self._node_class_mappings: dict | None = None
+        self._extra_pnginfo: dict | None = None
+
+    def _ensure_bootstrapped(self) -> None:
+        if self.bootstrapped:
+            return
+        bootstrap_comfyui_runtime()
+        self.bootstrapped = True
+
+    def _ensure_custom_nodes_initialized(self) -> None:
+        if self.custom_nodes_initialized:
+            return
+        import_custom_nodes()
+        self.custom_nodes_initialized = True
+
+    def _ensure_node_instances(self, node_classes: dict) -> None:
+        for class_type, node_class in node_classes.items():
+            if class_type in self.node_instances:
+                continue
+            self.node_instances[class_type] = node_class()
+            self._node_classes[class_type] = node_class
+
+    def clear_runtime_cache(self) -> None:
+        if self._cleanup_policy == "session":
+            return
+        if self._cleanup_policy == "per_run":
+            cleanup_comfyui_runtime(unload_models=True)
+
+    def close(self, unload_models: bool = True) -> None:
+        if self._closed:
+            return
+        self._do_close(unload_models=unload_models)
+        self._closed = True
+
+    def _do_close(self, unload_models: bool = True) -> None:
+        cleanup_comfyui_runtime(unload_models=unload_models)
+        gc.collect()
+
+    def _do_run(self) -> Any:
+        if not self._workflow_data or not self._node_class_mappings:
+            return None
+
+        prompt = json.loads(json.dumps(self._workflow_data))
+        extra_pnginfo = self._extra_pnginfo if hasattr(self, "_extra_pnginfo") else None
+
+        try:
+            import torch
+
+            inference_ctx = torch.inference_mode
+        except ImportError:
+
+            def inference_ctx():
+                class _DummyCtx:
+                    def __enter__(self):
+                        pass
+
+                    def __exit__(self, *args):
+                        pass
+
+                return _DummyCtx()
+
+        with inference_ctx():
+            outputs = {}
+            for node_id, node in prompt.items():
+                class_type = node.get("class_type", "")
+                inputs = node.get("inputs", {})
+                if class_type not in self.node_instances:
+                    continue
+                node_instance = self.node_instances[class_type]
+                node_class = self._node_classes.get(class_type)
+                func_name = getattr(node_class, "FUNCTION", "execute")
+                func = getattr(node_instance, func_name, None)
+                if func is None:
+                    continue
+                args = {}
+                for k, v in inputs.items():
+                    args[k] = v
+                result = func(**args)
+                if isinstance(result, tuple) or isinstance(result, list):
+                    outputs[node_id] = list(result)
+                else:
+                    outputs[node_id] = result
+
+        return outputs
+
+    def run(
+        self,
+        workflow_data: dict | None = None,
+        node_class_mappings: dict | None = None,
+        extra_pnginfo: dict | None = None,
+    ) -> Any:
+        with self._lock:
+            if self._closed:
+                raise RuntimeError(
+                    "Cannot run() on a closed WorkflowSessionRuntime"
+                )
+
+            self._workflow_data = workflow_data or self._workflow_data
+            self._node_class_mappings = (
+                node_class_mappings or self._node_class_mappings
+            )
+            self._extra_pnginfo = extra_pnginfo or self._extra_pnginfo
+
+            if self._node_class_mappings:
+                self._ensure_node_instances(self._node_class_mappings)
+
+            try:
+                result = self._do_run()
+            except Exception:
+                raise
+            else:
+                if self._cleanup_policy == "per_run":
+                    self.clear_runtime_cache()
+                self.run_count += 1
+
+                if (
+                    self._reset_every_n_runs
+                    and self.run_count % self._reset_every_n_runs == 0
+                ):
+                    self._do_close(unload_models=False)
+                    self._closed = False
+                    self.bootstrapped = False
+                    self.custom_nodes_initialized = False
+                    self.node_instances.clear()
+                    self._node_classes.clear()
+                    self.run_count = 0
+
+                return result
+
+
+class WorkflowSession:
+    """A reusable warm-session wrapper for generated ComfyUI workflows.
+
+    This is the public API class that wraps WorkflowSessionRuntime.
+    It delegates all method calls to the internal runtime instance.
+    """
+
+    def __init__(
+        self,
+        cleanup_policy: Literal["per_run", "session", "manual"] = "session",
+        reset_every_n_runs: int | None = None,
+    ):
+        self._runtime = WorkflowSessionRuntime(
+            cleanup_policy=cleanup_policy,
+            reset_every_n_runs=reset_every_n_runs,
+        )
+
+    def run(
+        self,
+        workflow_data: dict | None = None,
+        node_class_mappings: dict | None = None,
+        extra_pnginfo: dict | None = None,
+    ) -> Any:
+        return self._runtime.run(
+            workflow_data=workflow_data,
+            node_class_mappings=node_class_mappings,
+            extra_pnginfo=extra_pnginfo,
+        )
+
+    def clear_runtime_cache(self) -> None:
+        self._runtime.clear_runtime_cache()
+
+    def close(self, unload_models: bool = True) -> None:
+        self._runtime.close(unload_models=unload_models)

conftest.py

@@ -0,0 +1,6 @@
+"""Mock the ComfyUI 'server' module so the root __init__.py can be imported by pytest."""
+
+import sys
+from unittest.mock import MagicMock
+
+sys.modules["server"] = MagicMock()

pyproject.toml

@@ -2,16 +2,10 @@
 name = "comfyui-to-python-extension"
 description = "This custom node allows you to generate pure python code from your ComfyUI workflow with the click of a button. Great for rapid experimentation or production deployment."
 version = "2.1.0"
-license = "MIT"
+license = { text = "MIT License" }
 requires-python = ">=3.12"
 dependencies = ["black"]
 
-[tool.setuptools.packages.find]
-include = ["comfyui_to_python*"]
-
-[project.optional-dependencies]
-dev = ["ruff"]
-
 [project.urls]
 Repository = "https://github.com/pydn/ComfyUI-to-Python-Extension"
 #  Used by Comfy Registry https://comfyregistry.org

tests/conftest.py

@@ -0,0 +1,16 @@
+"""Pytest configuration to handle the ComfyUI extension __init__.py at repo root."""
+
+import sys
+from pathlib import Path
+
+import pytest
+
+
+@pytest.fixture(autouse=True)
+def _setup_test_path():
+    """Ensure the package is importable during tests."""
+    # Add the repo root to sys.path so `from comfyui_to_python import ...` works
+    repo_root = Path(__file__).parent.parent.resolve()
+    if str(repo_root) not in sys.path:
+        sys.path.insert(0, str(repo_root))
+    yield

tests/runtime/generated/text-to-image.py

@@ -1,154 +1,80 @@
 # Imports
-import importlib.util
 import json
-import logging
 import os
 import random
 import sys
 from typing import Sequence, Mapping, Any, Union
 
-log = logging.getLogger(__name__)
 
-
-def _find_from_extension_location() -> str | None:
-    """Walk up from this file's location to find ComfyUI root."""
-    ext_dir = os.path.dirname(os.path.realpath(__file__))
-    candidate = ext_dir
-    for _ in range(10):
-        parent = os.path.dirname(candidate)
-        if parent == candidate:
-            break
-        candidate = parent
-        if os.path.basename(candidate) == "ComfyUI":
-            if _is_comfyui_directory(candidate):
-                return candidate
-    return None
-
-
-def _is_comfyui_directory(path: str) -> bool:
-    """Verify a directory has ComfyUI structural markers.
-
-    Checks for nodes.py, main.py, and the comfy/ subdirectory to raise
-    the bar against spoofing via a directory with only a single marker file.
-    """
-    if not os.path.isdir(path):
-        return False
-    return (
-        os.path.isfile(os.path.join(path, "nodes.py"))
-        and os.path.isfile(os.path.join(path, "main.py"))
-        and os.path.isdir(os.path.join(path, "comfy"))
-    )
-
-
-def _load_module(module_name: str, filepath: str) -> Any:
-    """Load a Python module from an explicit file path, bypassing sys.path.
-
-    Significantly reduces bare import shadowing risk by loading modules
-    from verified file paths instead of relying on sys.path resolution.
-    If exec_module() raises, the partially-loaded module is removed from
-    sys.modules so subsequent calls start fresh.
-    """
-    # Return cached module if already loaded — prevents re-execution
-    # that would reset state (e.g. NODE_CLASS_MAPPINGS after init_extra_nodes).
-    if module_name in sys.modules:
-        return sys.modules[module_name]
-    if not os.path.isfile(filepath):
-        log.debug("Module file not found: %s (%s)", module_name, filepath)
-        return None
+def get_value_at_index(obj: Union[Sequence, Mapping], index: int) -> Any:
+    """Return a sequence or mapping result item by index."""
     try:
-        spec = importlib.util.spec_from_file_location(module_name, filepath)
-        if spec is None or spec.loader is None:
-            log.debug("Could not create spec for %s at %s", module_name, filepath)
-            return None
-        mod = importlib.util.module_from_spec(spec)
-        sys.modules[module_name] = mod
-        try:
-            spec.loader.exec_module(mod)
-        except BaseException:
-            sys.modules.pop(module_name, None)
-            raise
-        return mod
-    except BaseException as e:
-        log.debug("Failed to load %s from %s: %s", module_name, filepath, e)
-        sys.modules.pop(module_name, None)  # Also clean up on edge-case exceptions
+        return obj[index]
+    except KeyError:
+        return obj["result"][index]
+
+
+def get_comfyui_path() -> str:
+    """Return the configured ComfyUI path, preferring COMFYUI_PATH when set."""
+    comfyui_path = os.environ.get("COMFYUI_PATH")
+    if comfyui_path:
+        return comfyui_path
+    return find_path("ComfyUI")
+
+
+def find_path(name: str, path: str = None) -> str:
+    """Recursively search parent folders until the named entry is found."""
+    if path is None:
+        path = os.getcwd()
+
+    if name in os.listdir(path):
+        path_name = os.path.join(path, name)
+        print(f"{name} found: {path_name}")
+        return path_name
+
+    parent_directory = os.path.dirname(path)
+    if parent_directory == path:
         return None
 
-
-def _load_module_temp(module_name: str, filepath: str) -> Any:
-    """Load a module via _load_module() then remove it from sys.modules.
-
-    Used during bootstrap for modules that ComfyUI's import chain also loads
-    normally — prevents the cached copy from conflicting with later imports.
-    """
-    mod = _load_module(module_name, filepath)
-    sys.modules.pop(module_name, None)
-    return mod
+    return find_path(name, parent_directory)
 
 
 def add_comfyui_directory_to_sys_path() -> None:
-    """Add the ComfyUI checkout to sys.path (idempotent — insert-once, no gap)."""
+    """Add the ComfyUI checkout to sys.path."""
     comfyui_path = get_comfyui_path()
     if comfyui_path is not None and os.path.isdir(comfyui_path):
-        if comfyui_path not in sys.path:
-            sys.path.insert(0, comfyui_path)
-            log.debug("Added %s to sys.path", comfyui_path)
+        if comfyui_path in sys.path:
+            sys.path.remove(comfyui_path)
+        sys.path.insert(0, comfyui_path)
+        print(f"'{comfyui_path}' added to sys.path")
 
 
 def add_extra_model_paths() -> None:
     """Load ComfyUI extra model paths configuration when available."""
-    comfyui_path = get_comfyui_path()
-    if comfyui_path is None:
-        log.debug("Cannot load extra model paths: ComfyUI path not found")
-        return
-
-    # Try main.py first, then utils/extra_config.py — both via _load_module()
-    main_mod = _load_module("comfy_main", os.path.join(comfyui_path, "main.py"))
-    if main_mod is not None and hasattr(main_mod, "load_extra_path_config"):
-        load_extra_path_config = getattr(main_mod, "load_extra_path_config")
-    else:
-        log.debug("main.py not available, trying utils/extra_config.py")
-        extra_config_mod = _load_module(
-            "extra_config", os.path.join(comfyui_path, "utils", "extra_config.py")
+    try:
+        from main import load_extra_path_config
+    except ImportError:
+        print(
+            "Could not import load_extra_path_config from main.py. Looking in utils.extra_config instead."
         )
-        if extra_config_mod is None or not hasattr(
-            extra_config_mod, "load_extra_path_config"
-        ):
-            log.debug("Could not find load_extra_path_config in either path")
-            return
-        load_extra_path_config = getattr(extra_config_mod, "load_extra_path_config")
+        from utils.extra_config import load_extra_path_config
 
     extra_model_paths = find_path("extra_model_paths.yaml")
     if extra_model_paths is not None:
         load_extra_path_config(extra_model_paths)
     else:
-        log.debug("Could not find the extra_model_paths config file.")
+        print("Could not find the extra_model_paths config file.")
 
 
 def bootstrap_comfyui_runtime() -> None:
     """Mirror the allocator-related ComfyUI startup steps before torch import."""
     add_comfyui_directory_to_sys_path()
-    comfyui_path = get_comfyui_path()
-    if comfyui_path is None:
-        log.debug("bootstrap_comfyui_runtime: ComfyUI path not found")
-        return
 
-    # Use _load_module for file-based isolation, but with temporary names so
-    # the modules are removed from sys.modules after reading values.
-    # This prevents conflicts when ComfyUI's internal import chain (e.g.
-    # nodes.py -> comfy.cli_args) later loads these modules normally.
-    options_mod = _load_module_temp(
-        "_bootstrap_options", os.path.join(comfyui_path, "comfy", "options.py")
-    )
-    if options_mod is not None:
-        options_mod.enable_args_parsing()
+    import comfy.options
 
-    cli_args_mod = _load_module_temp(
-        "_bootstrap_cli_args", os.path.join(comfyui_path, "comfy", "cli_args.py")
-    )
-    args = getattr(cli_args_mod, "args", None) if cli_args_mod else None
+    comfy.options.enable_args_parsing()
 
-    if args is None:
-        return
+    from comfy.cli_args import args
 
     if os.name == "nt":
         os.environ["MIMALLOC_PURGE_DELAY"] = "0"
@@ -173,14 +99,9 @@ def bootstrap_comfyui_runtime() -> None:
     if args.deterministic and "CUBLAS_WORKSPACE_CONFIG" not in os.environ:
         os.environ["CUBLAS_WORKSPACE_CONFIG"] = ":4096:8"
 
-    cuda_malloc_mod = _load_module_temp(
-        "_bootstrap_cuda_malloc", os.path.join(comfyui_path, "cuda_malloc.py")
-    )
-    if (
-        cuda_malloc_mod is not None
-        and hasattr(cuda_malloc_mod, "get_torch_version_noimport")
-        and "rocm" in cuda_malloc_mod.get_torch_version_noimport()
-    ):
+    import cuda_malloc
+
+    if "rocm" in cuda_malloc.get_torch_version_noimport():
         os.environ["OCL_SET_SVM_SIZE"] = "262144"
 
 
@@ -224,51 +145,6 @@ def cleanup_comfyui_runtime(unload_models: bool | None = None) -> None:
     gc.collect()
 
 
-def find_path(name: str, max_depth: int = 20) -> str | None:
-    """Iteratively walk up from CWD to find a directory by name.
-
-    Depth-limited to prevent slow startup on deep trees.
-    Each candidate verified by _is_comfyui_directory() at the caller.
-    """
-    candidate = os.getcwd()
-    for _ in range(max_depth):
-        parent = os.path.dirname(candidate)
-        if parent == candidate:
-            break
-        candidate = parent
-        if os.path.basename(candidate) == name:
-            return candidate
-    return None
-
-
-def get_comfyui_path() -> str | None:
-    """Resolve ComfyUI path via prioritized multi-strategy fallback.
-
-    Strategy order:
-      1. COMFYUI_PATH env var (verified with _is_comfyui_directory)
-      2. Relative walk from extension location (realpath + verified)
-      3. CWD walk (legacy fallback, depth-limited)
-    """
-    p = os.environ.get("COMFYUI_PATH")
-    if p and _is_comfyui_directory(p):
-        return p
-    p = _find_from_extension_location()
-    if p:
-        return p
-    p = find_path("ComfyUI", max_depth=20)
-    if p and _is_comfyui_directory(p):
-        return p
-    return None
-
-
-def get_value_at_index(obj: Union[Sequence, Mapping], index: int) -> Any:
-    """Return a sequence or mapping result item by index."""
-    try:
-        return obj[index]
-    except KeyError:
-        return obj["result"][index]
-
-
 # Workflow data
 def build_workflow() -> dict[str, Any]:
     return {

tests/runtime/generated/upscale-model-loader.py

@@ -1,154 +1,80 @@
 # Imports
-import importlib.util
 import json
-import logging
 import os
 import random
 import sys
 from typing import Sequence, Mapping, Any, Union
 
-log = logging.getLogger(__name__)
 
-
-def _find_from_extension_location() -> str | None:
-    """Walk up from this file's location to find ComfyUI root."""
-    ext_dir = os.path.dirname(os.path.realpath(__file__))
-    candidate = ext_dir
-    for _ in range(10):
-        parent = os.path.dirname(candidate)
-        if parent == candidate:
-            break
-        candidate = parent
-        if os.path.basename(candidate) == "ComfyUI":
-            if _is_comfyui_directory(candidate):
-                return candidate
-    return None
-
-
-def _is_comfyui_directory(path: str) -> bool:
-    """Verify a directory has ComfyUI structural markers.
-
-    Checks for nodes.py, main.py, and the comfy/ subdirectory to raise
-    the bar against spoofing via a directory with only a single marker file.
-    """
-    if not os.path.isdir(path):
-        return False
-    return (
-        os.path.isfile(os.path.join(path, "nodes.py"))
-        and os.path.isfile(os.path.join(path, "main.py"))
-        and os.path.isdir(os.path.join(path, "comfy"))
-    )
-
-
-def _load_module(module_name: str, filepath: str) -> Any:
-    """Load a Python module from an explicit file path, bypassing sys.path.
-
-    Significantly reduces bare import shadowing risk by loading modules
-    from verified file paths instead of relying on sys.path resolution.
-    If exec_module() raises, the partially-loaded module is removed from
-    sys.modules so subsequent calls start fresh.
-    """
-    # Return cached module if already loaded — prevents re-execution
-    # that would reset state (e.g. NODE_CLASS_MAPPINGS after init_extra_nodes).
-    if module_name in sys.modules:
-        return sys.modules[module_name]
-    if not os.path.isfile(filepath):
-        log.debug("Module file not found: %s (%s)", module_name, filepath)
-        return None
+def get_value_at_index(obj: Union[Sequence, Mapping], index: int) -> Any:
+    """Return a sequence or mapping result item by index."""
     try:
-        spec = importlib.util.spec_from_file_location(module_name, filepath)
-        if spec is None or spec.loader is None:
-            log.debug("Could not create spec for %s at %s", module_name, filepath)
-            return None
-        mod = importlib.util.module_from_spec(spec)
-        sys.modules[module_name] = mod
-        try:
-            spec.loader.exec_module(mod)
-        except BaseException:
-            sys.modules.pop(module_name, None)
-            raise
-        return mod
-    except BaseException as e:
-        log.debug("Failed to load %s from %s: %s", module_name, filepath, e)
-        sys.modules.pop(module_name, None)  # Also clean up on edge-case exceptions
+        return obj[index]
+    except KeyError:
+        return obj["result"][index]
+
+
+def get_comfyui_path() -> str:
+    """Return the configured ComfyUI path, preferring COMFYUI_PATH when set."""
+    comfyui_path = os.environ.get("COMFYUI_PATH")
+    if comfyui_path:
+        return comfyui_path
+    return find_path("ComfyUI")
+
+
+def find_path(name: str, path: str = None) -> str:
+    """Recursively search parent folders until the named entry is found."""
+    if path is None:
+        path = os.getcwd()
+
+    if name in os.listdir(path):
+        path_name = os.path.join(path, name)
+        print(f"{name} found: {path_name}")
+        return path_name
+
+    parent_directory = os.path.dirname(path)
+    if parent_directory == path:
         return None
 
-
-def _load_module_temp(module_name: str, filepath: str) -> Any:
-    """Load a module via _load_module() then remove it from sys.modules.
-
-    Used during bootstrap for modules that ComfyUI's import chain also loads
-    normally — prevents the cached copy from conflicting with later imports.
-    """
-    mod = _load_module(module_name, filepath)
-    sys.modules.pop(module_name, None)
-    return mod
+    return find_path(name, parent_directory)
 
 
 def add_comfyui_directory_to_sys_path() -> None:
-    """Add the ComfyUI checkout to sys.path (idempotent — insert-once, no gap)."""
+    """Add the ComfyUI checkout to sys.path."""
     comfyui_path = get_comfyui_path()
     if comfyui_path is not None and os.path.isdir(comfyui_path):
-        if comfyui_path not in sys.path:
-            sys.path.insert(0, comfyui_path)
-            log.debug("Added %s to sys.path", comfyui_path)
+        if comfyui_path in sys.path:
+            sys.path.remove(comfyui_path)
+        sys.path.insert(0, comfyui_path)
+        print(f"'{comfyui_path}' added to sys.path")
 
 
 def add_extra_model_paths() -> None:
     """Load ComfyUI extra model paths configuration when available."""
-    comfyui_path = get_comfyui_path()
-    if comfyui_path is None:
-        log.debug("Cannot load extra model paths: ComfyUI path not found")
-        return
-
-    # Try main.py first, then utils/extra_config.py — both via _load_module()
-    main_mod = _load_module("comfy_main", os.path.join(comfyui_path, "main.py"))
-    if main_mod is not None and hasattr(main_mod, "load_extra_path_config"):
-        load_extra_path_config = getattr(main_mod, "load_extra_path_config")
-    else:
-        log.debug("main.py not available, trying utils/extra_config.py")
-        extra_config_mod = _load_module(
-            "extra_config", os.path.join(comfyui_path, "utils", "extra_config.py")
+    try:
+        from main import load_extra_path_config
+    except ImportError:
+        print(
+            "Could not import load_extra_path_config from main.py. Looking in utils.extra_config instead."
         )
-        if extra_config_mod is None or not hasattr(
-            extra_config_mod, "load_extra_path_config"
-        ):
-            log.debug("Could not find load_extra_path_config in either path")
-            return
-        load_extra_path_config = getattr(extra_config_mod, "load_extra_path_config")
+        from utils.extra_config import load_extra_path_config
 
     extra_model_paths = find_path("extra_model_paths.yaml")
     if extra_model_paths is not None:
         load_extra_path_config(extra_model_paths)
     else:
-        log.debug("Could not find the extra_model_paths config file.")
+        print("Could not find the extra_model_paths config file.")
 
 
 def bootstrap_comfyui_runtime() -> None:
     """Mirror the allocator-related ComfyUI startup steps before torch import."""
     add_comfyui_directory_to_sys_path()
-    comfyui_path = get_comfyui_path()
-    if comfyui_path is None:
-        log.debug("bootstrap_comfyui_runtime: ComfyUI path not found")
-        return
 
-    # Use _load_module for file-based isolation, but with temporary names so
-    # the modules are removed from sys.modules after reading values.
-    # This prevents conflicts when ComfyUI's internal import chain (e.g.
-    # nodes.py -> comfy.cli_args) later loads these modules normally.
-    options_mod = _load_module_temp(
-        "_bootstrap_options", os.path.join(comfyui_path, "comfy", "options.py")
-    )
-    if options_mod is not None:
-        options_mod.enable_args_parsing()
+    import comfy.options
 
-    cli_args_mod = _load_module_temp(
-        "_bootstrap_cli_args", os.path.join(comfyui_path, "comfy", "cli_args.py")
-    )
-    args = getattr(cli_args_mod, "args", None) if cli_args_mod else None
+    comfy.options.enable_args_parsing()
 
-    if args is None:
-        return
+    from comfy.cli_args import args
 
     if os.name == "nt":
         os.environ["MIMALLOC_PURGE_DELAY"] = "0"
@@ -173,14 +99,9 @@ def bootstrap_comfyui_runtime() -> None:
     if args.deterministic and "CUBLAS_WORKSPACE_CONFIG" not in os.environ:
         os.environ["CUBLAS_WORKSPACE_CONFIG"] = ":4096:8"
 
-    cuda_malloc_mod = _load_module_temp(
-        "_bootstrap_cuda_malloc", os.path.join(comfyui_path, "cuda_malloc.py")
-    )
-    if (
-        cuda_malloc_mod is not None
-        and hasattr(cuda_malloc_mod, "get_torch_version_noimport")
-        and "rocm" in cuda_malloc_mod.get_torch_version_noimport()
-    ):
+    import cuda_malloc
+
+    if "rocm" in cuda_malloc.get_torch_version_noimport():
         os.environ["OCL_SET_SVM_SIZE"] = "262144"
 
 
@@ -188,19 +109,6 @@ def cleanup_comfyui_runtime(unload_models: bool | None = None) -> None:
     """Best-effort cleanup for embedded or repeated generated-script execution."""
     import gc
 
-    def run_cleanup_hook(name: str, should_run: bool = True) -> None:
-        if not should_run or not hasattr(model_management, name):
-            return
-        cleanup_fn = getattr(model_management, name)
-        try:
-            cleanup_fn()
-        except Exception as exc:
-            warnings.warn(
-                f"ComfyUI cleanup hook {name} failed during teardown: {exc}",
-                RuntimeWarning,
-                stacklevel=2,
-            )
-
     should_unload = unload_models
     if should_unload is None:
         should_unload = os.environ.get(
@@ -218,111 +126,36 @@ def cleanup_comfyui_runtime(unload_models: bool | None = None) -> None:
         gc.collect()
         return
 
-    run_cleanup_hook("cleanup_models_gc")
-    run_cleanup_hook("unload_all_models", should_run=should_unload)
-    run_cleanup_hook("soft_empty_cache")
+    if hasattr(model_management, "cleanup_models_gc"):
+        model_management.cleanup_models_gc()
+    if should_unload and hasattr(model_management, "unload_all_models"):
+        model_management.unload_all_models()
+    if hasattr(model_management, "soft_empty_cache"):
+        model_management.soft_empty_cache()
     gc.collect()
 
 
-def find_path(name: str, max_depth: int = 20) -> str | None:
-    """Iteratively walk up from CWD to find a directory by name.
-
-    Depth-limited to prevent slow startup on deep trees.
-    Each candidate verified by _is_comfyui_directory() at the caller.
-    """
-    candidate = os.getcwd()
-    for _ in range(max_depth):
-        parent = os.path.dirname(candidate)
-        if parent == candidate:
-            break
-        candidate = parent
-        if os.path.basename(candidate) == name:
-            return candidate
-    return None
-
-
-def get_comfyui_path() -> str | None:
-    """Resolve ComfyUI path via prioritized multi-strategy fallback.
-
-    Strategy order:
-      1. COMFYUI_PATH env var (verified with _is_comfyui_directory)
-      2. Relative walk from extension location (realpath + verified)
-      3. CWD walk (legacy fallback, depth-limited)
-    """
-    p = os.environ.get("COMFYUI_PATH")
-    if p and _is_comfyui_directory(p):
-        return p
-    p = _find_from_extension_location()
-    if p:
-        return p
-    p = find_path("ComfyUI", max_depth=20)
-    if p and _is_comfyui_directory(p):
-        return p
-    return None
-
-
-def get_value_at_index(obj: Union[Sequence, Mapping], index: int) -> Any:
-    """Return a sequence or mapping result item by index."""
-    try:
-        return obj[index]
-    except KeyError:
-        return obj["result"][index]
-
-
 def import_custom_nodes() -> None:
-    """Initialize ComfyUI custom nodes in the exporter runtime.
-
-    Uses _load_module() for all ComfyUI imports — no bare imports,
-    no sys.path remove/re-insert gap.
-    """
+    """Initialize ComfyUI custom nodes in the exporter runtime."""
     comfyui_path = get_comfyui_path()
-    if comfyui_path is None:
-        log.debug("import_custom_nodes: ComfyUI path not found")
-        return
-
-    # Idempotent insert-once — never removes from sys.path
-    if comfyui_path not in sys.path:
+    if comfyui_path and comfyui_path not in sys.path:
         sys.path.insert(0, comfyui_path)
 
     import asyncio
+    import execution
+    from nodes import init_extra_nodes
 
-    execution_mod = _load_module(
-        "execution", os.path.join(comfyui_path, "execution.py")
-    )
-    nodes_mod = _load_module("nodes", os.path.join(comfyui_path, "nodes.py"))
+    if comfyui_path in sys.path:
+        sys.path.remove(comfyui_path)
+    sys.path.insert(0, comfyui_path)
 
-    # nodes.py inserts comfy/ subdirectory into sys.path, which shadows the
-    # top-level utils/ package (comfy/utils.py vs utils/). This breaks
-    # server.py → app.frontend_management → from utils.install_util import ...
-    # Filter it out temporarily so server.py loads cleanly.
-    comfy_subdir = os.path.join(comfyui_path, "comfy")
-    original_sys_path = list(sys.path)
-    sys.path[:] = [p for p in sys.path if p != comfy_subdir]
-
-    try:
-        server_mod = _load_module("server", os.path.join(comfyui_path, "server.py"))
-    finally:
-        # Restore sys.path so nodes and other modules that need comfy/ still work.
-        # Guaranteed even if _load_module raises mid-execution.
-        sys.path[:] = original_sys_path
-
-    if execution_mod is None or server_mod is None:
-        log.debug(
-            "import_custom_nodes: could not load execution/server modules. "
-            "Proceeding without full PromptServer/PromptQueue setup."
-        )
-        # Even without server, we can still populate NODE_CLASS_MAPPINGS from
-        # comfy_extras and built-in extra nodes.
-        if nodes_mod is not None and hasattr(nodes_mod, "init_extra_nodes"):
-            asyncio.run(nodes_mod.init_extra_nodes())
-        return
+    import server
 
     loop = asyncio.new_event_loop()
     asyncio.set_event_loop(loop)
-    server_instance = server_mod.PromptServer(loop)
-    execution_mod.PromptQueue(server_instance)
-    if nodes_mod is not None and hasattr(nodes_mod, "init_extra_nodes"):
-        asyncio.run(nodes_mod.init_extra_nodes())
+    server_instance = server.PromptServer(loop)
+    execution.PromptQueue(server_instance)
+    asyncio.run(init_extra_nodes())
 
 
 # Workflow data

tests/runtime/run_runtime_validation.py

@@ -320,6 +320,12 @@ def parse_args() -> argparse.Namespace:
         "--generated-path",
         help=argparse.SUPPRESS,
     )
+    parser.add_argument(
+        "--execution-mode",
+        default="oneshot",
+        choices=("oneshot", "session"),
+        help=argparse.SUPPRESS,
+    )
     args = parser.parse_args()
     if not args.internal_export and not args.tier:
         parser.error("--tier is required unless --internal-export is used.")
@@ -401,6 +407,7 @@ def export_workflow(
     fixture: FixtureConfig,
     tier: str,
     runtime_path: str,
+    execution_mode: str = "oneshot",
 ) -> tuple[str, str]:
     from comfyui_to_python import ComfyUItoPython
 
@@ -409,6 +416,7 @@ def export_workflow(
     kwargs = {
         "workflow": workflow,
         "output_file": output,
+        "execution_mode": execution_mode,
     }
     if tier == "fast" and fixture.fast_mapping_factory is not None:
         kwargs["node_class_mappings"] = fixture.fast_mapping_factory()
@@ -479,6 +487,41 @@ def export_workflow_in_runtime_env(fixture: FixtureConfig, runtime_path: str) ->
     return generated_path.read_text(encoding="utf-8")
 
 
+def export_session_workflow_in_runtime_env(
+    workflow_json: str,
+    execution_mode: str = "session",
+) -> str:
+    """Export a session workflow via subprocess in the ComfyUI runtime env.
+
+    Unlike ``export_workflow_in_runtime_env`` this does not require a fixture
+    config – it receives workflow JSON directly and re-enters the runtime
+    interpreter so ``ComfyUItoPython`` can import ComfyUI's nodes.
+    """
+    runtime_path = os.environ.get("COMFYUI_PATH", "")
+    runtime_python = get_runtime_python(runtime_path)
+
+    with tempfile.NamedTemporaryFile(
+        suffix=".json", mode="w", delete=False, encoding="utf-8"
+    ) as wf:
+        wf.write(workflow_json)
+        wf_path = wf.name
+
+    try:
+        temp_fixture = FixtureConfig(
+            name="session-mode-export",
+            path=Path(wf_path),
+        )
+        _, generated = export_workflow(
+            fixture=temp_fixture,
+            tier="runtime",
+            runtime_path=runtime_path,
+            execution_mode=execution_mode,
+        )
+        return generated
+    finally:
+        os.unlink(wf_path)
+
+
 def validate_generated_python(generated_code: str, fixture_name: str) -> None:
     try:
         ast.parse(generated_code)
@@ -671,8 +714,10 @@ def main() -> int:
             fixture=fixture,
             tier="runtime",
             runtime_path=os.environ.get("COMFYUI_PATH", ""),
+            execution_mode=args.execution_mode,
         )
         output_path.write_text(generated_code, encoding="utf-8")
+        print(generated_code, end="")
         return 0
 
     try:

tests/test_app_base_mappings.py

@@ -1,165 +0,0 @@
-"""Tests for ExportApplication base_node_class_mappings behavior.
-
-Verifies that after custom node import, base_node_class_mappings retains the
-pre-custom-node baseline so WorkflowPlanner correctly distinguishes between
-built-in nodes (direct import) and custom/extras nodes (NODE_CLASS_MAPPINGS lookup).
-"""
-
-import sys
-import unittest
-from io import StringIO
-
-
-def _make_workflow():
-    """Minimal workflow JSON string for testing."""
-    return '{"1": {"class_type": "CheckpointLoaderSimple", "inputs": {"ckpt_name": "test.safetensors"}}}'
-
-
-class TestBaseNodeClassMappings(unittest.TestCase):
-    """Tests for base_node_class_mappings stability across custom node reloads."""
-
-    def tearDown(self):
-        # Clean up any sys.modules pollution to avoid cross-test contamination
-        if "nodes" in sys.modules:
-            del sys.modules["nodes"]
-
-    def test_base_mappings_unchanged_after_custom_node_import(self):
-        """Given custom nodes are loaded, base_node_class_mappings stays at original baseline."""
-        from comfyui_to_python.app import ExportApplication
-
-        class FakeNode:
-            CATEGORY = "loaders"
-            FUNCTION = "execute"
-
-            @staticmethod
-            def INPUT_TYPES():
-                return {"required": {"ckpt_name": ("STRING",)}}
-
-            def execute(self, ckpt_name: str):
-                return ({},)
-
-        class FakeCustomNode:
-            CATEGORY = "custom"
-            FUNCTION = "execute"
-
-            @staticmethod
-            def INPUT_TYPES():
-                return {"required": {"ckpt_name": ("STRING",)}}
-
-            def execute(self, ckpt_name: str):
-                return ({},)
-
-        initial_mappings = {"CheckpointLoaderSimple": FakeNode}
-        after_import_mappings = dict(initial_mappings)
-        # Simulate custom nodes adding a new class
-        after_import_mappings["CustomNode"] = FakeCustomNode
-
-        call_count = [0]
-
-        def mock_node_loader():
-            return initial_mappings
-
-        def mock_custom_node_importer():
-            call_count[0] += 1
-            # Simulate populating sys.modules["nodes"].NODE_CLASS_MAPPINGS
-            nodes_mod_type = type("NodesModule", (), {})()
-            nodes_mod_type.NODE_CLASS_MAPPINGS = after_import_mappings
-            sys.modules["nodes"] = nodes_mod_type
-
-        output = StringIO()
-        app = ExportApplication(
-            workflow=_make_workflow(),
-            output_file=output,
-            node_mapping_loader=mock_node_loader,
-            custom_node_importer=mock_custom_node_importer,
-            needs_init_custom_nodes=True,
-        )
-
-        # base_node_class_mappings should be a deep copy of initial (pre-custom-node) state
-        self.assertEqual(app.base_node_class_mappings, initial_mappings)
-        app.execute()
-
-        # After execute(), base_node_class_mappings must still be the original baseline
-        # (not overwritten with after_import_mappings which includes custom nodes)
-        self.assertEqual(
-            app.base_node_class_mappings,
-            initial_mappings,
-            "base_node_class_mappings should remain at pre-custom-node baseline",
-        )
-
-    def test_base_mappings_is_deep_copy_not_shared_reference(self):
-        """Given node mappings change in-place, base_node_class_mappings is independent."""
-        from comfyui_to_python.app import ExportApplication
-
-        class FakeNode2:
-            CATEGORY = "loaders"
-            FUNCTION = "execute"
-
-            @staticmethod
-            def INPUT_TYPES():
-                return {"required": {"ckpt_name": ("STRING",)}}
-
-            def execute(self, ckpt_name: str):
-                return ({},)
-
-        initial_mappings = {"CheckpointLoaderSimple": FakeNode2}
-
-        def mock_node_loader():
-            return dict(initial_mappings)
-
-        output = StringIO()
-        app = ExportApplication(
-            workflow=_make_workflow(),
-            output_file=output,
-            node_mapping_loader=mock_node_loader,
-            custom_node_importer=lambda: None,
-        )
-
-        # Mutate the live mappings
-        app.node_class_mappings["NewKey"] = "value"
-
-        # base_node_class_mappings must not reflect the mutation (deep copy)
-        self.assertNotIn(
-            "NewKey",
-            app.base_node_class_mappings,
-            "base_node_class_mappings must be a deep copy, not shared reference",
-        )
-
-
-class TestRenderHardenedImports(unittest.TestCase):
-    """Integration test: render output contains hardened _load_module pattern."""
-
-    def test_generated_output_contains_exec_failure_cleanup(self):
-        """Given a GenerationPlan, rendered output includes try/except BaseException around exec_module()."""
-        from comfyui_to_python.generator.render import WorkflowRenderer
-        from comfyui_to_python.generator.model import GenerationPlan
-
-        plan = GenerationPlan(
-            workflow_data={"1": {"class_type": "CheckpointLoaderSimple"}},
-            metadata_workflow_data=None,
-            custom_nodes=False,
-            import_statements={},
-            special_functions_code=[],
-            loop_code=["result = some_node()"],
-            queue_size=1,
-        )
-        renderer = WorkflowRenderer()
-        generated = renderer.render(plan)
-
-        # The _load_module in rendered output must have the exec failure cleanup pattern
-        self.assertIn("try:", generated)
-        self.assertIn("spec.loader.exec_module(mod)", generated)
-        self.assertIn(
-            "except BaseException:",
-            generated,
-            "Must catch BaseException for edge-case safety",
-        )
-        self.assertIn(
-            "sys.modules.pop(module_name, None)",
-            generated,
-            "Must clean up sys.modules on exec failure",
-        )
-
-
-if __name__ == "__main__":
-    unittest.main()

tests/test_export_session_mode.py

@@ -0,0 +1,109 @@
+import json
+import unittest
+from io import StringIO
+
+from comfyui_to_python import ComfyUItoPython
+
+
+class DummyNode:
+    CATEGORY = "test"
+    FUNCTION = "execute"
+
+    @classmethod
+    def INPUT_TYPES(cls):
+        return {"required": {"value": ("STRING",)}}
+
+    def execute(self, value):
+        return (f"result:{value}",)
+
+
+class ExportSessionModeTest(unittest.TestCase):
+    """Tests for session mode export pipeline."""
+
+    def test_comfyui_to_python_passes_execution_mode(self):
+        workflow = {
+            "1": {
+                "class_type": "DummyNode",
+                "inputs": {"value": "test"},
+            }
+        }
+
+        output = StringIO()
+        ComfyUItoPython(
+            workflow=json.dumps(workflow),
+            output_file=output,
+            node_class_mappings={"DummyNode": DummyNode},
+            execution_mode="session",
+        )
+
+        generated = output.getvalue()
+
+        self.assertIn("class WorkflowSession:", generated)
+        self.assertIn("def run(self)", generated)
+        self.assertIn("def close(self, unload_models:", generated)
+
+    def test_session_mode_includes_backward_compat_main(self):
+        workflow = {
+            "1": {
+                "class_type": "DummyNode",
+                "inputs": {"value": "test"},
+            }
+        }
+
+        output = StringIO()
+        ComfyUItoPython(
+            workflow=json.dumps(workflow),
+            output_file=output,
+            node_class_mappings={"DummyNode": DummyNode},
+            execution_mode="session",
+        )
+
+        generated = output.getvalue()
+
+        self.assertIn("def main(", generated)
+        self.assertIn("WorkflowSession(", generated)
+        self.assertIn('cleanup_policy="per_run"', generated)
+        self.assertIn("if __name__ == \"__main__\":", generated)
+
+    def test_oneshot_mode_excludes_session_class(self):
+        workflow = {
+            "1": {
+                "class_type": "DummyNode",
+                "inputs": {"value": "test"},
+            }
+        }
+
+        output = StringIO()
+        ComfyUItoPython(
+            workflow=json.dumps(workflow),
+            output_file=output,
+            node_class_mappings={"DummyNode": DummyNode},
+            execution_mode="oneshot",
+        )
+
+        generated = output.getvalue()
+
+        self.assertNotIn("class WorkflowSession:", generated)
+
+    def test_default_mode_is_oneshot(self):
+        workflow = {
+            "1": {
+                "class_type": "DummyNode",
+                "inputs": {"value": "test"},
+            }
+        }
+
+        output = StringIO()
+        ComfyUItoPython(
+            workflow=json.dumps(workflow),
+            output_file=output,
+            node_class_mappings={"DummyNode": DummyNode},
+        )
+
+        generated = output.getvalue()
+
+        self.assertNotIn("class WorkflowSession:", generated)
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_import_path_resolution.py

@@ -1,712 +0,0 @@
-"""Tests for hardened import path resolution and importlib isolation layer.
-
-Covers Approach B from docs/specs/harden-import-path-resolution.html:
-- _is_comfyui_directory() structural verification
-- _load_module() centralized importlib isolation
-- get_comfyui_path() multi-strategy resolution
-- sys.path idempotence (no remove/re-insert gap)
-- Security: shadowing attack resistance
-- Logging hygiene (no print() path leaks)
-"""
-
-import os
-import sys
-import tempfile
-import unittest
-from unittest.mock import patch
-
-# Real ComfyUI checkout for integration-style unit tests (parent of this repo)
-_REAL_COMFYUI = os.path.realpath(
-    os.path.join(os.path.dirname(__file__), "..", "..", "ComfyUI")
-)
-
-
-def _make_fake_comfyui(tmpdir: str) -> str:
-    """Create a minimal fake ComfyUI directory in tmpdir with structural markers."""
-    comfyui_dir = os.path.join(tmpdir, "ComfyUI")
-    os.makedirs(comfyui_dir)
-    # Structural markers checked by _is_comfyui_directory()
-    with open(os.path.join(comfyui_dir, "nodes.py"), "w") as f:
-        f.write("# fake nodes.py\n")
-    with open(os.path.join(comfyui_dir, "main.py"), "w") as f:
-        f.write("# fake main.py\n")
-    os.makedirs(os.path.join(comfyui_dir, "comfy"))
-    return comfyui_dir
-
-
-class TestLoadModule(unittest.TestCase):
-    """Tests for _load_module() centralized importlib isolation."""
-
-    def tearDown(self):
-        # Clean up sys.modules cache between tests to avoid cross-test pollution
-        mod_names_to_clean = [
-            name for name in sys.modules if name.startswith("_test_mod_")
-        ]
-        for name in mod_names_to_clean:
-            del sys.modules[name]
-
-    def test_loads_module_from_explicit_file_path(self):
-        """Given a valid .py file, loads the module and returns it with expected attributes."""
-        from comfyui_to_python.node_runtime import _load_module
-
-        # Use a simple self-contained module (nodes.py requires torch)
-        with tempfile.TemporaryDirectory() as tmpdir:
-            mod_file = os.path.join(tmpdir, "mymodule.py")
-            with open(mod_file, "w") as f:
-                f.write("VALUE = 42\nNAME = 'test_module'\n")
-
-            mod = _load_module("_test_mod_simple", mod_file)
-        self.assertIsNotNone(mod)
-        self.assertEqual(mod.VALUE, 42)
-        self.assertEqual(mod.NAME, "test_module")
-
-    def test_returns_none_for_missing_file(self):
-        """Given a nonexistent file path, returns None without raising."""
-        from comfyui_to_python.node_runtime import _load_module
-
-        result = _load_module("_test_mod_missing", "/nonexistent/path/module.py")
-        self.assertIsNone(result)
-
-
-class TestShadowingResistance(unittest.TestCase):
-    """Security: _load_module() must ignore shadowed modules at sys.path[0]."""
-
-    def tearDown(self):
-        # Clean up sys.modules cache between tests
-        mod_names_to_clean = [
-            name for name in sys.modules if name.startswith("_sec_test_")
-        ]
-        for name in mod_names_to_clean:
-            del sys.modules[name]
-
-    def test_ignores_shadowed_main_at_syspath(self):
-        """Given a malicious main.py at sys.path[0], _load_module() loads from verified path."""
-        from comfyui_to_python.node_runtime import _load_module
-
-        with tempfile.TemporaryDirectory() as tmpdir:
-            # Legitimate module
-            legit_file = os.path.join(tmpdir, "legit_main.py")
-            with open(legit_file, "w") as f:
-                f.write("ORIGIN = 'legitimate'\n")
-
-            # Malicious shadow at sys.path[0]
-            attack_dir = os.path.join(tmpdir, "attack")
-            os.makedirs(attack_dir)
-            attack_file = os.path.join(attack_dir, "main.py")
-            with open(attack_file, "w") as f:
-                f.write("ORIGIN = 'malicious'\n")
-
-            # Insert malicious path at sys.path[0]
-            old_path = sys.path[:]
-            try:
-                sys.path.insert(0, attack_dir)
-                # _load_module with explicit legit path must return legitimate module
-                mod = _load_module("_sec_test_main", legit_file)
-            finally:
-                sys.path[:] = old_path
-
-        self.assertIsNotNone(mod)
-        self.assertEqual(mod.ORIGIN, "legitimate")
-
-    def test_ignores_shadowed_nodes_at_syspath(self):
-        """Given a malicious nodes.py at sys.path[0], _load_module() loads from verified path."""
-        from comfyui_to_python.node_runtime import _load_module
-
-        with tempfile.TemporaryDirectory() as tmpdir:
-            # Legitimate module
-            legit_file = os.path.join(tmpdir, "legit_nodes.py")
-            with open(legit_file, "w") as f:
-                f.write("NODE_CLASS_MAPPINGS = {'RealNode': 'real'}\n")
-
-            # Malicious shadow at sys.path[0]
-            attack_dir = os.path.join(tmpdir, "attack")
-            os.makedirs(attack_dir)
-            attack_file = os.path.join(attack_dir, "nodes.py")
-            with open(attack_file, "w") as f:
-                f.write("NODE_CLASS_MAPPINGS = {'FakeNode': 'malicious'}\n")
-
-            old_path = sys.path[:]
-            try:
-                sys.path.insert(0, attack_dir)
-                mod = _load_module("_sec_test_nodes", legit_file)
-            finally:
-                sys.path[:] = old_path
-
-        self.assertIsNotNone(mod)
-        self.assertEqual(mod.NODE_CLASS_MAPPINGS, {"RealNode": "real"})
-
-
-@unittest.skipUnless(os.path.isdir(_REAL_COMFYUI), "requires ../ComfyUI checkout")
-class TestSysPathIdempotence(unittest.TestCase):
-    """Tests for sys.path idempotence — no remove/re-insert gap."""
-
-    def test_add_comfyui_idempotent_no_remove_insert(self):
-        """Repeated calls to add_comfyui_directory_to_sys_path() are insert-once only."""
-        from comfyui_to_python.node_runtime import (
-            add_comfyui_directory_to_sys_path,
-        )
-
-        with patch.dict("os.environ", {"COMFYUI_PATH": _REAL_COMFYUI}, clear=False):
-            before_length = len(sys.path)
-            add_comfyui_directory_to_sys_path()
-            after_first = len(sys.path)
-
-            # Second call — must not change sys.path at all (no remove+re-insert)
-            add_comfyui_directory_to_sys_path()
-            after_second = len(sys.path)
-
-        self.assertEqual(after_first, after_second, "sys.path changed on second call")
-        # ComfyUI should be on sys.path
-        self.assertIn(_REAL_COMFYUI, sys.path)
-
-    def test_already_at_index_zero_is_noop(self):
-        """Given ComfyUI already at sys.path[0], add_comfyui_directory_to_sys_path() is a no-op."""
-        from comfyui_to_python.node_runtime import (
-            add_comfyui_directory_to_sys_path,
-        )
-
-        original_path = sys.path[:]
-        try:
-            with patch.dict("os.environ", {"COMFYUI_PATH": _REAL_COMFYUI}, clear=False):
-                # Put ComfyUI at index 0 manually
-                if _REAL_COMFYUI in sys.path:
-                    sys.path.remove(_REAL_COMFYUI)
-                sys.path.insert(0, _REAL_COMFYUI)
-                original_index = sys.path.index(_REAL_COMFYUI)
-
-                add_comfyui_directory_to_sys_path()
-                new_index = sys.path.index(_REAL_COMFYUI)
-        finally:
-            sys.path[:] = original_path
-
-        self.assertEqual(
-            original_index, new_index, "Index should not change when already at [0]"
-        )
-        self.assertEqual(new_index, 0)
-
-    def test_promotes_comfyui_to_sys_path_zero(self):
-        """Given ComfyUI at sys.path[5], add_comfyui_directory_to_sys_path() moves it to index 0."""
-        from comfyui_to_python.node_runtime import (
-            add_comfyui_directory_to_sys_path,
-        )
-
-        original_path = sys.path[:]
-        try:
-            with patch.dict("os.environ", {"COMFYUI_PATH": _REAL_COMFYUI}, clear=False):
-                # Insert ComfyUI deep in sys.path (simulating PYTHONPATH placement)
-                if _REAL_COMFYUI in sys.path:
-                    sys.path.remove(_REAL_COMFYUI)
-                sys.path.insert(5, _REAL_COMFYUI)
-
-                add_comfyui_directory_to_sys_path()
-                self.assertEqual(
-                    sys.path[0],
-                    _REAL_COMFYUI,
-                    "ComfyUI must be promoted to sys.path[0] when already present lower down",
-                )
-        finally:
-            sys.path[:] = original_path
-
-    def test_import_custom_nodes_no_gap_window(self):
-        """import_custom_nodes() must never remove ComfyUI from sys.path during execution."""
-        import inspect
-        from comfyui_to_python.node_runtime import import_custom_nodes
-
-        source = inspect.getsource(import_custom_nodes)
-        self.assertNotIn(
-            "sys.path.remove",
-            source,
-            "import_custom_nodes() must not remove ComfyUI from sys.path (gap window)",
-        )
-
-    def test_no_print_statements_in_node_runtime(self):
-        """node_runtime.py must use logging, not print(), to avoid path leaks to stdout."""
-        import inspect
-        from comfyui_to_python import node_runtime
-
-        source = inspect.getsource(node_runtime)
-        self.assertNotIn(
-            "print(",
-            source,
-            "node_runtime.py must not use print() — use logging.DEBUG instead",
-        )
-
-
-class TestGeneratedScriptIsolation(unittest.TestCase):
-    """Tests for generated script bootstrap (Phase 3)."""
-
-    def test_generated_script_embeds_load_module(self):
-        """render.py must embed _load_module() in generated scripts."""
-        from comfyui_to_python.generator.render import WorkflowRenderer
-        from comfyui_to_python.generator.model import GenerationPlan
-
-        plan = GenerationPlan(
-            workflow_data={"1": {"class_type": "CheckpointLoaderSimple"}},
-            metadata_workflow_data=None,
-            custom_nodes=False,
-            import_statements={},
-            special_functions_code=[],
-            loop_code=["result = some_node()"],
-            queue_size=1,
-        )
-        renderer = WorkflowRenderer()
-        generated = renderer.render(plan)
-
-        self.assertIn(
-            "_load_module", generated, "Generated script must embed _load_module()"
-        )
-        self.assertIn(
-            "spec_from_file_location",
-            generated,
-            "Generated script must use importlib isolation",
-        )
-
-    def test_generated_script_no_bare_comfyui_imports(self):
-        """Generated scripts must have zero bare imports of ComfyUI internals."""
-        from comfyui_to_python.generator.render import WorkflowRenderer
-        from comfyui_to_python.generator.model import GenerationPlan
-
-        plan = GenerationPlan(
-            workflow_data={"1": {"class_type": "CheckpointLoaderSimple"}},
-            metadata_workflow_data=None,
-            custom_nodes=True,
-            import_statements={},
-            special_functions_code=[],
-            loop_code=["result = some_node()"],
-            queue_size=1,
-        )
-        renderer = WorkflowRenderer()
-        generated = renderer.render(plan)
-
-        # These bare imports must NOT appear in generated output
-        bare_imports = [
-            "from main import",
-            "from utils.extra_config import",
-            "import execution",
-            "import server",
-            "from nodes import",
-            "import comfy.options",
-            "from comfy.cli_args import",
-        ]
-        for bare in bare_imports:
-            self.assertNotIn(
-                bare,
-                generated,
-                f"Generated script must not contain bare import: {bare}",
-            )
-
-
-class TestGetComfyuiPath(unittest.TestCase):
-    """Tests for get_comfyui_path() multi-strategy resolution."""
-
-    def test_resolves_via_valid_comfyui_path_env(self):
-        """Given COMFYUI_PATH env var set to valid ComfyUI dir, returns that path."""
-        from comfyui_to_python.node_runtime import get_comfyui_path
-
-        with tempfile.TemporaryDirectory() as tmpdir:
-            fake_comfyui = _make_fake_comfyui(tmpdir)
-            original_cwd = os.getcwd()
-            try:
-                os.chdir(fake_comfyui)  # strategy 2 would find it, but env wins first
-            finally:
-                pass
-            with patch.dict("os.environ", {"COMFYUI_PATH": fake_comfyui}, clear=False):
-                result = get_comfyui_path()
-            os.chdir(original_cwd)
-        self.assertEqual(result, fake_comfyui)
-
-    def test_falls_through_to_next_strategy_when_env_invalid(self):
-        """Given invalid COMFYUI_PATH (no nodes.py), rejects and falls through."""
-        from comfyui_to_python.node_runtime import get_comfyui_path
-
-        with patch.dict(
-            "os.environ", {"COMFYUI_PATH": "/tmp/not-comfyui"}, clear=False
-        ):
-            result = get_comfyui_path()
-        # The invalid path must be rejected (not returned as-is)
-        self.assertNotEqual(result, "/tmp/not-comfyui")
-        # In standalone/dev setups, no strategy may succeed -> None is valid
-        # In ComfyUI/custom_nodes/ install, relative strategy would find it
-
-
-class TestIsComfyuiDirectory(unittest.TestCase):
-    """Tests for _is_comfyui_directory() structural verification."""
-
-    def test_returns_true_for_real_comfyui_checkout(self):
-        """Given a valid ComfyUI directory with structural markers, returns True."""
-        from comfyui_to_python.node_runtime import _is_comfyui_directory
-
-        with tempfile.TemporaryDirectory() as tmpdir:
-            fake_comfyui = _make_fake_comfyui(tmpdir)
-            self.assertTrue(_is_comfyui_directory(fake_comfyui))
-
-    def test_rejects_directory_lacking_markers(self):
-        """Given a directory without nodes.py, returns False."""
-        from comfyui_to_python.node_runtime import _is_comfyui_directory
-
-        with tempfile.TemporaryDirectory() as tmpdir:
-            # Empty directory has no ComfyUI markers
-            self.assertFalse(_is_comfyui_directory(tmpdir))
-
-    def test_rejects_nonexistent_path(self):
-        """Given a nonexistent path, returns False."""
-        from comfyui_to_python.node_runtime import _is_comfyui_directory
-
-        self.assertFalse(_is_comfyui_directory("/nonexistent/path/to/comfyui"))
-
-
-class TestLoadModuleFailureCleanup(unittest.TestCase):
-    """Tests for _load_module() behavior when exec_module() raises."""
-
-    def tearDown(self):
-        mod_names_to_clean = [
-            name for name in sys.modules if name.startswith("_test_fail_")
-        ]
-        for name in mod_names_to_clean:
-            del sys.modules[name]
-
-    def test_broken_module_removed_from_sys_modules_on_failure(self):
-        """Given a module that raises during exec, it is removed from sys.modules."""
-        from comfyui_to_python.node_runtime import _load_module
-
-        with tempfile.TemporaryDirectory() as tmpdir:
-            mod_file = os.path.join(tmpdir, "badmodule.py")
-            with open(mod_file, "w") as f:
-                f.write("raise RuntimeError('mid-execution failure')\n")
-
-            result = _load_module("_test_fail_bad", mod_file)
-        self.assertIsNone(result)
-        self.assertNotIn(
-            "_test_fail_bad",
-            sys.modules,
-            "Failed module must not remain cached in sys.modules",
-        )
-
-    def test_broken_module_retried_cleanly(self):
-        """Given a failed load, a subsequent load attempt starts fresh (not from cache)."""
-        from comfyui_to_python.node_runtime import _load_module
-
-        with tempfile.TemporaryDirectory() as tmpdir:
-            bad_file = os.path.join(tmpdir, "bad.py")
-            good_file = os.path.join(tmpdir, "good.py")
-
-            # First load: bad module that raises
-            with open(bad_file, "w") as f:
-                f.write("raise RuntimeError('fail')\n")
-            result_bad = _load_module("_test_fail_retry", bad_file)
-            self.assertIsNone(result_bad)
-
-            # Overwrite with good module
-            os.remove(bad_file)
-            with open(good_file, "w") as f:
-                f.write("VALUE = 99\n")
-
-            # Reload same name from a different (working) file
-            result_good = _load_module("_test_fail_retry", good_file)
-        self.assertIsNotNone(result_good)
-        self.assertEqual(result_good.VALUE, 99)
-
-
-class TestCanonicalOptionsLoading(unittest.TestCase):
-    """Tests for canonical module loading of comfy.options in bootstrap."""
-
-    def tearDown(self):
-        # Clean up sys.modules cache
-        mod_names_to_clean = [
-            name for name in sys.modules if name.startswith("_test_opts_")
-        ]
-        for name in mod_names_to_clean:
-            del sys.modules[name]
-        if "comfy.options" in sys.modules:
-            del sys.modules["comfy.options"]
-
-    def test_load_under_canonical_name_caches_in_sys_modules(self):
-        """When _load_module() uses canonical name, the module is cached in sys.modules."""
-        from comfyui_to_python.node_runtime import _load_module
-
-        with tempfile.TemporaryDirectory() as tmpdir:
-            mod_file = os.path.join(tmpdir, "options.py")
-            with open(mod_file, "w") as f:
-                f.write("args_parsing = False\n")
-
-            # Load under canonical name
-            mod = _load_module("comfy.options", mod_file)
-        self.assertIsNotNone(mod)
-        # The canonical key must be in sys.modules so later imports see mutations
-        self.assertIn(
-            "comfy.options",
-            sys.modules,
-            "Canonical module load must cache under canonical name in sys.modules",
-        )
-
-
-class TestLoadModuleTemp(unittest.TestCase):
-    """Tests for _load_module_temp() removal from sys.modules."""
-
-    def tearDown(self):
-        mod_names_to_clean = [
-            name for name in sys.modules if name.startswith("_test_temp_")
-        ]
-        for name in mod_names_to_clean:
-            del sys.modules[name]
-
-    def test_module_removed_from_sys_modules_after_temp_load(self):
-        """After _load_module_temp(), the module key is absent from sys.modules."""
-        from comfyui_to_python.node_runtime import _load_module_temp
-
-        with tempfile.TemporaryDirectory() as tmpdir:
-            mod_file = os.path.join(tmpdir, "tempmod.py")
-            with open(mod_file, "w") as f:
-                f.write("VALUE = 123\n")
-
-            mod = _load_module_temp("_test_temp_mod", mod_file)
-        self.assertIsNotNone(mod)
-        self.assertEqual(mod.VALUE, 123)
-        self.assertNotIn(
-            "_test_temp_mod",
-            sys.modules,
-            "Temp module must be removed from sys.modules after load",
-        )
-
-    def test_module_removed_on_temp_load_failure(self):
-        """After _load_module_temp() fails, the module key is absent from sys.modules."""
-        from comfyui_to_python.node_runtime import _load_module_temp
-
-        with tempfile.TemporaryDirectory() as tmpdir:
-            mod_file = os.path.join(tmpdir, "badtemp.py")
-            with open(mod_file, "w") as f:
-                f.write("raise RuntimeError('fail')\n")
-
-            result = _load_module_temp("_test_temp_bad", mod_file)
-        self.assertIsNone(result)
-        self.assertNotIn(
-            "_test_temp_bad",
-            sys.modules,
-            "Failed temp module must be removed from sys.modules",
-        )
-
-
-class TestRepeatedNodeLoad(unittest.TestCase):
-    """Tests that repeated _load_module('nodes', ...) calls respect the cache."""
-
-    def tearDown(self):
-        if "_test_nodes" in sys.modules:
-            del sys.modules["_test_nodes"]
-
-    def test_repeated_load_preserves_cached_attributes(self):
-        """Second _load_module() for same name returns cached module, not a fresh copy."""
-        from comfyui_to_python.node_runtime import _load_module
-
-        with tempfile.TemporaryDirectory() as tmpdir:
-            mod_file = os.path.join(tmpdir, "mynodes.py")
-            with open(mod_file, "w") as f:
-                f.write("NODE_CLASS_MAPPINGS = {'A': 1}\n")
-
-            first = _load_module("_test_nodes", mod_file)
-            self.assertIsNotNone(first)
-
-            # Mutate the cached module to simulate init_extra_nodes()
-            first.NODE_CLASS_MAPPINGS["B"] = 2
-
-            # Second load with same name must return cached copy, preserving mutation
-            second = _load_module("_test_nodes", mod_file)
-        self.assertIs(first, second)
-        self.assertEqual(second.NODE_CLASS_MAPPINGS.get("B"), 2)
-
-
-class TestFindFile(unittest.TestCase):
-    """Tests for _find_file() — file discovery via directory walk."""
-
-    def test_finds_file_at_cwd_level(self):
-        """Given a file at CWD, _find_file() returns the full path to it."""
-        from comfyui_to_python.node_runtime import _find_file
-
-        original_cwd = os.getcwd()
-        with tempfile.TemporaryDirectory() as tmpdir:
-            target_file = os.path.join(tmpdir, "extra_model_paths.yaml")
-            with open(target_file, "w") as f:
-                f.write("config: value\n")
-
-            try:
-                os.chdir(tmpdir)
-                result = _find_file("extra_model_paths.yaml")
-            finally:
-                os.chdir(original_cwd)
-
-        self.assertEqual(result, target_file)
-
-    def test_finds_file_in_parent_directory(self):
-        """Given a file in a parent directory, _find_file() walks up and finds it."""
-        from comfyui_to_python.node_runtime import _find_file
-
-        original_cwd = os.getcwd()
-        with tempfile.TemporaryDirectory() as tmpdir:
-            target_file = os.path.join(tmpdir, "extra_model_paths.yaml")
-            with open(target_file, "w") as f:
-                f.write("config: value\n")
-
-            # Create a subdirectory and chdir into it
-            subdir = os.path.join(tmpdir, "subdir", "deeper")
-            os.makedirs(subdir)
-
-            try:
-                os.chdir(subdir)
-                result = _find_file("extra_model_paths.yaml")
-            finally:
-                os.chdir(original_cwd)
-
-        self.assertEqual(result, target_file)
-
-    def test_returns_none_when_file_not_found(self):
-        """Given a filename that doesn't exist within max_depth, returns None."""
-        from comfyui_to_python.node_runtime import _find_file
-
-        original_cwd = os.getcwd()
-        with tempfile.TemporaryDirectory() as tmpdir:
-            try:
-                os.chdir(tmpdir)
-                result = _find_file("nonexistent_config.yaml")
-            finally:
-                os.chdir(original_cwd)
-        self.assertIsNone(result)
-
-    def test_returns_none_for_directory_not_file(self):
-        """Given a directory with matching name, returns None (wants a file)."""
-        from comfyui_to_python.node_runtime import _find_file
-
-        original_cwd = os.getcwd()
-        with tempfile.TemporaryDirectory() as tmpdir:
-            # Create a directory (not file) with the target name
-            fake_dir = os.path.join(tmpdir, "extra_model_paths.yaml")
-            os.makedirs(fake_dir)
-
-            try:
-                os.chdir(tmpdir)
-                result = _find_file("extra_model_paths.yaml")
-            finally:
-                os.chdir(original_cwd)
-
-        self.assertIsNone(result)
-
-
-class TestFindPathCwdFirst(unittest.TestCase):
-    """Tests for find_path() checking CWD before walking to parent."""
-
-    def test_find_path_returns_cwd_when_cwd_is_match(self):
-        """Given CWD is named 'ComfyUI', find_path('ComfyUI') returns CWD itself."""
-        from comfyui_to_python.node_runtime import find_path
-
-        original_cwd = os.getcwd()
-        with tempfile.TemporaryDirectory() as tmpdir:
-            comfyui_dir = os.path.join(tmpdir, "ComfyUI")
-            os.makedirs(comfyui_dir)
-            try:
-                os.chdir(comfyui_dir)
-                result = find_path("ComfyUI")
-            finally:
-                os.chdir(original_cwd)
-
-        self.assertEqual(result, comfyui_dir)
-
-    def test_find_path_checks_cwd_before_parent(self):
-        """When CWD matches but parent also matches, CWD is returned (check-first)."""
-        from comfyui_to_python.node_runtime import find_path
-
-        original_cwd = os.getcwd()
-        with tempfile.TemporaryDirectory() as tmpdir:
-            # Create nested dirs: outer/ComfyUI/inner where inner is named ComfyUI too
-            outer_comfy = os.path.join(tmpdir, "ComfyUI")
-            os.makedirs(outer_comfy)
-            # Rename so we can create an inner dir also named ComfyUI
-            os.rename(outer_comfy, os.path.join(tmpdir, "ParentComfy"))
-            parent_comfy = os.path.join(tmpdir, "ParentComfy")
-            child_dir = os.path.join(parent_comfy, "subdir")
-            os.makedirs(child_dir)
-
-            try:
-                os.chdir(child_dir)
-                # When searching for ParentComfy, should find it at parent level
-                result = find_path("ParentComfy")
-            finally:
-                os.chdir(original_cwd)
-
-        self.assertEqual(result, parent_comfy)
-
-
-class TestGetComfyuiPathThirdStrategy(unittest.TestCase):
-    """Tests for get_comfyui_path() third-strategy verification."""
-
-    def test_fallback_rejects_non_comfyui_directory(self):
-        """Given a directory named ComfyUI without nodes.py, third strategy rejects it."""
-        from comfyui_to_python.node_runtime import get_comfyui_path
-
-        with tempfile.TemporaryDirectory() as tmpdir:
-            # Create a fake "ComfyUI" directory (no nodes.py)
-            fake_dir = os.path.join(tmpdir, "ComfyUI")
-            os.makedirs(fake_dir)
-
-            original_cwd = os.getcwd()
-            try:
-                os.chdir(tmpdir)
-                with patch.dict("os.environ", {}, clear=False):
-                    # Strip COMFYUI_PATH so fallback strategies activate
-                    env_copy = dict(os.environ)
-                    env_copy.pop("COMFYUI_PATH", None)
-                    with patch.dict("os.environ", env_copy, clear=True):
-                        result = get_comfyui_path()
-            finally:
-                os.chdir(original_cwd)
-
-        # Must NOT return the fake directory
-        self.assertNotEqual(result, fake_dir)
-
-
-@unittest.skipUnless(os.path.isdir(_REAL_COMFYUI), "requires ../ComfyUI checkout")
-class TestImportCustomNodesSysPathRestoration(unittest.TestCase):
-    """Tests for sys.path restoration in import_custom_nodes()."""
-
-    def test_sys_path_restored_on_crash(self):
-        """Given a crash during import_custom_nodes(), sys.path is still restored.
-
-        The crash fires on the "server" module load (after comfy/ filtering)
-        so that the try/finally sys.path restoration path is actually exercised.
-        """
-        from comfyui_to_python.node_runtime import import_custom_nodes
-
-        original_path = sys.path[:]
-        call_order = []
-
-        def fake_load(name, path):
-            call_order.append(name)
-            if name == "server":
-                raise RuntimeError("simulated crash during server load")
-            return None  # Let execution/nodes return None (falls through gracefully)
-
-        with (
-            patch(
-                "comfyui_to_python.node_runtime._load_module",
-                side_effect=fake_load,
-            ),
-            patch(
-                "comfyui_to_python.node_runtime.get_comfyui_path",
-                return_value=_REAL_COMFYUI if os.path.isdir(_REAL_COMFYUI) else None,
-            ),
-        ):
-            try:
-                import_custom_nodes()
-            except RuntimeError:
-                pass  # Expected crash
-
-        # When ComfyUI path is found, "execution" and "nodes" should be called
-        # before reaching the server module load inside try/finally.
-        self.assertIn("execution", call_order, "Expected 'execution' to be loaded")
-        self.assertIn("nodes", call_order, "Expected 'nodes' to be loaded")
-        self.assertIn("server", call_order, "Expected 'server' to trigger the crash")
-
-        # sys.path should still contain all original entries (no dangling state)
-        for entry in original_path:
-            self.assertIn(entry, sys.path)
-
-
-if __name__ == "__main__":
-    unittest.main()

tests/test_runtime_session.py

@@ -0,0 +1,469 @@
+import threading
+import unittest
+from unittest.mock import MagicMock, patch, call
+
+from comfyui_to_python.runtime_session import WorkflowSessionRuntime
+
+
+class StubNode:
+    FUNCTION = "execute"
+
+    @classmethod
+    def INPUT_TYPES(cls):
+        return {"required": {"value": ("STRING",)}}
+
+    def execute(self, value):
+        return (f"result:{value}",)
+
+
+class TestWorkflowSessionRuntimeInit(unittest.TestCase):
+    """Tests for WorkflowSessionRuntime initialization."""
+
+    def test_init_default_cleanup_policy_is_session(self):
+        runtime = WorkflowSessionRuntime()
+        self.assertEqual(runtime._cleanup_policy, "session")
+
+    def test_init_accepts_per_run_policy(self):
+        runtime = WorkflowSessionRuntime(cleanup_policy="per_run")
+        self.assertEqual(runtime._cleanup_policy, "per_run")
+
+    def test_init_accepts_manual_policy(self):
+        runtime = WorkflowSessionRuntime(cleanup_policy="manual")
+        self.assertEqual(runtime._cleanup_policy, "manual")
+
+    def test_init_rejects_invalid_policy(self):
+        with self.assertRaises(ValueError):
+            WorkflowSessionRuntime(cleanup_policy="invalid")
+
+    def test_init_sets_default_reset_every_n_runs_none(self):
+        runtime = WorkflowSessionRuntime()
+        self.assertIsNone(runtime._reset_every_n_runs)
+
+    def test_init_accepts_reset_every_n_runs(self):
+        runtime = WorkflowSessionRuntime(reset_every_n_runs=5)
+        self.assertEqual(runtime._reset_every_n_runs, 5)
+
+    def test_init_has_lock(self):
+        runtime = WorkflowSessionRuntime()
+        self.assertIsInstance(runtime._lock, type(threading.Lock()))
+
+    def test_init_bootstrapped_false(self):
+        runtime = WorkflowSessionRuntime()
+        self.assertFalse(runtime.bootstrapped)
+
+    def test_init_custom_nodes_initialized_false(self):
+        runtime = WorkflowSessionRuntime()
+        self.assertFalse(runtime.custom_nodes_initialized)
+
+    def test_init_run_count_zero(self):
+        runtime = WorkflowSessionRuntime()
+        self.assertEqual(runtime.run_count, 0)
+
+    def test_init_closed_false(self):
+        runtime = WorkflowSessionRuntime()
+        self.assertFalse(runtime._closed)
+
+
+class TestWorkflowSessionRuntimeLifecycle(unittest.TestCase):
+    """Tests for WorkflowSessionRuntime lifecycle management."""
+
+    def _make_runtime(self):
+        return WorkflowSessionRuntime()
+
+    @patch("comfyui_to_python.runtime_session.bootstrap_comfyui_runtime")
+    def test_ensure_bootstrapped_calls_bootstrap_once(self, mock_bootstrap):
+        runtime = self._make_runtime()
+        runtime._ensure_bootstrapped()
+        mock_bootstrap.assert_called_once()
+        self.assertTrue(runtime.bootstrapped)
+
+    @patch("comfyui_to_python.runtime_session.bootstrap_comfyui_runtime")
+    def test_ensure_bootstrapped_skips_if_already_bootstrapped(self, mock_bootstrap):
+        runtime = self._make_runtime()
+        runtime._ensure_bootstrapped()
+        runtime._ensure_bootstrapped()
+        self.assertEqual(mock_bootstrap.call_count, 1)
+
+    @patch("comfyui_to_python.runtime_session.import_custom_nodes")
+    def test_ensure_custom_nodes_init_calls_import_once(self, mock_import):
+        runtime = self._make_runtime()
+        runtime._ensure_custom_nodes_initialized()
+        mock_import.assert_called_once()
+        self.assertTrue(runtime.custom_nodes_initialized)
+
+    @patch("comfyui_to_python.runtime_session.import_custom_nodes")
+    def test_ensure_custom_nodes_init_skips_if_already_initialized(self, mock_import):
+        runtime = self._make_runtime()
+        runtime._ensure_custom_nodes_initialized()
+        runtime._ensure_custom_nodes_initialized()
+        self.assertEqual(mock_import.call_count, 1)
+
+    def test_close_sets_closed_flag(self):
+        runtime = self._make_runtime()
+        runtime.close(unload_models=True)
+        self.assertTrue(runtime._closed)
+
+    def test_close_is_idempotent(self):
+        runtime = self._make_runtime()
+        with patch.object(runtime, "_do_close") as mock_do_close:
+            runtime.close(unload_models=True)
+            runtime.close(unload_models=True)
+            self.assertEqual(mock_do_close.call_count, 1)
+
+
+class TestWorkflowSessionRuntimeExceptionSafety(unittest.TestCase):
+    """Tests that exceptions during run() do not corrupt session state."""
+
+    def _make_runtime(self):
+        return WorkflowSessionRuntime()
+
+    @patch("comfyui_to_python.runtime_session.WorkflowSessionRuntime._do_run")
+    def test_exception_preserves_state_flags(self, mock_do_run):
+        runtime = self._make_runtime()
+        runtime.bootstrapped = True
+        runtime.custom_nodes_initialized = True
+        mock_do_run.side_effect = RuntimeError("simulated failure")
+        with self.assertRaises(RuntimeError):
+            runtime.run()
+        self.assertTrue(runtime.bootstrapped)
+        self.assertTrue(runtime.custom_nodes_initialized)
+
+    @patch("comfyui_to_python.runtime_session.WorkflowSessionRuntime._do_run")
+    def test_run_count_not_incremented_on_exception(self, mock_do_run):
+        runtime = self._make_runtime()
+        mock_do_run.side_effect = RuntimeError("simulated failure")
+        with self.assertRaises(RuntimeError):
+            runtime.run()
+        self.assertEqual(runtime.run_count, 0)
+
+    def test_run_count_increments_after_successful_run(self):
+        runtime = self._make_runtime()
+        workflow_data = {
+            "1": {
+                "class_type": "StubNode",
+                "inputs": {"value": "test"},
+            }
+        }
+        node_mappings = {"StubNode": StubNode}
+        runtime._workflow_data = workflow_data
+        runtime._node_class_mappings = node_mappings
+        runtime.node_instances = {"StubNode": StubNode()}
+        runtime._node_classes = {"StubNode": StubNode}
+
+        runtime.run()
+
+        self.assertEqual(runtime.run_count, 1)
+
+    def test_run_count_resets_after_reset_every_n_runs(self):
+        runtime = WorkflowSessionRuntime(reset_every_n_runs=2)
+        workflow_data = {
+            "1": {
+                "class_type": "StubNode",
+                "inputs": {"value": "test"},
+            }
+        }
+        node_mappings = {"StubNode": StubNode}
+        runtime._workflow_data = workflow_data
+        runtime._node_class_mappings = node_mappings
+
+        runtime.run()
+        self.assertEqual(runtime.run_count, 1)
+        runtime.run()
+        self.assertEqual(runtime.run_count, 0)
+        self.assertFalse(runtime.bootstrapped)
+        self.assertFalse(runtime.custom_nodes_initialized)
+        self.assertEqual(runtime.node_instances, {})
+
+
+class TestWorkflowSessionRuntimeAlreadyClosed(unittest.TestCase):
+    """Tests for behavior after close()."""
+
+    def _make_runtime(self):
+        return WorkflowSessionRuntime()
+
+    def test_run_after_close_raises(self):
+        runtime = self._make_runtime()
+        runtime.close()
+        with self.assertRaises(RuntimeError):
+            runtime.run()
+
+
+class TestWorkflowSessionRuntimeNodeInstances(unittest.TestCase):
+    """Tests for cached node instance management."""
+
+    def _make_runtime(self):
+        return WorkflowSessionRuntime()
+
+    def test_ensure_node_instances_creates_and_caches_instances(self):
+        runtime = self._make_runtime()
+        node_class = MagicMock()
+        runtime._ensure_node_instances({"TestNode": node_class})
+        node_class.assert_called_once()
+        self.assertIn("TestNode", runtime.node_instances)
+        runtime._ensure_node_instances({"TestNode": node_class})
+        node_class.assert_called_once()
+
+    def test_node_instances_are_cached_across_runs(self):
+        runtime = WorkflowSessionRuntime()
+        workflow_data = {
+            "1": {
+                "class_type": "StubNode",
+                "inputs": {"value": "test"},
+            }
+        }
+        original_stub_init = StubNode.__init__
+        init_calls = []
+
+        def tracking_init(self, *args, **kwargs):
+            init_calls.append(1)
+            original_stub_init(self)
+
+        StubNode.__init__ = tracking_init
+        try:
+            runtime._workflow_data = workflow_data
+            runtime._node_class_mappings = {"StubNode": StubNode}
+
+            runtime.run()
+            runtime.run()
+
+            self.assertEqual(len(init_calls), 1)
+        finally:
+            StubNode.__init__ = original_stub_init
+
+
+class TestWorkflowSessionRuntimeClearRuntimeCache(unittest.TestCase):
+    """Tests for clear_runtime_cache behavior."""
+
+    def _make_runtime(self):
+        return WorkflowSessionRuntime()
+
+    @patch("comfyui_to_python.runtime_session.cleanup_comfyui_runtime")
+    def test_clear_runtime_cache_session_policy_skips_unload(self, mock_cleanup):
+        runtime = WorkflowSessionRuntime(cleanup_policy="session")
+        runtime.bootstrapped = True
+        runtime.clear_runtime_cache()
+        # session policy should NOT call unload_all_models
+        mock_cleanup.assert_not_called()
+
+    @patch("comfyui_to_python.runtime_session.cleanup_comfyui_runtime")
+    def test_clear_runtime_cache_per_run_policy_calls_full_cleanup(self, mock_cleanup):
+        runtime = WorkflowSessionRuntime(cleanup_policy="per_run")
+        runtime.bootstrapped = True
+        runtime.clear_runtime_cache()
+        mock_cleanup.assert_called_once_with(unload_models=True)
+
+    @patch("comfyui_to_python.runtime_session.cleanup_comfyui_runtime")
+    def test_clear_runtime_cache_manual_policy_no_cleanup(self, mock_cleanup):
+        runtime = WorkflowSessionRuntime(cleanup_policy="manual")
+        runtime.bootstrapped = True
+        runtime.clear_runtime_cache()
+        mock_cleanup.assert_not_called()
+
+
+class TestWorkflowSessionRuntimeDoClose(unittest.TestCase):
+    """Tests for _do_close internal method."""
+
+    def _make_runtime(self):
+        return WorkflowSessionRuntime()
+
+    @patch("comfyui_to_python.runtime_session.cleanup_comfyui_runtime")
+    @patch("comfyui_to_python.runtime_session.gc")
+    def test_do_close_calls_cleanup_and_gc(self, mock_gc, mock_cleanup):
+        runtime = self._make_runtime()
+        runtime._do_close(unload_models=True)
+        mock_cleanup.assert_called_once_with(unload_models=True)
+        mock_gc.collect.assert_called_once()
+
+        mock_cleanup.reset_mock()
+        mock_gc.reset_mock()
+        runtime = self._make_runtime()
+        runtime._do_close(unload_models=False)
+        mock_cleanup.assert_called_once_with(unload_models=False)
+        mock_gc.collect.assert_called_once()
+
+
+class TestWorkflowSessionRuntimeRun(unittest.TestCase):
+    """Tests for WorkflowSessionRuntime.run() workflow execution."""
+
+    def _make_runtime(self):
+        return WorkflowSessionRuntime()
+
+    def test_run_executes_workflow_nodes(self):
+        runtime = self._make_runtime()
+        workflow_data = {
+            "1": {
+                "class_type": "StubNode",
+                "inputs": {"value": "test"},
+            }
+        }
+        node_mappings = {"StubNode": StubNode}
+        runtime._workflow_data = workflow_data
+        runtime._node_class_mappings = node_mappings
+        runtime.node_instances = {"StubNode": StubNode()}
+        runtime._node_classes = {"StubNode": StubNode}
+
+        result = runtime.run()
+
+        self.assertIn("1", result)
+        self.assertEqual(result["1"], ["result:test"])
+
+    @patch(
+        "comfyui_to_python.runtime_session.WorkflowSessionRuntime._ensure_node_instances"
+    )
+    def test_run_with_no_workflow_data_returns_none(self, mock_ensure):
+        runtime = self._make_runtime()
+        result = runtime.run()
+        self.assertIsNone(result)
+
+    @patch(
+        "comfyui_to_python.runtime_session.WorkflowSessionRuntime._ensure_node_instances"
+    )
+    def test_run_with_no_node_mappings_returns_none(self, mock_ensure):
+        runtime = self._make_runtime()
+        runtime._workflow_data = {"1": {"class_type": "Test", "inputs": {}}}
+        result = runtime.run()
+        self.assertIsNone(result)
+
+    @patch(
+        "comfyui_to_python.runtime_session.WorkflowSessionRuntime._ensure_node_instances"
+    )
+    def test_run_node_not_in_mappings_is_skipped(self, mock_ensure):
+        workflow_data = {
+            "1": {
+                "class_type": "UnknownNode",
+                "inputs": {"value": "test"},
+            }
+        }
+        node_mappings = {"StubNode": StubNode}
+        runtime = self._make_runtime()
+        runtime._workflow_data = workflow_data
+        runtime._node_class_mappings = node_mappings
+
+        result = runtime.run()
+
+        self.assertEqual(result, {})
+
+    def test_run_node_returns_tuple_is_converted_to_list(self):
+        runtime = self._make_runtime()
+        workflow_data = {
+            "1": {
+                "class_type": "StubNode",
+                "inputs": {"value": "multi"},
+            }
+        }
+        node_mappings = {"StubNode": StubNode}
+        runtime._workflow_data = workflow_data
+        runtime._node_class_mappings = node_mappings
+        runtime.node_instances = {"StubNode": StubNode()}
+        runtime._node_classes = {"StubNode": StubNode}
+
+        result = runtime.run()
+
+        self.assertIsInstance(result["1"], list)
+        self.assertEqual(result["1"][0], "result:multi")
+
+    def test_run_persists_parameters_across_calls(self):
+        runtime = self._make_runtime()
+        workflow_data = {
+            "1": {
+                "class_type": "StubNode",
+                "inputs": {"value": "persist"},
+            }
+        }
+        node_mappings = {"StubNode": StubNode}
+
+        with patch.object(
+            runtime, "_do_run", return_value={"1": ["result:persist"]}
+        ) as mock_do_run:
+            runtime.run(
+                workflow_data=workflow_data, node_class_mappings=node_mappings
+            )
+            runtime.run()
+
+        self.assertEqual(mock_do_run.call_count, 2)
+
+    def test_run_does_not_corrupt_session_on_exception(self):
+        runtime = self._make_runtime()
+        workflow_data = {
+            "1": {
+                "class_type": "StubNode",
+                "inputs": {"value": "test"},
+            }
+        }
+        node_mappings = {"StubNode": StubNode}
+        runtime._workflow_data = workflow_data
+        runtime._node_class_mappings = node_mappings
+        mock_instance = MagicMock()
+        mock_instance.execute.side_effect = ValueError("boom")
+        runtime.node_instances = {"StubNode": mock_instance}
+        runtime._node_classes = {"StubNode": StubNode}
+
+        with self.assertRaises(ValueError):
+            runtime.run()
+
+        runtime.node_instances = {"StubNode": StubNode()}
+        runtime._node_classes = {"StubNode": StubNode}
+        result = runtime.run()
+        self.assertEqual(result["1"], ["result:test"])
+
+    @patch(
+        "comfyui_to_python.runtime_session.WorkflowSessionRuntime._ensure_node_instances"
+    )
+    def test_session_policy_does_not_clear_cache(self, mock_ensure):
+        runtime = WorkflowSessionRuntime(cleanup_policy="session")
+        workflow_data = {
+            "1": {
+                "class_type": "StubNode",
+                "inputs": {"value": "test"},
+            }
+        }
+        node_mappings = {"StubNode": StubNode}
+        runtime._workflow_data = workflow_data
+        runtime._node_class_mappings = node_mappings
+
+        with patch.object(runtime, "clear_runtime_cache") as mock_clear:
+            runtime.run()
+            mock_clear.assert_not_called()
+
+    @patch(
+        "comfyui_to_python.runtime_session.WorkflowSessionRuntime._ensure_node_instances"
+    )
+    def test_per_run_policy_clears_cache_after_each_run(self, mock_ensure):
+        runtime = WorkflowSessionRuntime(cleanup_policy="per_run")
+        workflow_data = {
+            "1": {
+                "class_type": "StubNode",
+                "inputs": {"value": "test"},
+            }
+        }
+        node_mappings = {"StubNode": StubNode}
+        runtime._workflow_data = workflow_data
+        runtime._node_class_mappings = node_mappings
+
+        with patch.object(runtime, "clear_runtime_cache") as mock_clear:
+            runtime.run()
+            mock_clear.assert_called_once()
+
+    @patch(
+        "comfyui_to_python.runtime_session.WorkflowSessionRuntime._ensure_node_instances"
+    )
+    def test_manual_policy_never_clears_cache(self, mock_ensure):
+        runtime = WorkflowSessionRuntime(cleanup_policy="manual")
+        workflow_data = {
+            "1": {
+                "class_type": "StubNode",
+                "inputs": {"value": "test"},
+            }
+        }
+        node_mappings = {"StubNode": StubNode}
+        runtime._workflow_data = workflow_data
+        runtime._node_class_mappings = node_mappings
+
+        with patch.object(runtime, "clear_runtime_cache") as mock_clear:
+            runtime.run()
+            runtime.run()
+            mock_clear.assert_not_called()
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_session_e2e.py

@@ -0,0 +1,370 @@
+"""End-to-end tests for session execution mode feature.
+
+Covers:
+- Code generation correctness (session vs oneshot)
+- Runtime execution of generated scripts
+- Multi-run session behavior
+"""
+
+import ast
+import json
+import os
+import subprocess
+import sys
+import tempfile
+import unittest
+from io import StringIO
+from pathlib import Path
+
+ROOT = Path(__file__).resolve().parents[1]
+COMFYUI_PATH = os.environ.get("COMFYUI_PATH", str(ROOT.parent / "ComfyUI"))
+
+# Minimal mock nodes for export
+class KSamplerMock:
+    CATEGORY = "sampling"
+    FUNCTION = "sample"
+    @classmethod
+    def INPUT_TYPES(cls):
+        return {
+            "required": {
+                "seed": ("INT", {"default": 0}),
+                "steps": ("INT", {"default": 20}),
+                "cfg": ("FLOAT", {"default": 8.0}),
+                "sampler_name": (["euler", "heun"],),
+                "scheduler": (["normal"],),
+                "denoise": ("FLOAT", {"default": 1.0}),
+                "model": ("MODEL",),
+                "positive": ("CONDITIONING",),
+                "negative": ("CONDITIONING",),
+                "latent_image": ("LATENT",),
+            }
+        }
+    def sample(self, seed, steps, cfg, sampler_name, scheduler, denoise, model, positive, negative, latent_image):
+        return ({"samples": latent_image},)
+
+class CheckpointLoaderMock:
+    CATEGORY = "loaders"
+    FUNCTION = "load_checkpoint"
+    @classmethod
+    def INPUT_TYPES(cls):
+        return {
+            "required": {
+                "ckpt_name": ("STRING",),
+            }
+        }
+    def load_checkpoint(self, ckpt_name):
+        return (None, None, None)
+
+class VAEDecodeMock:
+    CATEGORY = "latent"
+    FUNCTION = "decode"
+    @classmethod
+    def INPUT_TYPES(cls):
+        return {
+            "required": {
+                "samples": ("LATENT",),
+                "vae": ("VAE",),
+            }
+        }
+    def decode(self, samples, vae):
+        return (samples,)
+
+class CLIPTextEncodeMock:
+    CATEGORY = "conditioning"
+    FUNCTION = "encode"
+    @classmethod
+    def INPUT_TYPES(cls):
+        return {
+            "required": {
+                "text": ("STRING",),
+                "clip": ("CLIP",),
+            }
+        }
+    def encode(self, text, clip):
+        return ([],)
+
+class EmptyLatentImageMock:
+    CATEGORY = "latent"
+    FUNCTION = "generate"
+    @classmethod
+    def INPUT_TYPES(cls):
+        return {
+            "required": {
+                "width": ("INT", {"default": 512}),
+                "height": ("INT", {"default": 512}),
+                "batch_size": ("INT", {"default": 1}),
+            }
+        }
+    def generate(self, width, height, batch_size):
+        return ({"samples": {}},)
+
+class SaveImageMock:
+    CATEGORY = "image"
+    FUNCTION = "save"
+    @classmethod
+    def INPUT_TYPES(cls):
+        return {
+            "required": {
+                "images": ("IMAGE",),
+                "filename_prefix": ("STRING",),
+            }
+        }
+    def save(self, images, filename_prefix):
+        return ()
+
+NODENAMES = {
+    "CheckpointLoaderSimple": CheckpointLoaderMock,
+    "CLIPTextEncode": CLIPTextEncodeMock,
+    "KSampler": KSamplerMock,
+    "VAEDecode": VAEDecodeMock,
+    "EmptyLatentImage": EmptyLatentImageMock,
+    "SaveImage": SaveImageMock,
+}
+
+TEXT_TO_IMAGE_WORKFLOW = {
+    "1": {
+        "class_type": "CheckpointLoaderSimple",
+        "inputs": {"ckpt_name": "v1-5-pruned-emaonly-fp16.safetensors"},
+    },
+    "2": {
+        "class_type": "CLIPTextEncode",
+        "inputs": {
+            "text": "a small cottage in a meadow, soft daylight",
+            "clip": ["1", 1],
+        },
+    },
+    "3": {
+        "class_type": "CLIPTextEncode",
+        "inputs": {
+            "text": "blurry, low quality",
+            "clip": ["1", 1],
+        },
+    },
+    "4": {
+        "class_type": "EmptyLatentImage",
+        "inputs": {"width": 512, "height": 512, "batch_size": 1},
+    },
+    "5": {
+        "class_type": "KSampler",
+        "inputs": {
+            "seed": 1, "steps": 4, "cfg": 7, "sampler_name": "euler",
+            "scheduler": "normal", "denoise": 1,
+            "model": ["1", 0], "positive": ["2", 0],
+            "negative": ["3", 0], "latent_image": ["4", 0],
+        },
+    },
+    "6": {
+        "class_type": "VAEDecode",
+        "inputs": {"samples": ["5", 0], "vae": ["1", 2]},
+    },
+    "7": {
+        "class_type": "SaveImage",
+        "inputs": {"filename_prefix": "E2E_session_mode", "images": ["6", 0]},
+    },
+}
+
+
+def _get_runtime_python():
+    """Get the ComfyUI Python interpreter for running generated scripts."""
+    rt_python = Path(COMFYUI_PATH) / ".venv" / "bin" / "python"
+    if rt_python.is_file():
+        return str(rt_python)
+    return sys.executable
+
+
+def _export_session_workflow_in_runtime_env(
+    workflow_json,
+    execution_mode="session",
+):
+    """Export workflow via subprocess in ComfyUI runtime env.
+
+    Re-enters the runtime interpreter so ``ComfyUItoPython`` can import
+    ComfyUI's nodes.py (which requires torch, not available in the test venv).
+    """
+    runtime_python = _get_runtime_python()
+    env = os.environ.copy()
+    env["COMFYUI_PATH"] = COMFYUI_PATH
+    env["PYTHONPATH"] = os.pathsep.join([str(ROOT), env.get("PYTHONPATH", "")]).rstrip(
+        os.pathsep
+    )
+
+    tmp_path = tempfile.mktemp(suffix=".py")
+
+    try:
+        result = subprocess.run(
+            [
+                runtime_python,
+                str(Path(__file__).resolve().parents[0] / "runtime" / "run_runtime_validation.py"),
+                "--internal-export",
+                "text-to-image",
+                "--execution-mode",
+                execution_mode,
+                "--generated-path",
+                tmp_path,
+            ],
+            cwd=ROOT,
+            env=env,
+            capture_output=True,
+            text=True,
+        )
+        if result.returncode != 0:
+            output = (result.stderr or result.stdout or "").strip()
+            if "Missing runtime dependency" not in output and "ModuleNotFoundError" not in output:
+                classification = "repo regression"
+            else:
+                classification = "environment/setup failure"
+            raise RuntimeError(
+                f"Runtime export failed: [{classification}] {output}"
+            )
+        # Read generated code from the temp file (stdout is polluted by
+        # ComfyUI runtime prints such as the sys.path line).
+        return Path(tmp_path).read_text()
+    finally:
+        os.unlink(tmp_path)
+
+
+def _export_workflow(execution_mode="oneshot"):
+    """Export workflow to a string using ComfyUItoPython (unit tests, mock nodes)."""
+    from comfyui_to_python import ComfyUItoPython
+    output = StringIO()
+    ComfyUItoPython(
+        workflow=json.dumps(TEXT_TO_IMAGE_WORKFLOW),
+        output_file=output,
+        node_class_mappings=NODENAMES,
+        execution_mode=execution_mode,
+    )
+    return output.getvalue()
+
+
+class SessionCodeGenerationTest(unittest.TestCase):
+    """Unit tests for session mode code generation."""
+
+    def test_oneshot_code_has_bootstrap_helpers_no_session(self):
+        """Oneshot mode: bootstrap/cleanup helpers present, no WorkflowSession class."""
+        generated = _export_workflow("oneshot")
+        ast.parse(generated)
+        self.assertIn("bootstrap_comfyui_runtime()", generated)
+        self.assertIn("cleanup_comfyui_runtime(", generated)
+        self.assertNotIn("class WorkflowSession", generated)
+        self.assertNotIn("session.run()", generated)
+
+    def test_session_code_has_workflow_session_class_and_session_methods(self):
+        """Session mode: WorkflowSession class with run() and close() present."""
+        generated = _export_workflow("session")
+        ast.parse(generated)
+        self.assertIn("class WorkflowSession", generated)
+        self.assertIn("def run(self)", generated)
+        self.assertIn("def close(self, unload_models", generated)
+        self.assertIn("session.run()", generated)
+        self.assertIn("session.close(", generated)
+
+
+class SessionModeExecutionTest(unittest.TestCase):
+    """E2E tests for session mode script execution."""
+
+    @unittest.skipIf(not Path(COMFYUI_PATH).is_dir(), "ComfyUI checkout not available")
+    def test_oneshot_e2e_text_to_image(self):
+        """Oneshot mode: generate and run text-to-image workflow, verify PNG output."""
+        with tempfile.TemporaryDirectory() as tmpdir:
+            script_path = Path(tmpdir) / "oneshot.py"
+            env = os.environ.copy()
+            env["COMFYUI_PATH"] = COMFYUI_PATH
+            env["PYTHONPATH"] = os.pathsep.join([str(ROOT), env.get("PYTHONPATH", "")])
+            runtime_py = _get_runtime_python()
+
+            # Export inside ComfyUI env so node mappings resolve
+            generated = _export_session_workflow_in_runtime_env(
+                json.dumps(TEXT_TO_IMAGE_WORKFLOW),
+                execution_mode="oneshot",
+            )
+            script_path.write_text(generated)
+
+            # Run
+            result = subprocess.run(
+                [runtime_py, str(script_path), "--cpu"],
+                cwd=ROOT,
+                env=env,
+                capture_output=True,
+                text=True,
+                timeout=300,
+            )
+            self.assertEqual(result.returncode, 0, f"stderr: {result.stderr}")
+
+            # Check output
+            output_dir = Path(COMFYUI_PATH) / "output"
+            new_outputs = list(output_dir.glob("E2E_text_to_image*.png"))
+            self.assertTrue(len(new_outputs) > 0, "No PNG output produced")
+
+    @unittest.skipIf(not Path(COMFYUI_PATH).is_dir(), "ComfyUI checkout not available")
+    def test_session_e2e_text_to_image(self):
+        """Session mode: generate session-mode script, verify WorkflowSession present and run."""
+        with tempfile.TemporaryDirectory() as tmpdir:
+            script_path = Path(tmpdir) / "session.py"
+            env = os.environ.copy()
+            env["COMFYUI_PATH"] = COMFYUI_PATH
+            env["PYTHONPATH"] = os.pathsep.join([str(ROOT), env.get("PYTHONPATH", "")])
+            runtime_py = _get_runtime_python()
+
+            # Export inside ComfyUI env
+            generated = _export_session_workflow_in_runtime_env(
+                json.dumps(TEXT_TO_IMAGE_WORKFLOW),
+                execution_mode="session",
+            )
+            script_path.write_text(generated)
+
+            # Verify code structure
+            self.assertIn("class WorkflowSession", generated)
+
+            # Run
+            result = subprocess.run(
+                [runtime_py, str(script_path), "--cpu"],
+                cwd=ROOT,
+                env=env,
+                capture_output=True,
+                text=True,
+                timeout=300,
+            )
+            self.assertEqual(result.returncode, 0, f"stderr: {result.stderr}")
+
+            # Check output
+            output_dir = Path(COMFYUI_PATH) / "output"
+            new_outputs = list(output_dir.glob("E2E_text_to_image*.png"))
+            self.assertTrue(len(new_outputs) > 0, "No PNG output produced")
+
+    @unittest.skipIf(not Path(COMFYUI_PATH).is_dir(), "ComfyUI checkout not available")
+    def test_session_e2e_multiple_runs(self):
+        """Session mode: generate script that calls session.run() 3x in a row, verify no crash."""
+        with tempfile.TemporaryDirectory() as tmpdir:
+            script_path = Path(tmpdir) / "session_multi.py"
+            env = os.environ.copy()
+            env["COMFYUI_PATH"] = COMFYUI_PATH
+            env["PYTHONPATH"] = os.pathsep.join([str(ROOT), env.get("PYTHONPATH", "")])
+            runtime_py = _get_runtime_python()
+
+            # Export session mode inside ComfyUI env
+            generated = _export_session_workflow_in_runtime_env(
+                json.dumps(TEXT_TO_IMAGE_WORKFLOW),
+                execution_mode="session",
+            )
+
+            # Modify to run 3x
+            modified = generated.replace(
+                "        session.run()",
+                "        session.run()\n        session.run()\n        session.run()",
+            )
+            script_path.write_text(modified)
+
+            # Run
+            result = subprocess.run(
+                [runtime_py, str(script_path), "--cpu"],
+                cwd=ROOT,
+                env=env,
+                capture_output=True,
+                text=True,
+                timeout=600,
+            )
+            self.assertEqual(result.returncode, 0, f"stderr: {result.stderr}")
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_session_renderer.py

@@ -0,0 +1,273 @@
+import json
+import unittest
+from io import StringIO
+from unittest.mock import patch
+
+from comfyui_to_python import ComfyUItoPython
+
+
+class LoadImage:
+    CATEGORY = "image"
+    FUNCTION = "load_image"
+
+    @classmethod
+    def INPUT_TYPES(cls):
+        return {"required": {"image": ("STRING",)}}
+
+    def load_image(self, image):
+        return (image,)
+
+
+class SessionRendererTest(unittest.TestCase):
+    """Tests for session mode code generation in the renderer."""
+
+    def test_session_mode_generates_workflow_session_class(self):
+        workflow = {
+            "1": {
+                "class_type": "LoadImage",
+                "inputs": {"image": "example.png"},
+            }
+        }
+
+        output = StringIO()
+        ComfyUItoPython(
+            workflow=json.dumps(workflow),
+            output_file=output,
+            node_class_mappings={"LoadImage": LoadImage},
+            execution_mode="session",
+        )
+
+        generated = output.getvalue()
+        self.assertIn("class WorkflowSession", generated)
+
+    def test_session_mode_generates_main_wrapper(self):
+        workflow = {
+            "1": {
+                "class_type": "LoadImage",
+                "inputs": {"image": "example.png"},
+            }
+        }
+
+        output = StringIO()
+        ComfyUItoPython(
+            workflow=json.dumps(workflow),
+            output_file=output,
+            node_class_mappings={"LoadImage": LoadImage},
+            execution_mode="session",
+        )
+
+        generated = output.getvalue()
+        self.assertIn("def main(unload_models: bool | None = None)", generated)
+        self.assertIn("WorkflowSession(", generated)
+
+    def test_session_mode_main_creates_session_with_per_run_policy(self):
+        workflow = {
+            "1": {
+                "class_type": "LoadImage",
+                "inputs": {"image": "example.png"},
+            }
+        }
+
+        output = StringIO()
+        ComfyUItoPython(
+            workflow=json.dumps(workflow),
+            output_file=output,
+            node_class_mappings={"LoadImage": LoadImage},
+            execution_mode="session",
+        )
+
+        generated = output.getvalue()
+        self.assertIn('cleanup_policy="per_run"', generated)
+
+    def test_session_mode_main_has_try_finally_close(self):
+        workflow = {
+            "1": {
+                "class_type": "LoadImage",
+                "inputs": {"image": "example.png"},
+            }
+        }
+
+        output = StringIO()
+        ComfyUItoPython(
+            workflow=json.dumps(workflow),
+            output_file=output,
+            node_class_mappings={"LoadImage": LoadImage},
+            execution_mode="session",
+        )
+
+        generated = output.getvalue()
+        self.assertIn("try:", generated)
+        self.assertIn("session.run()", generated)
+        self.assertIn("finally:", generated)
+        self.assertIn("session.close(", generated)
+
+    def test_session_mode_generates_run_method(self):
+        workflow = {
+            "1": {
+                "class_type": "LoadImage",
+                "inputs": {"image": "example.png"},
+            }
+        }
+
+        output = StringIO()
+        ComfyUItoPython(
+            workflow=json.dumps(workflow),
+            output_file=output,
+            node_class_mappings={"LoadImage": LoadImage},
+            execution_mode="session",
+        )
+
+        generated = output.getvalue()
+        self.assertIn("def run(self", generated)
+
+    def test_session_mode_generates_close_method(self):
+        workflow = {
+            "1": {
+                "class_type": "LoadImage",
+                "inputs": {"image": "example.png"},
+            }
+        }
+
+        output = StringIO()
+        ComfyUItoPython(
+            workflow=json.dumps(workflow),
+            output_file=output,
+            node_class_mappings={"LoadImage": LoadImage},
+            execution_mode="session",
+        )
+
+        generated = output.getvalue()
+        self.assertIn("def close(self, unload_models: bool | None = None)", generated)
+
+    def test_session_mode_oneshot_generates_same_code(self):
+        workflow = {
+            "1": {
+                "class_type": "LoadImage",
+                "inputs": {"image": "example.png"},
+            }
+        }
+
+        output = StringIO()
+        ComfyUItoPython(
+            workflow=json.dumps(workflow),
+            output_file=output,
+            node_class_mappings={"LoadImage": LoadImage},
+            execution_mode="oneshot",
+        )
+
+        generated = output.getvalue()
+        self.assertNotIn("class WorkflowSession", generated)
+        self.assertIn("def main(unload_models: bool | None = None)", generated)
+        self.assertIn("bootstrap_comfyui_runtime()", generated)
+        self.assertIn("cleanup_comfyui_runtime(unload_models=unload_models)", generated)
+
+    def test_session_mode_oneshot_default(self):
+        workflow = {
+            "1": {
+                "class_type": "LoadImage",
+                "inputs": {"image": "example.png"},
+            }
+        }
+
+        output = StringIO()
+        ComfyUItoPython(
+            workflow=json.dumps(workflow),
+            output_file=output,
+            node_class_mappings={"LoadImage": LoadImage},
+        )
+
+        generated = output.getvalue()
+        self.assertNotIn("class WorkflowSession", generated)
+        self.assertIn("def main(unload_models: bool | None = None)", generated)
+
+    def test_session_mode_script_is_executable(self):
+        workflow = {
+            "1": {
+                "class_type": "LoadImage",
+                "inputs": {"image": "example.png"},
+            }
+        }
+
+        output = StringIO()
+        ComfyUItoPython(
+            workflow=json.dumps(workflow),
+            output_file=output,
+            node_class_mappings={"LoadImage": LoadImage},
+            execution_mode="session",
+        )
+
+        generated = output.getvalue()
+
+        # Should not raise — checks that generated code has valid syntax
+        compile(generated, "<generated>", "exec")
+
+        # main() should be callable
+        globals_dict = {"__name__": "generated_workflow_module"}
+        with patch("comfyui_to_python.runtime_session.WorkflowSessionRuntime"):
+            exec(generated, globals_dict)
+        self.assertIn("WorkflowSession", globals_dict)
+        self.assertIn("main", globals_dict)
+        self.assertTrue(callable(globals_dict["main"]))
+
+    def test_session_mode_generates_workflow_literal(self):
+        workflow = {
+            "1": {
+                "class_type": "LoadImage",
+                "inputs": {"image": "example.png"},
+            }
+        }
+
+        output = StringIO()
+        ComfyUItoPython(
+            workflow=json.dumps(workflow),
+            output_file=output,
+            node_class_mappings={"LoadImage": LoadImage},
+            execution_mode="session",
+        )
+
+        generated = output.getvalue()
+        self.assertIn("def build_workflow()", generated)
+        self.assertIn('return', generated)
+        self.assertIn('"class_type": "LoadImage"', generated)
+
+    def test_session_mode_generates_bootstrap_helper(self):
+        workflow = {
+            "1": {
+                "class_type": "LoadImage",
+                "inputs": {"image": "example.png"},
+            }
+        }
+
+        output = StringIO()
+        ComfyUItoPython(
+            workflow=json.dumps(workflow),
+            output_file=output,
+            node_class_mappings={"LoadImage": LoadImage},
+            execution_mode="session",
+        )
+
+        generated = output.getvalue()
+        self.assertIn("def bootstrap_comfyui_runtime()", generated)
+
+    def test_session_mode_generates_cleanup_helper(self):
+        workflow = {
+            "1": {
+                "class_type": "LoadImage",
+                "inputs": {"image": "example.png"},
+            }
+        }
+
+        output = StringIO()
+        ComfyUItoPython(
+            workflow=json.dumps(workflow),
+            output_file=output,
+            node_class_mappings={"LoadImage": LoadImage},
+            execution_mode="session",
+        )
+
+        generated = output.getvalue()
+        self.assertIn("def cleanup_comfyui_runtime(", generated)
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_upscale_model_loader_export.py

@@ -146,12 +146,9 @@ class UpscaleModelLoaderExportTest(unittest.TestCase):
 
         self.assertIn("def bootstrap_comfyui_runtime()", generated)
         self.assertIn("def cleanup_comfyui_runtime(", generated)
-        # Hardened imports use _load_module_temp() with temporary names,
-        # not bare 'import comfy.options' — avoids sys.modules pollution
-        self.assertIn("_load_module_temp", generated)
-        self.assertIn("options.py", generated)
-        self.assertIn("enable_args_parsing()", generated)
-        self.assertIn("cuda_malloc.py", generated)
+        self.assertIn("import comfy.options", generated)
+        self.assertIn("comfy.options.enable_args_parsing()", generated)
+        self.assertIn("import cuda_malloc", generated)
         self.assertNotIn("\nbootstrap_comfyui_runtime()\n", generated)
         self.assertIn(
             "def main(unload_models: bool | None = None):\n"
@@ -180,9 +177,7 @@ class UpscaleModelLoaderExportTest(unittest.TestCase):
             main_section.index("add_extra_model_paths()"),
             main_section.index("import torch"),
         )
-        # cuda_malloc is loaded via _load_module() inside bootstrap_comfyui_runtime()
-        # which runs before 'import torch' in main() — ordering preserved
-        self.assertIn("cuda_malloc", generated)
+        self.assertLess(generated.index("import cuda_malloc"), generated.index("import torch"))
         self.assertIn(
             "    finally:\n        cleanup_comfyui_runtime(unload_models=unload_models)",
             main_section,
@@ -388,13 +383,10 @@ class UpscaleModelLoaderExportTest(unittest.TestCase):
 
             input_file.write_text(json.dumps(workflow), encoding="utf-8")
 
-            with (
-                patch(
-                    "comfyui_to_python.get_node_class_mappings",
-                    return_value={"LoadImage": LoadImage},
-                ),
-                patch("comfyui_to_python.import_custom_nodes"),
-            ):
+            with patch(
+                "comfyui_to_python.get_node_class_mappings",
+                return_value={"LoadImage": LoadImage},
+            ), patch("comfyui_to_python.import_custom_nodes"):
                 run(
                     input_file=str(input_file),
                     output_file=str(output_file),

tests/test_workflow_session.py

@@ -0,0 +1,75 @@
+import unittest
+from unittest.mock import MagicMock, patch
+
+from comfyui_to_python.runtime_session import WorkflowSession
+
+
+class TestWorkflowSessionInit(unittest.TestCase):
+    """Tests for WorkflowSession initialization."""
+
+    def test_init_creates_runtime(self):
+        session = WorkflowSession()
+        self.assertIsNotNone(session._runtime)
+
+    def test_init_passes_cleanup_policy(self):
+        session = WorkflowSession(cleanup_policy="per_run")
+        self.assertEqual(session._runtime._cleanup_policy, "per_run")
+
+    def test_init_passes_reset_every_n_runs(self):
+        session = WorkflowSession(reset_every_n_runs=5)
+        self.assertEqual(session._runtime._reset_every_n_runs, 5)
+
+
+class TestWorkflowSessionDelegation(unittest.TestCase):
+    """Tests for WorkflowSession public API delegation to internal runtime."""
+
+    def _make_session(self):
+        return WorkflowSession()
+
+    def test_run_delegates_to_runtime(self):
+        session = self._make_session()
+        workflow_data = {
+            "1": {
+                "class_type": "StubNode",
+                "inputs": {"value": "test"},
+            }
+        }
+        node_mappings = {"StubNode": MagicMock()}
+        session._runtime.node_instances = {"StubNode": MagicMock()}
+        session._runtime._node_classes = {"StubNode": MagicMock()}
+
+        with patch.object(
+            session._runtime, "run", return_value={"1": ["result"]}
+        ) as mock_run:
+            session.run(workflow_data=workflow_data, node_class_mappings=node_mappings)
+            mock_run.assert_called_once_with(
+                workflow_data=workflow_data,
+                node_class_mappings=node_mappings,
+                extra_pnginfo=None,
+            )
+
+    def test_clear_runtime_cache_delegates_to_runtime(self):
+        session = self._make_session()
+        with patch.object(
+            session._runtime, "clear_runtime_cache"
+        ) as mock_clear:
+            session.clear_runtime_cache()
+            mock_clear.assert_called_once()
+
+    def test_close_delegates_to_runtime(self):
+        session = self._make_session()
+        with patch.object(
+            session._runtime, "close"
+        ) as mock_close:
+            session.close(unload_models=True)
+            mock_close.assert_called_once_with(unload_models=True)
+
+    def test_run_raises_after_close(self):
+        session = self._make_session()
+        session.close()
+        with self.assertRaises(RuntimeError):
+            session.run()
+
+
+if __name__ == "__main__":
+    unittest.main()

uv.lock

@@ -63,17 +63,8 @@ dependencies = [
     { name = "black" },
 ]
 
-[package.optional-dependencies]
-dev = [
-    { name = "ruff" },
-]
-
 [package.metadata]
-requires-dist = [
-    { name = "black" },
-    { name = "ruff", marker = "extra == 'dev'" },
-]
-provides-extras = ["dev"]
+requires-dist = [{ name = "black" }]
 
 [[package]]
 name = "mypy-extensions"
@@ -139,28 +130,3 @@ wheels = [
     { url = "https://files.pythonhosted.org/packages/39/2a/cbbf9250020a4a8dd53ba83a46c097b69e5eb49dd14e708f496f548c6612/pytokens-0.4.1-cp314-cp314t-win_amd64.whl", hash = "sha256:584c80c24b078eec1e227079d56dc22ff755e0ba8654d8383b2c549107528918", size = 116287, upload-time = "2026-01-30T01:03:30.912Z" },
     { url = "https://files.pythonhosted.org/packages/c6/78/397db326746f0a342855b81216ae1f0a32965deccfd7c830a2dbc66d2483/pytokens-0.4.1-py3-none-any.whl", hash = "sha256:26cef14744a8385f35d0e095dc8b3a7583f6c953c2e3d269c7f82484bf5ad2de", size = 13729, upload-time = "2026-01-30T01:03:45.029Z" },
 ]
-
-[[package]]
-name = "ruff"
-version = "0.15.12"
-source = { registry = "https://pypi.org/simple" }
-sdist = { url = "https://files.pythonhosted.org/packages/99/43/3291f1cc9106f4c63bdce7a8d0df5047fe8422a75b091c16b5e9355e0b11/ruff-0.15.12.tar.gz", hash = "sha256:ecea26adb26b4232c0c2ca19ccbc0083a68344180bba2a600605538ce51a40a6", size = 4643852, upload-time = "2026-04-24T18:17:14.305Z" }
-wheels = [
-    { url = "https://files.pythonhosted.org/packages/c3/6e/e78ffb61d4686f3d96ba3df2c801161843746dcbcbb17a1e927d4829312b/ruff-0.15.12-py3-none-linux_armv6l.whl", hash = "sha256:f86f176e188e94d6bdbc09f09bfd9dc729059ad93d0e7390b5a73efe19f8861c", size = 10640713, upload-time = "2026-04-24T18:17:22.841Z" },
-    { url = "https://files.pythonhosted.org/packages/ae/08/a317bc231fb9e7b93e4ef3089501e51922ff88d6936ce5cf870c4fe55419/ruff-0.15.12-py3-none-macosx_10_12_x86_64.whl", hash = "sha256:e3bcd123364c3770b8e1b7baaf343cc99a35f197c5c6e8af79015c666c423a6c", size = 11069267, upload-time = "2026-04-24T18:17:30.105Z" },
-    { url = "https://files.pythonhosted.org/packages/aa/a4/f828e9718d3dce1f5f11c39c4f65afd32783c8b2aebb2e3d259e492c47bd/ruff-0.15.12-py3-none-macosx_11_0_arm64.whl", hash = "sha256:fe87510d000220aa1ed530d4448a7c696a0cae1213e5ec30e5874287b66557b5", size = 10397182, upload-time = "2026-04-24T18:17:07.177Z" },
-    { url = "https://files.pythonhosted.org/packages/71/e0/3310fc6d1b5e1fdea22bf3b1b807c7e187b581021b0d7d4514cccdb5fb71/ruff-0.15.12-py3-none-manylinux_2_17_aarch64.manylinux2014_aarch64.whl", hash = "sha256:84a1630093121375a3e2a95b4a6dc7b59e2b4ee76216e32d81aae550a832d002", size = 10758012, upload-time = "2026-04-24T18:16:55.759Z" },
-    { url = "https://files.pythonhosted.org/packages/11/c1/a606911aee04c324ddaa883ae418f3569792fd3c4a10c50e0dd0a2311e1e/ruff-0.15.12-py3-none-manylinux_2_17_armv7l.manylinux2014_armv7l.whl", hash = "sha256:fb129f40f114f089ebe0ca56c0d251cf2061b17651d464bb6478dc01e69f11f5", size = 10447479, upload-time = "2026-04-24T18:16:51.677Z" },
-    { url = "https://files.pythonhosted.org/packages/9d/68/4201e8444f0894f21ab4aeeaee68aa4f10b51613514a20d80bd628d57e88/ruff-0.15.12-py3-none-manylinux_2_17_i686.manylinux2014_i686.whl", hash = "sha256:b0c862b172d695db7598426b8af465e7e9ac00a3ea2a3630ee67eb82e366aaa6", size = 11234040, upload-time = "2026-04-24T18:17:16.529Z" },
-    { url = "https://files.pythonhosted.org/packages/34/ff/8a6d6cf4ccc23fd67060874e832c18919d1557a0611ebef03fdb01fff11e/ruff-0.15.12-py3-none-manylinux_2_17_ppc64le.manylinux2014_ppc64le.whl", hash = "sha256:2849ea9f3484c3aca43a82f484210370319e7170df4dfe4843395ddf6c57bc33", size = 12087377, upload-time = "2026-04-24T18:17:04.944Z" },
-    { url = "https://files.pythonhosted.org/packages/85/f6/c669cf73f5152f623d34e69866a46d5e6185816b19fcd5b6dd8a2d299922/ruff-0.15.12-py3-none-manylinux_2_17_s390x.manylinux2014_s390x.whl", hash = "sha256:9e77c7e51c07fe396826d5969a5b846d9cd4c402535835fb6e21ce8b28fef847", size = 11367784, upload-time = "2026-04-24T18:17:25.409Z" },
-    { url = "https://files.pythonhosted.org/packages/e8/39/c61d193b8a1daaa8977f7dea9e8d8ba866e02ea7b65d32f6861693aa4c12/ruff-0.15.12-py3-none-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:83b2f4f2f3b1026b5fb449b467d9264bf22067b600f7b6f41fc5958909f449d0", size = 11344088, upload-time = "2026-04-24T18:17:12.258Z" },
-    { url = "https://files.pythonhosted.org/packages/c2/8d/49afab3645e31e12c590acb6d3b5b69d7aab5b81926dbaf7461f9441f37a/ruff-0.15.12-py3-none-manylinux_2_31_riscv64.whl", hash = "sha256:9ba3b8f1afd7e2e43d8943e55f249e13f9682fde09711644a6e7290eb4f3e339", size = 11271770, upload-time = "2026-04-24T18:17:02.457Z" },
-    { url = "https://files.pythonhosted.org/packages/46/06/33f41fe94403e2b755481cdfb9b7ef3e4e0ed031c4581124658d935d52b4/ruff-0.15.12-py3-none-musllinux_1_2_aarch64.whl", hash = "sha256:e852ba9fdc890655e1d78f2df1499efbe0e54126bd405362154a75e2bde159c5", size = 10719355, upload-time = "2026-04-24T18:17:27.648Z" },
-    { url = "https://files.pythonhosted.org/packages/0d/59/18aa4e014debbf559670e4048e39260a85c7fcee84acfd761ac01e7b8d35/ruff-0.15.12-py3-none-musllinux_1_2_armv7l.whl", hash = "sha256:dd8aed930da53780d22fc70bdf84452c843cf64f8cb4eb38984319c24c5cd5fd", size = 10462758, upload-time = "2026-04-24T18:17:32.347Z" },
-    { url = "https://files.pythonhosted.org/packages/25/e7/cc9f16fd0f3b5fddcbd7ec3d6ae30c8f3fde1047f32a4093a98d633c6570/ruff-0.15.12-py3-none-musllinux_1_2_i686.whl", hash = "sha256:01da3988d225628b709493d7dc67c3b9b12c0210016b08690ef9bd27970b262b", size = 10953498, upload-time = "2026-04-24T18:17:20.674Z" },
-    { url = "https://files.pythonhosted.org/packages/72/7a/a9ba7f98c7a575978698f4230c5e8cc54bbc761af34f560818f933dafa0c/ruff-0.15.12-py3-none-musllinux_1_2_x86_64.whl", hash = "sha256:9cae0f92bd5700d1213188b31cd3bdd2b315361296d10b96b8e2337d3d11f53e", size = 11447765, upload-time = "2026-04-24T18:17:09.755Z" },
-    { url = "https://files.pythonhosted.org/packages/ea/f9/0ae446942c846b8266059ad8a30702a35afae55f5cdc54c5adf8d7afdc27/ruff-0.15.12-py3-none-win32.whl", hash = "sha256:d0185894e038d7043ba8fd6aee7499ece6462dc0ea9f1e260c7451807c714c20", size = 10657277, upload-time = "2026-04-24T18:17:18.591Z" },
-    { url = "https://files.pythonhosted.org/packages/33/f1/9614e03e1cdcbf9437570b5400ced8a720b5db22b28d8e0f1bda429f660d/ruff-0.15.12-py3-none-win_amd64.whl", hash = "sha256:c87a162d61ab3adca47c03f7f717c68672edec7d1b5499e652331780fe74950d", size = 11837758, upload-time = "2026-04-24T18:17:00.113Z" },
-    { url = "https://files.pythonhosted.org/packages/c0/98/6beb4b351e472e5f4c4613f7c35a5290b8be2497e183825310c4c3a3984b/ruff-0.15.12-py3-none-win_arm64.whl", hash = "sha256:a538f7a82d061cee7be55542aca1d86d1393d55d81d4fcc314370f4340930d4f", size = 11120821, upload-time = "2026-04-24T18:16:57.979Z" },
-]