Break apart session processor and the running of each session into se…

invokeai/app/api/dependencies.py

@@ -29,7 +29,7 @@
 from ..services.model_manager.model_manager_default import ModelManagerService
 from ..services.model_records import ModelRecordServiceSQL
 from ..services.names.names_default import SimpleNameService
-from ..services.session_processor.session_processor_default import DefaultSessionProcessor
+from ..services.session_processor.session_processor_default import DefaultSessionProcessor, DefaultSessionRunner
 from ..services.session_queue.session_queue_sqlite import SqliteSessionQueue
 from ..services.urls.urls_default import LocalUrlService
 from ..services.workflow_records.workflow_records_sqlite import SqliteWorkflowRecordsStorage
@@ -103,7 +103,7 @@ def initialize(config: InvokeAIAppConfig, event_handler_id: int, logger: Logger
         )
         names = SimpleNameService()
         performance_statistics = InvocationStatsService()
-        session_processor = DefaultSessionProcessor()
+        session_processor = DefaultSessionProcessor(session_runner=DefaultSessionRunner())
         session_queue = SqliteSessionQueue(db=db)
         urls = LocalUrlService()
         workflow_records = SqliteWorkflowRecordsStorage(db=db)

invokeai/app/services/events/events_base.py

@@ -121,7 +121,8 @@ def emit_invocation_error(
         node: dict,
         source_node_id: str,
         error_type: str,
-        error: str,
+        error_message: str,
+        error_traceback: str,
         user_id: str | None,
         project_id: str | None,
     ) -> None:
@@ -136,7 +137,8 @@ def emit_invocation_error(
                 "node": node,
                 "source_node_id": source_node_id,
                 "error_type": error_type,
-                "error": error,
+                "error_message": error_message,
+                "error_traceback": error_traceback,
                 "user_id": user_id,
                 "project_id": project_id,
             },
@@ -257,7 +259,9 @@ def emit_queue_item_status_changed(
                     "status": session_queue_item.status,
                     "batch_id": session_queue_item.batch_id,
                     "session_id": session_queue_item.session_id,
-                    "error": session_queue_item.error,
+                    "error_type": session_queue_item.error_type,
+                    "error_message": session_queue_item.error_message,
+                    "error_traceback": session_queue_item.error_traceback,
                     "created_at": str(session_queue_item.created_at) if session_queue_item.created_at else None,
                     "updated_at": str(session_queue_item.updated_at) if session_queue_item.updated_at else None,
                     "started_at": str(session_queue_item.started_at) if session_queue_item.started_at else None,

invokeai/app/services/session_processor/session_processor_base.py

@@ -1,6 +1,49 @@
 from abc import ABC, abstractmethod
+from threading import Event
+from typing import Optional, Protocol
 
+from invokeai.app.invocations.baseinvocation import BaseInvocation, BaseInvocationOutput
+from invokeai.app.services.invocation_services import InvocationServices
 from invokeai.app.services.session_processor.session_processor_common import SessionProcessorStatus
+from invokeai.app.services.session_queue.session_queue_common import SessionQueueItem
+from invokeai.app.util.profiler import Profiler
+
+
+class SessionRunnerBase(ABC):
+    """
+    Base class for session runner.
+    """
+
+    @abstractmethod
+    def start(self, services: InvocationServices, cancel_event: Event, profiler: Optional[Profiler] = None) -> None:
+        """Starts the session runner.
+
+        Args:
+            services: The invocation services.
+            cancel_event: The cancel event.
+            profiler: The profiler to use for session profiling via cProfile. Omit to disable profiling. Basic session
+                stats will be still be recorded and logged when profiling is disabled.
+        """
+        pass
+
+    @abstractmethod
+    def run(self, queue_item: SessionQueueItem) -> None:
+        """Runs a session.
+
+        Args:
+            queue_item: The session to run.
+        """
+        pass
+
+    @abstractmethod
+    def run_node(self, invocation: BaseInvocation, queue_item: SessionQueueItem) -> None:
+        """Run a single node in the graph.
+
+        Args:
+            invocation: The invocation to run.
+            queue_item: The session queue item.
+        """
+        pass
 
 
 class SessionProcessorBase(ABC):
@@ -26,3 +69,85 @@ def pause(self) -> SessionProcessorStatus:
     def get_status(self) -> SessionProcessorStatus:
         """Gets the status of the session processor"""
         pass
+
+
+class OnBeforeRunNode(Protocol):
+    def __call__(self, invocation: BaseInvocation, queue_item: SessionQueueItem) -> None:
+        """Callback to run before executing a node.
+
+        Args:
+            invocation: The invocation that will be executed.
+            queue_item: The session queue item.
+        """
+        ...
+
+
+class OnAfterRunNode(Protocol):
+    def __call__(self, invocation: BaseInvocation, queue_item: SessionQueueItem, output: BaseInvocationOutput) -> None:
+        """Callback to run before executing a node.
+
+        Args:
+            invocation: The invocation that was executed.
+            queue_item: The session queue item.
+        """
+        ...
+
+
+class OnNodeError(Protocol):
+    def __call__(
+        self,
+        invocation: BaseInvocation,
+        queue_item: SessionQueueItem,
+        error_type: str,
+        error_message: str,
+        error_traceback: str,
+    ) -> None:
+        """Callback to run when a node has an error.
+
+        Args:
+            invocation: The invocation that errored.
+            queue_item: The session queue item.
+            error_type: The type of error, e.g. "ValueError".
+            error_message: The error message, e.g. "Invalid value".
+            error_traceback: The stringified error traceback.
+        """
+        ...
+
+
+class OnBeforeRunSession(Protocol):
+    def __call__(self, queue_item: SessionQueueItem) -> None:
+        """Callback to run before executing a session.
+
+        Args:
+            queue_item: The session queue item.
+        """
+        ...
+
+
+class OnAfterRunSession(Protocol):
+    def __call__(self, queue_item: SessionQueueItem) -> None:
+        """Callback to run after executing a session.
+
+        Args:
+            queue_item: The session queue item.
+        """
+        ...
+
+
+class OnNonFatalProcessorError(Protocol):
+    def __call__(
+        self,
+        queue_item: Optional[SessionQueueItem],
+        error_type: str,
+        error_message: str,
+        error_traceback: str,
+    ) -> None:
+        """Callback to run when a non-fatal error occurs in the processor.
+
+        Args:
+            queue_item: The session queue item, if one was being executed when the error occurred.
+            error_type: The type of error, e.g. "ValueError".
+            error_message: The error message, e.g. "Invalid value".
+            error_traceback: The stringified error traceback.
+        """
+        ...