feat(app): move output annotation checking to run_app

Also change import order to ensure CLI args are handled correctly. Had to do this bc importing `InvocationRegistry` before parsing args resulted in the `--root` CLI arg being ignored.

Author: psychedelicious

invokeai/app/invocations/baseinvocation.py

@@ -643,6 +643,16 @@ def wrapper(cls: Type[TBaseInvocation]) -> Type[TBaseInvocation]:
 
         fields["type"] = (invocation_type_annotation, invocation_type_field_info)
 
+        # Invocation outputs must be registered using the @invocation_output decorator, but it is possible that the
+        # output is registered _after_ this invocation is registered. It depends on module import ordering.
+        #
+        # We can only confirm the output for an invocation is registered after all modules are imported. There's
+        # only really one good time to do that - during application startup, in `run_app.py`, after loading all
+        # custom nodes.
+        #
+        # We can still do some basic validation here - ensure the invoke method is defined and returns an instance
+        # of BaseInvocationOutput.
+
         # Validate the `invoke()` method is implemented
         if "invoke" in cls.__abstractmethods__:
             raise ValueError(f'Invocation "{invocation_type}" must implement the "invoke" method')

invokeai/app/run_app.py

@@ -1,12 +1,3 @@
-import uvicorn
-
-from invokeai.app.invocations.load_custom_nodes import load_custom_nodes
-from invokeai.app.services.config.config_default import get_config
-from invokeai.app.util.torch_cuda_allocator import configure_torch_cuda_allocator
-from invokeai.backend.util.logging import InvokeAILogger
-from invokeai.frontend.cli.arg_parser import InvokeAIArgs
-
-
 def get_app():
     """Import the app and event loop. We wrap this in a function to more explicitly control when it happens, because
     importing from api_app does a bunch of stuff - it's more like calling a function than importing a module.
@@ -18,9 +9,20 @@ def get_app():
 
 def run_app() -> None:
     """The main entrypoint for the app."""
-    # Parse the CLI arguments.
+    from invokeai.frontend.cli.arg_parser import InvokeAIArgs
+
+    # Parse the CLI arguments before doing anything else, which ensures CLI args correctly override settings from other
+    # sources like `invokeai.yaml` or env vars.
     InvokeAIArgs.parse_args()
 
+    import uvicorn
+
+    from invokeai.app.invocations.baseinvocation import InvocationRegistry
+    from invokeai.app.invocations.load_custom_nodes import load_custom_nodes
+    from invokeai.app.services.config.config_default import get_config
+    from invokeai.app.util.torch_cuda_allocator import configure_torch_cuda_allocator
+    from invokeai.backend.util.logging import InvokeAILogger
+
     # Load config.
     app_config = get_config()
 
@@ -66,6 +68,15 @@ def run_app() -> None:
     # core nodes have been imported so that we can catch when a custom node clobbers a core node.
     load_custom_nodes(custom_nodes_path=app_config.custom_nodes_path, logger=logger)
 
+    # Check all invocations and ensure their outputs are registered.
+    for invocation in InvocationRegistry.get_invocation_classes():
+        invocation_type = invocation.get_type()
+        output_annotation = invocation.get_output_annotation()
+        if output_annotation not in InvocationRegistry.get_output_classes():
+            logger.warning(
+                f'Invocation "{invocation_type}" has unregistered output class "{output_annotation.__name__}"'
+            )
+
     if app_config.dev_reload:
         # load_custom_nodes seems to bypass jurrigged's import sniffer, so be sure to call it *after* they're already
         # imported.

invokeai/app/util/custom_openapi.py

@@ -88,12 +88,6 @@ def openapi() -> dict[str, Any]:
             invocation_output_map_properties[invocation_type] = json_schema["output"]
             invocation_output_map_required.append(invocation_type)
 
-            output_annotation = invocation.get_output_annotation()
-            if output_annotation not in InvocationRegistry.get_output_classes():
-                logger.warning(
-                    f'Invocation "{invocation_type}"\' has unregistered output class {output_annotation.__name__} (did you forget @invocation_output?)'
-                )
-
         # Add the output map to the schema
         openapi_schema["components"]["schemas"]["InvocationOutputMap"] = {
             "type": "object",