Extract Mistral Workflows into standalone pipelex-mistralai-workflows package - #873
Extract Mistral Workflows into standalone pipelex-mistralai-workflows package#873lchoquel wants to merge 17 commits into
Conversation
- Updated `mistralai` dependency to version 2.4.4 in `pyproject.toml`. - Added a new `TODOS.md` file for planning and progress tracking of Mistral Workflows integration. - Introduced new activities and bridge for running Pipelex pipes within Mistral Workflows. - Implemented execution modes for Pipelex pipes in Mistral Workflows. - Added exception handling specific to Mistral Workflows. - Updated various imports to align with the new `mistralai.client` structure. - Enhanced type checking and error handling in workflow scripts. - Added integration tests for the new Mistral Workflows activities.
- Introduced the `pipelex.plugins.mistralai_workflows` plugin, enabling invocation of Pipelex pipes within Mistral Workflows activities. - Added three execution modes: `DIRECT`, `TEMPORAL_BLOCKING`, and `TEMPORAL_FIRE_AND_FORGET`, allowing flexible integration options. - Created documentation for the new plugin and usage recipes, detailing installation and execution modes. - Implemented integration tests for both blocking and fire-and-forget modes to ensure functionality. - Updated `CHANGELOG.md` to reflect these additions and improvements.
- Introduced `pipelex_run_pipe_offloaded` activity to handle large payloads by leveraging Mistral's `ActivityInOutOffloadingInterceptor`, allowing seamless streaming of oversized data through blob storage. - Updated `CHANGELOG.md` to document the new offloaded activity and its usage. - Added integration tests for the offloaded activity to ensure correct payload handling and functionality. - Enhanced documentation with examples for using the new offloaded activity and its configuration requirements.
- Added `pipelex_run_pipe_streaming` activity to enable real-time progress tracking of Pipelex pipes within Mistral Workflows, emitting lifecycle events (`CustomTaskStarted`, `CustomTaskInProgress`, `CustomTaskCompleted`, `CustomTaskFailed`). - Enhanced `DIRECT` execution mode to provide per-step event updates, allowing detailed observability of pipe execution. - Updated `CHANGELOG.md` to document the new streaming activity and its features. - Created integration tests for the streaming activity to validate event emissions and functionality. - Improved documentation with examples for using the new streaming variant in workflows.
- Implemented per-step event updates for the `pipelex_run_pipe_streaming` activity in `DIRECT` execution mode, allowing detailed observability of pipe execution through `CustomTaskInProgress` events. - Updated `CHANGELOG.md` to reflect the new features and improvements in streaming support. - Enhanced integration tests to validate the emission of per-step events during multi-step pipe executions. - Improved documentation with examples for utilizing the new streaming features in workflows.
Remove the `output: PipelexPipeRunOutput | None = None` initializer and the trailing `assert output is not None` in `_run_streaming_with_per_step_events`. The narrowing was assert-based (stripped under `python -O`) and only existed to satisfy the type checker after a `try`/`finally` cleanup block. With the pre-init dropped, pyright narrows `output` from the `try` body alone, since the post-`try/finally` code is only reachable on success. Extend the drain comment to call out that the cleanup serves both the happy path (snapshot ordering) and the failure path (publishing pending in-progress events before `Task.__aexit__` emits `CustomTaskFailed`).
- Updated the `instructor` dependency to a new commit hash for improved functionality.
Relocate the in-tree TODOS.md to wip/mistral-workflows-sub-module.md and add wip/mistral-workflows-plugin-extract.md outlining the extraction of the Mistral Workflows integration into its own pip-installable plugin package scaffolded from pipelex-starter-python.
…ce docs and clarify package structure
…ning and clarification on line references
…gration - Introduced a new `pipelex.runtime_bridge` package with core components for executing Pipelex pipes in various modes (DIRECT, TEMPORAL_BLOCKING, TEMPORAL_FIRE_AND_FORGET). - Added `ensure_pipelex_booted` function for idempotent initialization of Pipelex. - Created input/output models (`PipelexPipeRunInput`, `PipelexPipeRunOutput`) for structured data handling. - Implemented error handling with custom exceptions for runtime bridge operations. - Developed comprehensive integration tests for the runtime bridge, covering direct execution and library crate handling. - Updated documentation and TODOs to reflect the new structure and functionality.
- Extracted Mistral Workflows integration from `pipelex` into a new package `pipelex-mistralai-workflows`, allowing for independent installation and usage. - Removed the `pipelex.plugins.mistralai_workflows` modules and updated the core runtime bridge to be framework-agnostic. - Updated `CHANGELOG.md` to reflect the changes and migration instructions for users. - Deleted outdated documentation and tests related to the previous plugin structure. - Ensured no behavioral changes; all existing functionalities remain intact under the new package structure.
… updates and clarifying remaining tasks for plugin extraction
![chatgpt-codex-connector[bot]](https://avatars.githubusercontent.com/in/1144995?s=60&v=4){kind=small}
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: 5ccd4c33ba
ℹ️ About Codex in GitHub
Your team has set up Codex to review pull requests in this repo. Reviews are triggered when you
- Open a pull request for review
- Mark a draft as ready
- Comment "@codex review".
If Codex has suggestions, it will comment; otherwise it will react with 👍.
Codex can also answer questions or update the PR. Try commenting "@codex address that feedback".
| library_manager.teardown(library_id=library_id) | ||
| teardown_current_library() |
There was a problem hiding this comment.
When library_crate_dump is used from a context that already has an active library, this finally block always clears the ContextVar via teardown_current_library() instead of restoring the previous library id. That makes the bridge call leak state back to its caller: subsequent code in the same async context that expects the previously selected library will fail with “No current library set” or resolve against no library. Capture the current library before set_current_library(...) and restore it here, as the Temporal hydration helper does.
Useful? React with 👍 / 👎.
Greptile SummaryExtracts the Mistral Workflows integration out of the
Confidence Score: 3/5The core bridge logic and SDK migration are sound, but the temporal unit tests silently depend on an optional extra and will break in base CI environments, and the The temporal dispatch unit tests do not mock
Important Files Changed
|
![cubic-dev-ai[bot]](https://avatars.githubusercontent.com/in/1082092?s=60&v=4){kind=small}
![greptile-apps[bot]](https://avatars.githubusercontent.com/in/867647?s=60&v=4){kind=small}
| async def test_temporal_blocking_dispatches_to_temporal_pipe_run(self, mocker: MockerFixture) -> None: | ||
| fake_job = _make_fake_pipe_job(mocker=mocker, pipe_code="fake_pipe", pipeline_run_id="caller-run-id") | ||
| mocker.patch( | ||
| "pipelex.runtime_bridge.bridge.build_pipe_job_from_input", | ||
| return_value=fake_job, | ||
| ) | ||
|
|
||
| fake_output = PipeOutput( | ||
| working_memory=WorkingMemoryFactory.make_empty(), | ||
| pipeline_run_id="temporal-run-id", | ||
| ) | ||
| fake_temporal_run = mocker.AsyncMock(return_value=fake_output) | ||
| fake_factory = mocker.patch("pipelex.temporal.tprl_pipe.temporal_pipe_run.make_temporal_pipe_run") | ||
| fake_factory.return_value.run = fake_temporal_run | ||
|
|
||
| result = await run_pipe_via_bridge( | ||
| PipelexPipeRunInput( | ||
| pipe_code="fake_pipe", | ||
| execution_mode=PipelexExecutionMode.TEMPORAL_BLOCKING, | ||
| ) | ||
| ) | ||
|
|
||
| fake_factory.assert_called_once() | ||
| assert fake_temporal_run.await_count == 1 | ||
| assert result.is_completed is True | ||
| assert result.workflow_id == "temporal-run-id" |
There was a problem hiding this comment.
temporalio being installed
_require_pipelex_temporal_extra() is called inside run_pipe_via_bridge before the lazy import of make_temporal_pipe_run. In any CI environment that runs these unit tests without the temporal extra, the call will raise MissingPipelexTemporalExtraError before the mocked factory is ever reached — making the test fail with the wrong error. Adding mocker.patch("pipelex.runtime_bridge.bridge._require_pipelex_temporal_extra") to both temporal dispatch tests would make them truly isolated from the optional dependency. The same issue applies to test_temporal_fire_and_forget_returns_workflow_id_without_completion.
Prompt To Fix With AI
This is a comment left during a code review.
Path: tests/unit/pipelex/runtime_bridge/test_dispatch.py
Line: 69-94
Comment:
**Temporal unit tests depend on `temporalio` being installed**
`_require_pipelex_temporal_extra()` is called inside `run_pipe_via_bridge` *before* the lazy import of `make_temporal_pipe_run`. In any CI environment that runs these unit tests without the `temporal` extra, the call will raise `MissingPipelexTemporalExtraError` before the mocked factory is ever reached — making the test fail with the wrong error. Adding `mocker.patch("pipelex.runtime_bridge.bridge._require_pipelex_temporal_extra")` to both temporal dispatch tests would make them truly isolated from the optional dependency. The same issue applies to `test_temporal_fire_and_forget_returns_workflow_id_without_completion`.
How can I resolve this? If you propose a fix, please make it concise.…workflows # Conflicts: # CHANGELOG.md
1 participant
Summary
pipelex.runtime_bridgepackage — boundary types,run_pipe_via_bridge,PipelexExecutionMode,ensure_pipelex_booted, agnostic exceptions. Any host runtime can now embed Pipelex through this surface, not just Mistral Workflows.pipelex[mistralai-workflows]extra and the entirepipelex.plugins.mistralai_workflows.*subtree (activities, streaming, dependency wrapper, docs, tests). Mistral-Workflows-shaped activities now live in the newpipelex-mistralai-workflowsrepo on PyPI.CHANGELOG.md[Unreleased]with the migration note, drops themistralai-workflowsmypy override block frompyproject.toml, deletes the twodocs/under-the-hood/mistralai-workflows-*pages and theirmkdocs.ymlentries.runtime_bridgeunit + integration tests cover dispatch, validation, execution-mode selection, input model construction, and direct in-process bridge execution. Layer-2/3 tests (Mistral activity contract, Temporal end-to-end) move to the new repo's CI.Migration (for existing users of
pipelex[mistralai-workflows])```text
pip install pipelex-mistralai-workflows
imports change: pipelex.plugins.mistralai_workflows.* → pipelex_mistralai_workflows.*
```
No behavior changes; activities, boundary types, and execution modes are identical to the previous in-tree plugin.
Test plan
Follow-ups (tracked in `TODOS.md` Stream D)
Summary by cubic
Extracted the Mistral Workflows integration into the standalone
pipelex-mistralai-workflowspackage and introduced a framework-agnosticpipelex.runtime_bridgeto run Pipelex pipes from any host runtime. Upgraded to themistralaiv2 client and bumped Python/Temporal versions.New Features
pipelex.runtime_bridge: boundary models,run_pipe_via_bridge,PipelexExecutionMode,ensure_pipelex_booted, and bridge-specific errors.pipelex[mistralai-workflows]extra and in-treepipelex.plugins.mistralai_workflows.*; Mistral-shaped activities now live inpipelex-mistralai-workflows.mistralai.client.*(Mistral, models, errors) and added unit/integration tests for the bridge; host-runtime tests moved to the new repo.Migration
pipelex-mistralai-workflowsand update imports:pipelex.plugins.mistralai_workflows.*→pipelex_mistralai_workflows.*.mistralai >= 2.4.4,temporalio == 1.24.0. If you import Mistral types directly, switch tomistralai.client.*.Written for commit a5fb4dd. Summary will update on new commits.