Feature/edge

[andre-merzky]

No description provided.

[gemini-code-assist]

Code Review

This pull request introduces two new execution backends: EdgeExecutionBackend, which enables remote task execution via RADICAL Edge with built-in submission batching and SSE notification handling, and NoopExecutionBackend for performance benchmarking. Additionally, the PR updates task routing logic to use .get() for safer field access, adds traceback aggregation to the Dragon backend, and adjusts logging levels. The review feedback focuses on improving the responsiveness of the Edge backend by wrapping blocking network calls in asyncio.to_thread, ensuring robust task routing by using explicit 'is not None' checks, and maintaining interface consistency in the Noop backend by aligning method signatures with the base class.

[gemini-code-assist]

The BridgeClient initialization and subsequent calls to get_edge_client and get_plugin are likely blocking network operations. Performing these directly in an async method will block the event loop. These calls should be wrapped in asyncio.to_thread to maintain responsiveness.

Suggested change

	self._bc = BridgeClient(url=self._bridge_url)
	ec = self._bc.get_edge_client(self._edge_name)
	session_kwargs = {'backends': self._remote_backends}
	if self._notify_batch_window is not None:
	session_kwargs['notify_batch_window'] = self._notify_batch_window
	if self._notify_batch_size is not None:
	session_kwargs['notify_batch_size'] = self._notify_batch_size
	self._rh = ec.get_plugin(self._plugin_name, **session_kwargs)
	self._bc = await asyncio.to_thread(BridgeClient, url=self._bridge_url)
	ec = await asyncio.to_thread(self._bc.get_edge_client, self._edge_name)
	session_kwargs = {'backends': self._remote_backends}
	if self._notify_batch_window is not None:
	session_kwargs['notify_batch_window'] = self._notify_batch_window
	if self._notify_batch_size is not None:
	session_kwargs['notify_batch_size'] = self._notify_batch_size
	self._rh = await asyncio.to_thread(ec.get_plugin, self._plugin_name, **session_kwargs)

[gemini-code-assist]

The use of truthiness checks (if data.get("prompt"):) can lead to incorrect routing if the field contains a falsy value that is otherwise valid (e.g., an empty string). Since the goal is to avoid misrouting when keys are explicitly set to None, using an explicit is not None check is safer and more robust.

Suggested change

	if data.get("prompt"):
	return AITask(**data)
	elif "executable" in data or "function" in data:
	elif data.get("executable") or data.get("function"):
	return ComputeTask(**data)
	if data.get("prompt") is not None:
	return AITask(**data)
	elif data.get("executable") is not None or data.get("function") is not None:
	return ComputeTask(**data)

[gemini-code-assist]

The implementation of submit_tasks returns a list of tasks, which deviates from the BaseBackend interface (which specifies a None return type). Additionally, the submitted list is not used by the caller in Session. It's better to align with the base class signature.

Suggested change

	async def submit_tasks(self, tasks: list[dict[str, Any]]) -> list[asyncio.Task]:
	if self._backend_state != BackendMainStates.RUNNING:
	self._backend_state = BackendMainStates.RUNNING
	submitted = []
	for task in tasks:
	task.update({
	"return_value": True,
	"stdout": "",
	"stderr": "",
	"exit_code": 0,
	})
	self.tasks[task["uid"]] = task
	future = asyncio.create_task(self._complete(task))
	submitted.append(future)
	return submitted
	async def submit_tasks(self, tasks: list[dict[str, Any]]) -> None:
	if self._backend_state != BackendMainStates.RUNNING:
	self._backend_state = BackendMainStates.RUNNING
	for task in tasks:
	task.update({
	"return_value": True,
	"stdout": "",
	"stderr": "",
	"exit_code": 0,
	})
	self.tasks[task["uid"]] = task
	asyncio.create_task(self._complete(task))

[gemini-code-assist]

The signature of task_state_cb does not match the abstract method defined in BaseBackend. This will cause a TypeError if the method is called with the expected arguments (task and state).

Suggested change

	def task_state_cb(self):
	pass
	def task_state_cb(self, task: dict, state: str) -> None:
	pass

[AymenFJA]

I think we should remove prof

[AymenFJA]

Although this is correct, V1 and V2 will be removed in Q3, and we are no longer maintaining them