refactor: harden readability contracts

Author: pydn-hermes-agent

comfyui_to_python/generator/embedded_modules.py

@@ -14,19 +14,46 @@
 
 import ast
 from pathlib import Path
-from typing import Any
-
 
 # Modules whose top-level definitions should be embedded in generated scripts.
-# Order matters: dependencies first, so functions are defined before callers reference them.
+# Order matters: dependencies first, so functions are defined before callers use them.
 _SOURCE_FILES: list[str] = [
     "runtime/module_loader.py",  # _load_module, _bootstrap_import — no internal deps
-    "runtime/path_discovery.py",  # get_comfyui_path, find_path, etc. — uses stdlib only
-    "runtime/bootstrap.py",  # _discover_cli_options, _filter_args — depends on module_loader
-    "node_runtime.py",  # public API facade + bootstrap/cleanup — imports from all above
+    "runtime/path_discovery.py",  # get_comfyui_path, find_path, etc.
+    "runtime/bootstrap.py",  # CLI filtering; depends on module_loader
+    "node_runtime.py",  # public API facade + bootstrap/cleanup
 ]
 
 
+APPROVED_EMBEDDED_NAMES: frozenset[str] = frozenset(
+    {
+        "_apply_device_settings",
+        "_apply_directory_overrides",
+        "_bootstrap_import",
+        "_discover_comfyui_cli_options",
+        "_filter_comfyui_args",
+        "_find_file",
+        "_find_from_extension_location",
+        "_get_base_option",
+        "_init_extra_nodes",
+        "_is_comfyui_directory",
+        "_load_custom_node_modules",
+        "_load_module",
+        "_load_module_temp",
+        "_parse_parser_actions",
+        "add_comfyui_directory_to_sys_path",
+        "add_extra_model_paths",
+        "bootstrap_comfyui_runtime",
+        "cleanup_comfyui_runtime",
+        "find_path",
+        "get_comfyui_path",
+        "get_node_class_mappings",
+        "get_value_at_index",
+        "import_custom_nodes",
+    }
+)
+
+
 def _strip_imports(source: str) -> str:
     """Remove all top-level import statements from Python source code.
 
@@ -99,6 +126,22 @@ def get_embedded_helpers() -> str:
     return "\n".join(parts)
 
 
+def verify_embedded_surface_matches_manifest() -> list[str]:
+    """Return embedded helper names that differ from the approved surface."""
+    actual = list_embedded_names()
+    differences = [
+        *(
+            f"missing approved embedded name: {name}"
+            for name in sorted(APPROVED_EMBEDDED_NAMES - actual)
+        ),
+        *(
+            f"unexpected embedded name: {name}"
+            for name in sorted(actual - APPROVED_EMBEDDED_NAMES)
+        ),
+    ]
+    return differences
+
+
 def list_embedded_names() -> set[str]:
     """Return the set of all top-level names that will be embedded.
 
@@ -194,7 +237,8 @@ def verify_no_missing_cross_calls() -> list[str]:
                     is_nested = _is_nested_or_local_def(filepath, caller_name)
                     if not is_nested:
                         unresolved.append(
-                            f"{rel_path}: calls '{caller_name}' (not embedded, not builtin)"
+                            f"{rel_path}: calls '{caller_name}' "
+                            "(not embedded, not builtin)"
                         )
 
     return unresolved

comfyui_to_python/generator/planner.py

@@ -51,12 +51,7 @@ def build_plan(
             input_value_types = self.get_input_value_types(input_types)
             class_def = self.node_class_mappings[class_type]()
 
-            missing_required_variable = False
-            if "required" in input_types.keys():
-                for required in input_types["required"]:
-                    if required not in inputs.keys():
-                        missing_required_variable = True
-            if missing_required_variable:
+            if self._node_has_missing_required_inputs(input_types, inputs):
                 continue
 
             if class_type not in initialized_objects:
@@ -85,22 +80,7 @@ def build_plan(
                 if no_params or key in class_def_params
             }
 
-            hidden_inputs = input_types.get("hidden", {})
-            if "unique_id" in hidden_inputs and (
-                no_params or "unique_id" in class_def_params
-            ):
-                inputs["unique_id"] = random.randint(1, 2**64)
-            if "prompt" in hidden_inputs and (
-                no_params or "prompt" in class_def_params
-            ):
-                inputs["prompt"] = {"variable_name": "prompt"}
-            if "extra_pnginfo" in hidden_inputs and (
-                no_params or "extra_pnginfo" in class_def_params
-            ):
-                inputs["extra_pnginfo"] = {"variable_name": "extra_pnginfo"}
-            if "hidden" not in input_types and class_def_params is not None:
-                if "unique_id" in class_def_params:
-                    inputs["unique_id"] = random.randint(1, 2**64)
+            self._apply_hidden_inputs(inputs, input_types, class_def_params)
 
             executed_variables[idx] = (
                 f"{self.clean_variable_name(class_type)}_"
@@ -135,6 +115,34 @@ def build_plan(
             custom_nodes=custom_nodes,
         )
 
+    @staticmethod
+    def _node_has_missing_required_inputs(input_types: dict, inputs: dict) -> bool:
+        """Return True when a node is missing any required input."""
+        return any(
+            required not in inputs for required in input_types.get("required", {})
+        )
+
+    @staticmethod
+    def _apply_hidden_inputs(
+        inputs: dict, input_types: dict, class_def_params: list | None
+    ) -> None:
+        """Inject ComfyUI hidden runtime inputs accepted by the node function."""
+        hidden_inputs = input_types.get("hidden", {})
+        no_params = class_def_params is None
+        if "unique_id" in hidden_inputs and (
+            no_params or "unique_id" in class_def_params
+        ):
+            inputs["unique_id"] = random.randint(1, 2**64)
+        if "prompt" in hidden_inputs and (no_params or "prompt" in class_def_params):
+            inputs["prompt"] = {"variable_name": "prompt"}
+        if "extra_pnginfo" in hidden_inputs and (
+            no_params or "extra_pnginfo" in class_def_params
+        ):
+            inputs["extra_pnginfo"] = {"variable_name": "extra_pnginfo"}
+        if "hidden" not in input_types and class_def_params is not None:
+            if "unique_id" in class_def_params:
+                inputs["unique_id"] = random.randint(1, 2**64)
+
     def create_function_call_code(
         self,
         obj_name: str,
@@ -164,12 +172,16 @@ def create_prompt_seed_sync_code(
         for key in ("seed", "noise_seed"):
             if key not in inputs:
                 continue
-            randomized_seed_variable = f"node_{self.sanitize_node_id(str(node_id))}_{self.clean_variable_name(key)}"
+            randomized_seed_variable = (
+                f"node_{self.sanitize_node_id(str(node_id))}_"
+                f"{self.clean_variable_name(key)}"
+            )
             randomized_seed_code = self.get_randomized_seed_code(
                 input_value_types.get(key)
             )
             seed_sync_lines.append(
-                f'{randomized_seed_variable} = prompt["{node_id}"]["inputs"]["{key}"] = {randomized_seed_code}'
+                f'{randomized_seed_variable} = prompt["{node_id}"]["inputs"]'
+                f'["{key}"] = {randomized_seed_code}'
             )
             inputs[key] = {"variable_name": randomized_seed_variable}
 
@@ -256,7 +268,9 @@ def update_inputs(self, inputs: dict, executed_variables: dict) -> dict:
                 isinstance(inputs[key], list)
                 and inputs[key][0] in executed_variables.keys()
             ):
-                inputs[key] = {
-                    "variable_name": f"get_value_at_index({executed_variables[inputs[key][0]]}, {inputs[key][1]})"
-                }
+                variable_name = (
+                    f"get_value_at_index({executed_variables[inputs[key][0]]}, "
+                    f"{inputs[key][1]})"
+                )
+                inputs[key] = {"variable_name": variable_name}
         return inputs

comfyui_to_python/generator/render.py

@@ -16,19 +16,23 @@ class WorkflowRenderer:
     """Render a generation plan into the final standalone Python source."""
 
     def render(self, plan: GenerationPlan) -> str:
-        workflow_literal = self.format_python_literal(plan.workflow_data)
-        if plan.metadata_workflow_data is None:
-            extra_pnginfo_literal = "None"
-        else:
-            extra_pnginfo_literal = self.format_python_literal(
-                {"workflow": plan.metadata_workflow_data}
-            )
+        final_code = "\n".join(
+            self._build_static_imports(plan)
+            + [""]
+            + self._build_workflow_section(plan)
+            + [""]
+            + self._build_execution_section(plan)
+            + [""]
+            + self._build_entrypoint_section()
+        )
+        return black.format_str(final_code, mode=black.Mode())
 
+    def _build_static_imports(self, plan: GenerationPlan) -> list[str]:
+        """Build imports and embedded helper definitions for standalone scripts."""
         # Auto-discover all helpers from contributing runtime modules.
         # Reads source files, strips imports, embeds definitions — so internal
         # cross-calls always resolve (no NameError from missing __all__ entries).
         embedded_helpers = get_embedded_helpers()
-
         static_imports = [
             "# Imports",
             "import gc",
@@ -42,20 +46,22 @@ def render(self, plan: GenerationPlan) -> str:
             "from typing import Sequence, Mapping, Any, Union",
             "",
             "log = logging.getLogger(__name__)",
-        ] + [embedded_helpers]
-
+            embedded_helpers,
+        ]
         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
+        return static_imports
 
-        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 = [
+    def _build_workflow_section(self, plan: GenerationPlan) -> list[str]:
+        """Build workflow and PNG metadata literals."""
+        workflow_literal = self.format_python_literal(plan.workflow_data)
+        if plan.metadata_workflow_data is None:
+            extra_pnginfo_literal = "None"
+        else:
+            extra_pnginfo_literal = self.format_python_literal(
+                {"workflow": plan.metadata_workflow_data}
+            )
+        return [
             "# Workflow data",
             "def build_workflow() -> dict[str, Any]:",
             f"    return {workflow_literal}",
@@ -68,17 +74,22 @@ def render(self, plan: GenerationPlan) -> str:
             "extra_pnginfo = build_extra_pnginfo()",
         ]
 
+    def _build_execution_section(self, plan: GenerationPlan) -> list[str]:
+        """Build the bootstrap, import, loop, and cleanup body."""
         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 plan.custom_nodes:
+            execution_section.append("    import_custom_nodes()")
+
+        imports_code = self._build_imports_code(plan)
         if imports_code:
             execution_section.extend(["", "    # Node imports"])
             execution_section.extend(f"    {line}" for line in imports_code)
+
         execution_section.extend(
             [
                 "",
@@ -105,24 +116,26 @@ def render(self, plan: GenerationPlan) -> str:
                 "        cleanup_comfyui_runtime(unload_models=unload_models)",
             ]
         )
+        return execution_section
 
-        entrypoint_section = [
+    @staticmethod
+    def _build_imports_code(plan: GenerationPlan) -> list[str]:
+        """Build sorted node import statements for the main function."""
+        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}")
+        return imports_code
+
+    @staticmethod
+    def _build_entrypoint_section() -> list[str]:
+        """Build the standalone script entrypoint."""
+        return [
             "# Entrypoint",
             'if __name__ == "__main__":',
             "    main()",
         ]
 
-        final_code = "\n".join(
-            static_imports
-            + [""]
-            + workflow_section
-            + [""]
-            + execution_section
-            + [""]
-            + entrypoint_section
-        )
-        return black.format_str(final_code, mode=black.Mode())
-
     @staticmethod
     def format_python_literal(value: Any) -> str:
         return pformat(value, sort_dicts=False)

comfyui_to_python/node_runtime.py

@@ -165,17 +165,19 @@ def bootstrap_comfyui_runtime() -> None:
     # argparse crashes when bootstrap runs inside a subprocess (e.g. test
     # runner) where sys.argv contains non-ComfyUI arguments.
     original_argv = sys.argv
-    sys.argv = _filter_comfyui_args(sys.argv)
+    try:
+        sys.argv = _filter_comfyui_args(sys.argv)
 
-    # Load via _bootstrap_import() for namespace-package-safe imports.
-    options_mod = _bootstrap_import("comfy.options")
-    if options_mod is not None:
-        options_mod.enable_args_parsing()
+        # Load via _bootstrap_import() for namespace-package-safe imports.
+        options_mod = _bootstrap_import("comfy.options")
+        if options_mod is not None:
+            options_mod.enable_args_parsing()
 
-    cli_args_mod = _bootstrap_import("comfy.cli_args")
+        cli_args_mod = _bootstrap_import("comfy.cli_args")
+    finally:
+        # Restore original argv so downstream and error paths see real inputs.
+        sys.argv = original_argv
 
-    # Restore original argv so that downstream code sees what was actually passed
-    sys.argv = original_argv
     args = getattr(cli_args_mod, "args", None) if cli_args_mod else None
 
     if os.name == "nt":

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

@@ -602,17 +602,19 @@ def bootstrap_comfyui_runtime() -> None:
     # argparse crashes when bootstrap runs inside a subprocess (e.g. test
     # runner) where sys.argv contains non-ComfyUI arguments.
     original_argv = sys.argv
-    sys.argv = _filter_comfyui_args(sys.argv)
+    try:
+        sys.argv = _filter_comfyui_args(sys.argv)
 
-    # Load via _bootstrap_import() for namespace-package-safe imports.
-    options_mod = _bootstrap_import("comfy.options")
-    if options_mod is not None:
-        options_mod.enable_args_parsing()
+        # Load via _bootstrap_import() for namespace-package-safe imports.
+        options_mod = _bootstrap_import("comfy.options")
+        if options_mod is not None:
+            options_mod.enable_args_parsing()
 
-    cli_args_mod = _bootstrap_import("comfy.cli_args")
+        cli_args_mod = _bootstrap_import("comfy.cli_args")
+    finally:
+        # Restore original argv so downstream and error paths see real inputs.
+        sys.argv = original_argv
 
-    # Restore original argv so that downstream code sees what was actually passed
-    sys.argv = original_argv
     args = getattr(cli_args_mod, "args", None) if cli_args_mod else None
 
     if os.name == "nt":