ROSE As A Service-Core#91
Conversation
- Adds radical.edge rose plugin - ROSE Rest API via the introduced plguin
1-Fix pre-commits 2-Sets the RAAS branch up to data with main new changes
RASS iteration
…nflicts Fix ROSE As a Service merge conflicts
Summary of ChangesHello, I'm Gemini Code Assist1! I'm currently reviewing this pull request and will post my feedback shortly. In the meantime, here's a summary to help you and other reviewers quickly get up to speed! This pull request introduces the foundational "ROSE Service" to transform how Active Learning workflows are managed and executed. It establishes a robust, daemon-based system that allows users to submit, monitor, and control ROSE workflows through various interfaces, including a command-line interface, a programmatic Python client utilizing file-based inter-process communication, and a comprehensive REST API integrated via a RADICAL-Edge plugin. This enhancement significantly improves the flexibility and integration capabilities of ROSE, enabling it to function as a persistent, managed service for complex scientific workflows. Highlights
🧠 New Feature in Public Preview: You can now enable Memory to help Gemini Code Assist learn from your team's feedback. This makes future code reviews more consistent and personalized to your project's style. Click here to enable Memory in your admin console. Using Gemini Code AssistThe full guide for Gemini Code Assist can be found on our documentation page, here are some quick tips. Invoking Gemini You can request assistance from Gemini at any point by creating a comment using either
Customization To customize Gemini Code Assist for GitHub experience, repository maintainers can create a configuration file and/or provide a custom code review style guide (such as PEP-8 for Python) by creating and adding files to a Limitations & Feedback Gemini Code Assist may make mistakes. Please leave feedback on any instances where its feedback is incorrect or counter productive. You can react with 👍 and 👎 on @gemini-code-assist comments. If you're interested in giving your feedback about your experience with Gemini Code Assist for GitHub and other Google products, sign up here. Footnotes
|
![gemini-code-assist[bot]](https://avatars.githubusercontent.com/in/956858?s=60&v=4){kind=small}
There was a problem hiding this comment.
Code Review
This pull request introduces a new ROSE Service for managing Active Learning workflows, providing CLI, Python client, and REST API interfaces. Key changes include new service components, client implementations, example workflows, and updates to pyproject.toml for dependencies and entry points. The review identified several critical and high-severity issues, such as hardcoded absolute paths in example YAMLs, redundant code in the plugin example, fragile workflow ID retrieval in client examples, and missing f-string prefixes in logging. Further improvements were suggested for client directory creation robustness, documentation clarity, dependency management, and error handling.
| path: /home/aymen/ve/raas/bin/python3 | ||
| config: | ||
| args: ["/home/aymen/RADICAL/ROSE-AS-A-SERVICE/ROSE/examples/active_learn/basic/sim.py"] |
There was a problem hiding this comment.
This example workflow file contains hardcoded absolute paths to a specific user's home directory (e.g., /home/aymen/...). This makes the example unusable for other users and in different environments. Please replace these with relative paths, placeholders (e.g., <path_to_your_script>), or use environment variables to make the example portable.
| state = status.get("state", "UNKNOWN") | ||
|
|
||
| if state in terminal: | ||
| log.info("-" * 50) | ||
| if state == "COMPLETED": | ||
| log.info(f"Workflow {wf_id} completed successfully") | ||
| elif state == "FAILED": | ||
| log.error(f"Workflow {wf_id} failed: {status.get('error')}") | ||
| else: | ||
| log.warning(f"Workflow {wf_id} was canceled") | ||
| break |
There was a problem hiding this comment.
This block of code appears to be redundant and is likely unreachable. The while loop starting on line 145 contains a try...except block that checks for a terminal state and breaks out of the loop if one is found (line 157). This means the code from line 161 onwards will never be executed. This looks like a copy-paste error and should be removed to avoid confusion and dead code.
| wf_id = None | ||
| print("Waiting for service to assign Workflow ID...") | ||
| for _ in range(20): | ||
| await asyncio.sleep(1) | ||
| registry = client.list_workflows() | ||
| if registry: | ||
| wf_id = list(registry.keys())[0] | ||
| print(f"Assigned Workflow ID: {wf_id}") | ||
| break |
There was a problem hiding this comment.
The current implementation for retrieving the wf_id polls list_workflows() and assumes the first workflow in the registry is the one just submitted. This is fragile and not guaranteed to be correct, especially in a multi-workflow scenario. A more robust approach is to derive the wf_id directly from the req_id using the static method ServiceClient.get_wf_id(), as documented in the README.md.
| wf_id = None | |
| print("Waiting for service to assign Workflow ID...") | |
| for _ in range(20): | |
| await asyncio.sleep(1) | |
| registry = client.list_workflows() | |
| if registry: | |
| wf_id = list(registry.keys())[0] | |
| print(f"Assigned Workflow ID: {wf_id}") | |
| break | |
| wf_id = ServiceClient.get_wf_id(req_id) | |
| print(f"Assigned Workflow ID: {wf_id}") |
| wf_id = None | ||
| for _ in range(10): | ||
| await asyncio.sleep(1) | ||
| registry = client.list_workflows() | ||
| if registry: | ||
| wf_id = list(registry.keys())[0] | ||
| state = registry[wf_id]["state"] | ||
| print(f"Workflow {wf_id} State: {state}") | ||
| if state in ["RUNNING", "COMPLETED"]: | ||
| break |
There was a problem hiding this comment.
The method used to obtain the wf_id by polling list_workflows() and taking the first key is not robust. It's better to derive the wf_id directly from the req_id returned by submit_workflow using ServiceClient.get_wf_id(). This ensures you are targeting the correct workflow, especially in scenarios where multiple workflows could be present.
| wf_id = None | |
| for _ in range(10): | |
| await asyncio.sleep(1) | |
| registry = client.list_workflows() | |
| if registry: | |
| wf_id = list(registry.keys())[0] | |
| state = registry[wf_id]["state"] | |
| print(f"Workflow {wf_id} State: {state}") | |
| if state in ["RUNNING", "COMPLETED"]: | |
| break | |
| wf_id = ServiceClient.get_wf_id(req_id) | |
| print(f"Assigned Workflow ID: {wf_id}") | |
| # Poll for status until running | |
| for _ in range(10): | |
| await asyncio.sleep(1) | |
| status = client.get_workflow_status(wf_id) | |
| if status: | |
| state = status['state'] | |
| print(f"Workflow {wf_id} State: {state}") | |
| if state in ["RUNNING", "COMPLETED"]: | |
| break |
| wf.start_time = time.time() | ||
| self._notify_state(wf) | ||
|
|
||
| log.info("[{self.sid}] Running workflow {wf_id}") |
There was a problem hiding this comment.
This log message is missing the f prefix for an f-string, so it will log the literal string [{self.sid}] Running workflow {wf_id} instead of the formatted values. This will make debugging difficult. The same issue exists on line 251. Please add the f prefix to all similar log messages.
| log.info("[{self.sid}] Running workflow {wf_id}") | |
| log.info(f"[{self.sid}] Running workflow {wf_id}") |
|
|
||
| backend = LocalExecutionBackend() | ||
| self.engine = await WorkflowEngine.create(backend) | ||
| logger.info("Service initialized at {self.service_root}") |
There was a problem hiding this comment.
This log message is missing the f prefix for an f-string. It will log the literal string Service initialized at {self.service_root} instead of the formatted path. This makes the log message unhelpful for debugging.
This issue is present in multiple places in this file. Please fix all occurrences, including:
- line 224
- line 250
- line 281
- line 301
- line 305
- line 310
- line 342
- line 348
| logger.info("Service initialized at {self.service_root}") | |
| logger.info(f"Service initialized at {self.service_root}") |
| ## Additional Files | ||
|
|
||
| | File | Description | | ||
| |------|-------------| | ||
| | `example_rose_plugin.py` | REST API example using RADICAL-Edge plugin | | ||
| | `example_minimal.py` | Minimal example (30 lines) for quick reference | | ||
| | `debug_workflow.yaml` | Fast test workflow (2 iterations, ~6 seconds) | |
There was a problem hiding this comment.
The "Additional Files" section at the end of the file is redundant as all the files listed here (example_rose_plugin.py, example_minimal.py, debug_workflow.yaml) are already described in the table at the beginning of the document (lines 9-17). To improve clarity and avoid repetition, I suggest removing this section.
|
|
||
| service = [ | ||
| "fastapi>=0.100", | ||
| "radical.edge @ git+https://github.com/radical-cybertools/radical.edge", |
There was a problem hiding this comment.
Using a direct git repository URL for a dependency can lead to issues with package resolution and installation for end-users, as it's not a standard versioned package from PyPI. If a stable version of radical.edge is available on PyPI, it would be more robust to depend on that instead. If not, this is acceptable but be aware of the potential for future breakages if the remote repository changes.
| except Exception as e: | ||
| print(f"Error submitting workflow: {e}") | ||
| sys.exit(1) |
There was a problem hiding this comment.
Catching a broad Exception can hide bugs and make debugging more difficult. It's better to catch more specific exceptions that you expect the client to raise, such as FileNotFoundError if the workflow file doesn't exist, or custom exceptions from the ServiceClient. This will make the error handling more robust and provide more specific feedback to the user.
| eliminating the need for a separate ServiceManager daemon. Each RoseSession | ||
| maintains its own WorkflowEngine and executes learner loops as async tasks. | ||
|
|
||
| :: |
There was a problem hiding this comment.
The ::: syntax used here for a code block is not standard for Python docstrings (which often use reStructuredText or Markdown). For better rendering in documentation tools, you should use a standard code block format, such as ````.
| :: |
radical.edge changed the notification pattern from a per-session _notify closure to a _plugin reference with _dispatch_notify(). Update all callsites in RoseSession to match. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
Four new tests in TestRoseSession: - test_submit_workflow_dispatches_submitted_notification: checks that submit_workflow fires workflow_state/SUBMITTED via _dispatch_notify - test_notify_state_calls_dispatch_notify: checks _notify_state payload - test_notify_state_no_plugin_does_not_raise: guards against AttributeError when _plugin is None (bare unit-test creation without a plugin) - test_task_event_dispatched_via_plugin: verifies the _on_done callback in _run_workflow fires a task_event notification Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
2 participants
No description provided.