feat(app): handle preparation errors as node errors

We were not handling node preparation errors as node errors before. Here's the explanation, copied from a comment that is no longer required:

---

TODO(psyche): Sessions only support errors on nodes, not on the session itself. When an error occurs outside
node execution, it bubbles up to the processor where it is treated as a queue item error.

Nodes are pydantic models. When we prepare a node in `session.next()`, we set its inputs. This can cause a
pydantic validation error. For example, consider a resize image node which has a constraint on its `width`
input field - it must be greater than zero. During preparation, if the width is set to zero, pydantic will
raise a validation error.

When this happens, it breaks the flow before `invocation` is set. We can't set an error on the invocation
because we didn't get far enough to get it - we don't know its id. Hence, we just set it as a queue item error.

---

This change wraps the node preparation step with exception handling. A new `NodeInputError` exception is raised when there is a validation error. This error has the node (in the state it was in just prior to the error) and an identifier of the input that failed.

This allows us to mark the node that failed preparation as errored, correctly making such errors _node_ errors and not _processor_ errors. It's much easier to diagnose these situations. The error messages look like this:

> Node b5ac87c6-0678-4b8c-96b9-d215aee12175 has invalid incoming input for height

Some of the exception handling logic is cleaned up.

invokeai/app/services/session_processor/session_processor_default.py

@@ -21,6 +21,7 @@ from invokeai.app.services.session_processor.session_processor_base import (
 )
 from invokeai.app.services.session_processor.session_processor_common import CanceledException
 from invokeai.app.services.session_queue.session_queue_common import SessionQueueItem
+from invokeai.app.services.shared.graph import NodeInputError
 from invokeai.app.services.shared.invocation_context import InvocationContextData, build_invocation_context
 from invokeai.app.util.profiler import Profiler
 
@@ -66,18 +67,16 @@ class DefaultSessionRunner(SessionRunnerBase):
 
         # Loop over invocations until the session is complete or canceled
         while True:
-            # TODO(psyche): Sessions only support errors on nodes, not on the session itself. When an error occurs outside
-            # node execution, it bubbles up to the processor where it is treated as a queue item error.
-            #
-            # Nodes are pydantic models. When we prepare a node in `session.next()`, we set its inputs. This can cause a
-            # pydantic validation error. For example, consider a resize image node which has a constraint on its `width`
-            # input field - it must be greater than zero. During preparation, if the width is set to zero, pydantic will
-            # raise a validation error.
-            #
-            # When this happens, it breaks the flow before `invocation` is set. We can't set an error on the invocation
-            # because we didn't get far enough to get it - we don't know its id. Hence, we just set it as a queue item error.
-
+            try:
                 invocation = queue_item.session.next()
+            # Anything other than a `NodeInputError` is handled as a processor error
+            except NodeInputError as e:
+                # Must extract the exception traceback here to not lose its stacktrace when we change scope
+                traceback = e.__traceback__
+                assert traceback is not None
+                self._on_node_error(e.node, queue_item, type(e), e, traceback)
+                break
+
             if invocation is None or self._cancel_event.is_set():
                 break
             self.run_node(invocation, queue_item)

invokeai/app/services/shared/graph.py

@@ -8,6 +8,7 @@ import networkx as nx
 from pydantic import (
     BaseModel,
     GetJsonSchemaHandler,
+    ValidationError,
     field_validator,
 )
 from pydantic.fields import Field
@@ -190,6 +191,39 @@ class UnknownGraphValidationError(ValueError):
     pass
 
 
+class NodeInputError(ValueError):
+    """Raised when a node fails preparation. This occurs when a node's inputs are being set from its incomers, but an
+    input fails validation.
+
+    Attributes:
+        node: The node that failed preparation. Note: only successfully set fields will be accurate. Review the error to
+            determine which field caused the failure.
+    """
+
+    def __init__(self, node: BaseInvocation, e: ValidationError):
+        self.original_error = e
+        self.node = node
+        # When preparing a node, we set each input one-at-a-time. We may thus safely assume that the first error
+        # represents the first input that failed.
+        self.failed_input = loc_to_dot_sep(e.errors()[0]["loc"])
+        super().__init__(f"Node {node.id} has invalid incoming input for {self.failed_input}")
+
+
+def loc_to_dot_sep(loc: tuple[Union[str, int], ...]) -> str:
+    """Helper to pretty-print pydantic error locations as dot-separated strings.
+    Taken from https://docs.pydantic.dev/latest/errors/errors/#customize-error-messages
+    """
+    path = ""
+    for i, x in enumerate(loc):
+        if isinstance(x, str):
+            if i > 0:
+                path += "."
+            path += x
+        else:
+            path += f"[{x}]"
+    return path
+
+
 @invocation_output("iterate_output")
 class IterateInvocationOutput(BaseInvocationOutput):
     """Used to connect iteration outputs. Will be expanded to a specific output."""
@@ -821,7 +855,10 @@ class GraphExecutionState(BaseModel):
 
         # Get values from edges
         if next_node is not None:
+            try:
                 self._prepare_inputs(next_node)
+            except ValidationError as e:
+                raise NodeInputError(next_node, e)
 
         # If next is still none, there's no next node, return None
         return next_node