From 1883d2b5885e203c330a28fc2884f77ee31dbd38 Mon Sep 17 00:00:00 2001
From: Mark Vrijlandt <mark.vrijlandt@tno.nl>
Date: Tue, 12 May 2026 18:30:28 +0200
Subject: [PATCH] fix_orchestrator getting stuck not processing result

---
 .idea/runConfigurations/main.xml | 26 +++++++++++++++++++
 .vscode/launch.json              | 15 +++++++++++
 pyproject.toml                   | 10 ++++----
 src/omotes_orchestrator/main.py  | 44 ++++++++++++++++++++++----------
 4 files changed, 76 insertions(+), 19 deletions(-)
 create mode 100644 .idea/runConfigurations/main.xml
 create mode 100644 .vscode/launch.json

diff --git a/.idea/runConfigurations/main.xml b/.idea/runConfigurations/main.xml
new file mode 100644
index 0000000..92bd722
--- /dev/null
+++ b/.idea/runConfigurations/main.xml
@@ -0,0 +1,26 @@
+<component name="ProjectRunConfigurationManager">
+  <configuration default="false" name="main" type="PythonConfigurationType" factoryName="Python" nameIsGenerated="true">
+    <module name="orchestrator" />
+    <option name="ENV_FILES" value="" />
+    <option name="INTERPRETER_OPTIONS" value="" />
+    <option name="PARENT_ENVS" value="true" />
+    <envs>
+      <env name="PYTHONUNBUFFERED" value="1" />
+    </envs>
+    <option name="SDK_HOME" value="" />
+    <option name="WORKING_DIRECTORY" value="$PROJECT_DIR$/src" />
+    <option name="IS_MODULE_SDK" value="true" />
+    <option name="ADD_CONTENT_ROOTS" value="true" />
+    <option name="ADD_SOURCE_ROOTS" value="true" />
+    <EXTENSION ID="PythonCoverageRunConfigurationExtension" runner="coverage.py" />
+    <option name="RUN_TOOL" value="" />
+    <option name="SCRIPT_NAME" value="$PROJECT_DIR$/src/omotes_orchestrator/main.py" />
+    <option name="PARAMETERS" value="" />
+    <option name="SHOW_COMMAND_LINE" value="false" />
+    <option name="EMULATE_TERMINAL" value="false" />
+    <option name="MODULE_MODE" value="false" />
+    <option name="REDIRECT_INPUT" value="false" />
+    <option name="INPUT_FILE" value="" />
+    <method v="2" />
+  </configuration>
+</component>
\ No newline at end of file
diff --git a/.vscode/launch.json b/.vscode/launch.json
new file mode 100644
index 0000000..033c6ba
--- /dev/null
+++ b/.vscode/launch.json
@@ -0,0 +1,15 @@
+{
+  "version": "0.2.0",
+  "configurations": [
+    {
+      "name": "Python: Orchestrator",
+      "type": "debugpy",
+      "request": "launch",
+      "module": "omotes_orchestrator.main",
+      "cwd": "${workspaceFolder}/src",
+      "envFile": "${workspaceFolder}/.env",
+      "console": "integratedTerminal",
+      "justMyCode": true
+    }
+  ]
+}
diff --git a/pyproject.toml b/pyproject.toml
index e153e96..a3fba05 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -20,7 +20,7 @@ classifiers = [
     "Topic :: Scientific/Engineering :: Physics",
 ]
 readme = "README.md"
-license = {file = "LICENSE"}
+license = { file = "LICENSE" }
 requires-python = ">=3.10"
 
 dependencies = [
@@ -53,7 +53,7 @@ dev = [
     "isort == 5.13.2",
     "build ~= 1.2.2",
     "sqlalchemy[mypy]",
-    "types-protobuf ~= 4.24.0"
+    "types-protobuf ~= 4.24.0",
 ]
 
 [project.urls]
@@ -93,11 +93,11 @@ ignore = [
     'C408', # Suggestion to use dict() over {}
     'W503', # Starting lines with operators.
     'D104', # Missing docstring in public package
-    'D100'  # Missing docstring in public module
+    'D100', # Missing docstring in public module
 ]
 per-file-ignores = [
     '__init__.py:F401',
-    './unit_test/*:D100,D101,D102,D103,D106,D107'
+    './unit_test/*:D100,D101,D102,D103,D106,D107',
 ]
 max-line-length = 100
 count = true
@@ -135,4 +135,4 @@ ignore_missing_imports = true
 
 [[tool.mypy.overrides]]
 module = "influxdb.*"
-ignore_missing_imports = true
\ No newline at end of file
+ignore_missing_imports = true
diff --git a/src/omotes_orchestrator/main.py b/src/omotes_orchestrator/main.py
index 9e74d2a..3470223 100644
--- a/src/omotes_orchestrator/main.py
+++ b/src/omotes_orchestrator/main.py
@@ -224,14 +224,7 @@ def start(self) -> None:
 
         self.celery_if.start()
 
-        self.worker_if.start()
-        self.worker_if.connect_to_worker_task_results(
-            callback_on_worker_task_result=self.task_result_received
-        )
-        self.worker_if.connect_to_worker_task_progress_updates(
-            callback_on_worker_task_progress_update=self.task_progress_update
-        )
-
+        # Initialize SDK-side broker and exchange before worker callbacks can emit results.
         self.omotes_sdk_if.start()
         self.omotes_sdk_if.connect_to_job_submissions(
             callback_on_new_job=self.new_job_submitted_handler
@@ -245,6 +238,14 @@ def start(self) -> None:
         )
         self.omotes_sdk_if.send_available_workflows()
 
+        self.worker_if.start()
+        self.worker_if.connect_to_worker_task_results(
+            callback_on_worker_task_result=self.task_result_received
+        )
+        self.worker_if.connect_to_worker_task_progress_updates(
+            callback_on_worker_task_progress_update=self.task_progress_update
+        )
+
         self.postgres_job_manager.start()
         self.timeout_job_manager.start()
         self.esdl_time_series_manager.start()
@@ -459,12 +460,27 @@ def _guard_time_series_data_from_cleanup(
         :param job_id: ID of the job that created this ESDL.
         :param job_reference: Reference of the job that created this ESDL.
         """
-        output_esdl_id = pyesdl_from_string(esdl).energy_system.id
-        self.postgresql_if.put_new_esdl_time_series_info(
-            output_esdl_id=output_esdl_id,
-            job_id=job_id,
-            job_reference=job_reference,
-        )
+        try:
+            output_esdl_id = pyesdl_from_string(esdl).energy_system.id
+            if not isinstance(output_esdl_id, str) or not output_esdl_id:
+                logger.warning(
+                    "Could not guard time series data for job %s: missing or invalid "
+                    "energy system id in output ESDL.",
+                    job_id,
+                )
+                return
+
+            self.postgresql_if.put_new_esdl_time_series_info(
+                output_esdl_id=output_esdl_id,
+                job_id=job_id,
+                job_reference=job_reference,
+            )
+        except Exception:
+            logger.exception(
+                "Could not parse output ESDL to guard time series data for job %s. "
+                "Continuing without guard.",
+                job_id,
+            )
 
     def task_result_received(self, task_result: TaskResult) -> None:
         """When a task result is received from a worker through RabbitMQ, Celery side.