feat(app): iterate on processor split

- Add `OnNodeError` and `OnNonFatalProcessorError` callbacks
- Move all session/node callbacks to `SessionRunner` - this ensures we dump perf stats before resetting them and generally makes sense to me
- Remove `complete` event from `SessionRunner`, it's essentially the same as `OnAfterRunSession`
- Remove extraneous `next_invocation` block, which would treat a processor error as a node error
- Simplify loops
- Add some callbacks for testing, to be removed before merge

Author: psychedelicious

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,41 @@ def initialize(config: InvokeAIAppConfig, event_handler_id: int, logger: Logger
         )
         names = SimpleNameService()
         performance_statistics = InvocationStatsService()
-        session_processor = DefaultSessionProcessor()
+
+        def on_before_run_session(queue_item):
+            print("BEFORE RUN SESSION", queue_item.item_id)
+            return True
+
+        def on_before_run_node(invocation, queue_item):
+            print("BEFORE RUN NODE", invocation.id)
+            return True
+
+        def on_after_run_node(invocation, queue_item, outputs):
+            print("AFTER RUN NODE", invocation.id)
+            return True
+
+        def on_node_error(invocation, queue_item, exc_type, exc_value, exc_traceback):
+            print("NODE ERROR", invocation.id)
+            return True
+
+        def on_after_run_session(queue_item):
+            print("AFTER RUN SESSION", queue_item.item_id)
+            return True
+
+        def on_non_fatal_processor_error(queue_item, exc_type, exc_value, exc_traceback):
+            print("NON FATAL PROCESSOR ERROR", exc_value)
+            return True
+
+        session_processor = DefaultSessionProcessor(
+            DefaultSessionRunner(
+                on_before_run_session,
+                on_before_run_node,
+                on_after_run_node,
+                on_node_error,
+                on_after_run_session,
+            ),
+            on_non_fatal_processor_error,
+        )
         session_queue = SqliteSessionQueue(db=db)
         urls = LocalUrlService()
         workflow_records = SqliteWorkflowRecordsStorage(db=db)

invokeai/app/services/session_processor/session_processor_base.py

@@ -1,6 +1,7 @@
 from abc import ABC, abstractmethod
 from threading import Event
 
+from invokeai.app.invocations.baseinvocation import BaseInvocation
 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
@@ -22,12 +23,7 @@ def run(self, queue_item: SessionQueueItem) -> None:
         pass
 
     @abstractmethod
-    def complete(self, queue_item: SessionQueueItem) -> None:
-        """Completes the session"""
-        pass
-
-    @abstractmethod
-    def run_node(self, node_id: str, queue_item: SessionQueueItem) -> None:
+    def run_node(self, invocation: BaseInvocation, queue_item: SessionQueueItem) -> None:
         """Runs an already prepared node on the session"""
         pass