Final set of fixes for forked/remote algo combo (#130)

esd_services_api_client/nexus/algorithms/__init__.py

@@ -21,3 +21,4 @@
 from esd_services_api_client.nexus.algorithms.recursive import *
 from esd_services_api_client.nexus.algorithms.distributed import *
 from esd_services_api_client.nexus.algorithms._baseline_algorithm import *
+from esd_services_api_client.nexus.algorithms._remote_algorithm import *

esd_services_api_client/nexus/algorithms/_remote_algorithm.py

@@ -23,6 +23,7 @@
 
 from adapta.metrics import MetricsProvider
 from adapta.utils.decorators import run_time_metrics_async
+from injector import inject
 
 from esd_services_api_client.crystal import CrystalConnector, AlgorithmConfiguration
 from esd_services_api_client.nexus.abstractions.algrorithm_cache import InputCache
@@ -43,6 +44,7 @@ class RemoteAlgorithm(NexusObject[TPayload, AlgorithmResult]):
     Base class for all algorithm implementations.
     """
 
+    @inject
     def __init__(
         self,
         metrics_provider: MetricsProvider,

esd_services_api_client/nexus/algorithms/forked_algorithm.py

@@ -143,13 +143,15 @@ async def _measured_run(**run_args) -> AlgorithmResult:
 
             return await self._main_run(**run_args)
 
-        if self._is_forked(**kwargs):
+        if await self._is_forked(**kwargs):
             self._inputs = await self._fork_inputs(**kwargs)
         else:
             self._inputs = await self._main_inputs(**kwargs)
 
         # evaluate if additional forks will be spawned
-        forks = await partial(self._get_forks, **self._inputs, **kwargs)()
+        forks: list[RemoteAlgorithm] = await partial(
+            self._get_forks, **self._inputs, **kwargs
+        )()
 
         if len(forks) > 0:
             self._logger.info(
@@ -168,6 +170,6 @@ async def _measured_run(**run_args) -> AlgorithmResult:
         )()
 
         # now await callback scheduling
-        await asyncio.wait([fork.run(**kwargs) for fork in forks])
+        await asyncio.wait([asyncio.create_task(fork.run(**kwargs)) for fork in forks])
 
         return run_result