refactor into task_executor api and runner facade

Author: lukasbindreiter

CHANGELOG.md

@@ -7,6 +7,11 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+### Added
+
+- `tilebox-workflows`: Added worker runtime mode for executing registered tasks over the worker RPC service, backed by a
+  shared task executor facade used by both worker and polling runners.
+
 ## [0.52.0]
 
 ### Added

tilebox-workflows/tests/runner/testdata/recordings/fibonacci_workflow.rpcs.bin

@@ -1,3 +1,3 @@
 version https://git-lfs.github.com/spec/v1
-oid sha256:24ff20b8b66a30cf11e5998477066b500830280160eaa5dd1e1c5895ea920a47
-size 9032
+oid sha256:ede65a3f196290e2aee71cddc45f67c9e9b2c6eb1374bf6beb0b8406f1f4d8c4
+size 8923

tilebox-workflows/tests/runner/testdata/recordings/flaky_task.rpcs.bin

@@ -1,3 +1,3 @@
 version https://git-lfs.github.com/spec/v1
-oid sha256:b8081ec6c43e20ee4ef13ff7ae89a24f695cd85e10c0346341d77a9fec5a285f
-size 6061
+oid sha256:54b3862b1e4c863f27e9f2e29958e3122ea8d197459fdb87e7b3d81ff0373f59
+size 5932

tilebox-workflows/tests/runner/testdata/recordings/optional_subbranch.rpcs.bin

@@ -1,3 +1,3 @@
 version https://git-lfs.github.com/spec/v1
-oid sha256:cac0cd1cd31ec5c62949704159a113d2f69a249c35997eac113a2ac695d196d5
-size 4430
+oid sha256:432d42a185a620843815a821a9c04617459b351dc873a2dc31101657639d13e0
+size 4325

tilebox-workflows/tests/runner/testdata/recordings/optional_subtask.rpcs.bin

@@ -1,3 +1,3 @@
 version https://git-lfs.github.com/spec/v1
-oid sha256:3eee76fab8b6018fbba6e0332a8a5a193a1095ceab3e4a0b2e2b586a5d520f9b
-size 3526
+oid sha256:1260ba9f7798ef3b53a76beac7943ee4857f7c5035c10425702ff121dc64966d
+size 3413

tilebox-workflows/tests/runner/testdata/recordings/progress.rpcs.bin

@@ -1,3 +1,3 @@
 version https://git-lfs.github.com/spec/v1
-oid sha256:432fe539191e3c122d88dfe9a9ad7ff4c797df7ab0390c1ab69f454ed8d80858
-size 4359
+oid sha256:e30083e0e53ed859997e01bf4abdd1c06d873ab3ac96f84497778e7d267d8481
+size 4258

tilebox-workflows/tilebox/workflows/client.py

@@ -1,7 +1,7 @@
 import logging
 import os
 import warnings
-from uuid import uuid4
+from uuid import UUID, uuid4
 
 from _tilebox.grpc.channel import open_channel, parse_channel_info
 from tilebox.workflows.automations.client import AutomationClient, AutomationService
@@ -21,6 +21,7 @@
     _create_tilebox_logger_provider,
 )
 from tilebox.workflows.observability.tracing import WorkflowTracer
+from tilebox.workflows.runner.executor import LazyStorageLocations
 from tilebox.workflows.runner.runner import Runner
 from tilebox.workflows.runner.task_runner import TaskRunner, _LeaseRenewer
 from tilebox.workflows.runner.task_service import TaskService
@@ -29,7 +30,12 @@
 
 class Client:
     def __init__(
-        self, *, url: str = "https://api.tilebox.com", token: str | None = None, name: str | None = None
+        self,
+        *,
+        url: str = "https://api.tilebox.com",
+        token: str | None = None,
+        name: str | None = None,
+        client_id: UUID | None = None,
     ) -> None:
         """
         Create a Tilebox workflows client.
@@ -40,13 +46,14 @@ def __init__(
             name: An optional name of the client, used as service.name for telemetry. If not set, defaults to
                 the service name provided by `tilebox.workflows.observability.tracing.configure_otel_tracing`,
                 or "tilebox-python" if no external tracer is configured.
+            client_id: An optional stable id used to scope internal loggers. Defaults to a random id.
         """
         token = _token_from_env(url, token)
         self._auth: dict[str, str] = {"token": token, "url": url}
         self._channel = open_channel(url, token)
 
         # configure logging and tracing
-        self._client_id = uuid4()  # a random uuid to scope loggers to this client instance
+        self._client_id = client_id or uuid4()  # a random uuid to scope loggers to this client instance
         self._logger_provider = _create_tilebox_logger_provider(service=name, url=url, token=token)
 
         # task logger is the logger available for users to emit logs from within a Task.execute method, via
@@ -134,18 +141,9 @@ def runner(
 
         found_cluster = self.clusters().find(to_cluster_slug(cluster or ""))
 
-        try:
-            storage_locations = self.automations().storage_locations()
-        except:  # noqa: E722
-            # if fetching storage locations fails, we just disable this feature, and don't crash all runners
-            # lets refactor this to a lazy loading mechanism in the future
-            storage_locations = []
-
         runner_context_type = runner_definition.context or RunnerContext
-        runner_context = runner_context_type(
-            self._tracer,
-            storage_locations=storage_locations,
-        )
+        runner_context = runner_context_type(self._tracer)
+        runner_context.storage_locations = LazyStorageLocations(self, runner_context)
 
         task_runner = TaskRunner(
             TaskService(self._channel),