feat(pipeline): allow composing multiple modules via RunConfig.additional_modules

- Add RunConfig.additional_modules with normalization and to_dict serialization
- PipelineRunner: resolve/import module names or objects, pass modules to driver.Builder.with_modules,
  support reload for all modules, and handle async path
- PipelineVisualizer: accept additional_modules to render composite DAGs
- Utils/config: merge/dedupe additional_modules, add RunConfigBuilder.with_additional_modules,
  and plumb merge into merge_run_config_with_kwargs
- Docs/examples/specs/tasks/openspec: document feature and add README usage & example pipeline setup
- Add examples/hello-world/pipelines/setup.py and update hello_world example to reference composition
- Tests: add coverage for import, async path, reload, merge/dedupe behavior
- Changelog and pyproject version bump to 0.34.0; update CHANGELOG entry and uv.lock updates

Author: legout

CHANGELOG.md

@@ -1,5 +1,10 @@
 # Changelog
 
+## [0.34.0] - 2025-11-03
+
+### Added
+- Allow `PipelineManager.run`/`RunConfig` to load `additional_modules`, enabling multi-module pipeline execution and visualization.
+
 ## [0.31.0] - 2025-09-26
 
 ### Changes
@@ -992,4 +997,3 @@
 
 
 
-

examples/hello-world/pipelines/hello_world.py

@@ -17,40 +17,38 @@
 ).pipeline.h_params
 
 
-@config.when(range=10_000)
-def spend__10000() -> pd.Series:
-    """Returns a series of spend data."""
-    # time.sleep(2)
-    return pd.Series(range(10_000)) * 10
+# @config.when(range=10_000)
+# def spend__10000() -> pd.Series:
+#     """Returns a series of spend data."""
+#     # time.sleep(2)
+#     return pd.Series(range(10_000)) * 10
 
 
-@config.when(range=10_000)
-def signups__10000() -> pd.Series:
-    """Returns a series of signups data."""
-    time.sleep(1)
-    print(10_000)
-    return pd.Series(range(10_000))
+# @config.when(range=10_000)
+# def signups__10000() -> pd.Series:
+#     """Returns a series of signups data."""
+#     time.sleep(1)
+#     print(10_000)
+#     return pd.Series(range(10_000))
 
 
-@config.when(range=1_000)
-def spend__1000() -> pd.Series:
-    """Returns a series of spend data."""
-    # time.sleep(2)
-    print(1_000)
-    return pd.Series(range(10_000)) * 10
+# @config.when(range=1_000)
+# def spend__1000() -> pd.Series:
+#     """Returns a series of spend data."""
+#     # time.sleep(2)
+#     print(1_000)
+#     return pd.Series(range(10_000)) * 10
 
 
-@config.when(range=1_000)
-def signups__1000() -> pd.Series:
-    """Returns a series of signups data."""
-    time.sleep(1)
-    print(1_000)
-    return pd.Series(range(10_000))
+# @config.when(range=1_000)
+# def signups__1000() -> pd.Series:
+#     """Returns a series of signups data."""
+#     time.sleep(1)
+#     print(1_000)
+#     return pd.Series(range(10_000))
 
 
-@parameterize(
-    **PARAMS.avg_x_wk_spend
-)  # (avg_x_wk_spend={"rolling": value(3)})  #
+@parameterize(**PARAMS.avg_x_wk_spend)  # (avg_x_wk_spend={"rolling": value(3)})  #
 def avg_x_wk_spend(spend: pd.Series, rolling: int) -> pd.Series:
     """Rolling x week average spend."""
     # time.sleep(2)
@@ -68,9 +66,7 @@ def spend_mean(spend: pd.Series) -> float:
     return spend.mean()
 
 
-@parameterize(
-    **PARAMS.spend_zero_mean
-)  # (spend_zero_mean={"offset": value(0)})  #
+@parameterize(**PARAMS.spend_zero_mean)  # (spend_zero_mean={"offset": value(0)})  #
 def spend_zero_mean(spend: pd.Series, spend_mean: float, offset: int) -> pd.Series:
     """Shows function that takes a scalar. In this case to zero mean spend."""
     return spend - spend_mean + offset

examples/hello-world/pipelines/setup.py

@@ -0,0 +1,40 @@
+import sys
+import time
+from pathlib import Path
+
+import pandas as pd
+from hamilton.function_modifiers import config, parameterize
+from loguru import logger
+
+from flowerpower.cfg import Config
+
+
+@config.when(range=10_000)
+def spend__10000() -> pd.Series:
+    """Returns a series of spend data."""
+    # time.sleep(2)
+    return pd.Series(range(10_000)) * 10
+
+
+@config.when(range=10_000)
+def signups__10000() -> pd.Series:
+    """Returns a series of signups data."""
+    time.sleep(1)
+    print(10_000)
+    return pd.Series(range(10_000))
+
+
+@config.when(range=1_000)
+def spend__1000() -> pd.Series:
+    """Returns a series of spend data."""
+    # time.sleep(2)
+    print(1_000)
+    return pd.Series(range(10_000)) * 10
+
+
+@config.when(range=1_000)
+def signups__1000() -> pd.Series:
+    """Returns a series of signups data."""
+    time.sleep(1)
+    print(1_000)
+    return pd.Series(range(10_000))

openspec/changes/archive/2025-11-03-update-run-allow-additional-modules/proposal.md

@@ -0,0 +1,33 @@
+## Why
+Running a single pipeline module limits composition. Many FlowerPower projects benefit from shared setup/teardown steps (e.g., DB connections, secrets bootstrapping) and auxiliary pipelines. Hamilton natively supports composing multiple modules via `driver.Builder.with_modules(...)`. FlowerPower should expose this ability so users can run `pm.run("pipeline_1", additional_modules=["setup"])` without custom wiring.
+
+## What Changes
+- Add runtime support for specifying additional pipeline modules to load alongside the main pipeline.
+- Extend `RunConfig` with `additional_modules` (list of strings or modules at runtime) and plumb through to the runner.
+- Update runner to import and pass all modules to `driver.Builder.with_modules(*modules)`.
+- Ensure `reload=True` reloads all loaded modules.
+- Support both sync and async execution paths.
+
+Non‑breaking: Defaults preserve current single‑module behavior.
+
+## Impact
+- Affected specs: `pipeline-execution`
+- Affected code:
+  - `src/flowerpower/cfg/pipeline/run.py`
+  - `src/flowerpower/utils/config.py`
+  - `src/flowerpower/pipeline/runner.py`
+  - `src/flowerpower/pipeline/manager.py` (docs / passthrough)
+  - Tests under `tests/pipeline/` and `tests/cfg/`
+
+## Out of Scope
+- Changes to persistence, scheduling, or adapters behavior beyond module loading.
+- DAG conflict resolution beyond standard Hamilton rules (last definition wins).
+
+## Risks / Mitigations
+- Import errors for missing modules → raise clear ImportError with actionable path/name tips.
+- Conflicting node names across modules → document precedence (order) and suggest namespacing.
+
+## Rollout
+- Add tests for sync/async, reload, error handling.
+- Update README usage and CHANGELOG.
+

openspec/changes/archive/2025-11-03-update-run-allow-additional-modules/specs/pipeline-execution/spec.md

@@ -0,0 +1,36 @@
+## ADDED Requirements
+
+### Requirement: Run With Additional Modules
+The system SHALL support executing a pipeline while loading additional Python modules into the Hamilton driver so that DAG nodes can be composed across modules.
+
+#### Scenario: Execute with string module names
+- WHEN a user runs `pm.run("pipeline_1", additional_modules=["setup"])`
+- THEN the system imports both `setup` and `pipeline_1` modules and executes the composed DAG
+- AND functions in all loaded modules are available for dependency resolution
+
+#### Scenario: Execute with module objects
+- WHEN a user passes actual module objects in `additional_modules`
+- THEN the system uses these module objects directly without re-importing
+- AND execution succeeds identically to using string names
+
+#### Scenario: Missing module raises helpful error
+- WHEN any `additional_modules` entry cannot be imported
+- THEN the system raises an ImportError that indicates the attempted name and suggests `pipelines.<name>` and `'-'→'_'` formatting
+
+#### Scenario: Precedence order determines node override
+- WHEN multiple modules define the same node name
+- THEN the last module in the `.with_modules(*modules)` order SHALL take precedence (Hamilton standard behavior)
+
+#### Scenario: Reload reloads all loaded modules
+- GIVEN `reload=True` is set
+- WHEN execution starts
+- THEN the system reloads the main pipeline module and each additional module
+
+#### Scenario: Works for async execution
+- WHEN running with `run_async(...)` and `additional_modules` provided
+- THEN the system composes the same set of modules and executes asynchronously
+
+#### Scenario: Cross-module final_vars
+- WHEN `final_vars` references nodes defined in any loaded module
+- THEN the system resolves and returns those nodes successfully
+

openspec/changes/archive/2025-11-03-update-run-allow-additional-modules/tasks.md

@@ -0,0 +1,19 @@
+## 1. Implementation
+- [x] 1.1 Add `additional_modules` to `RunConfig` (list[str] | None)
+- [x] 1.2 Extend kwargs merge to support `additional_modules` (merge + dedupe)
+- [x] 1.3 Runner: resolve additional modules (string or module) and call `.with_modules(*modules)`
+- [x] 1.4 Reload logic: reload all additional modules when `reload=True`
+- [x] 1.5 PipelineManager.run docs: document `additional_modules`
+- [x] 1.6 (Optional) Visualizer: accept `additional_modules` to render composite DAG
+
+## 2. Tests
+- [x] 2.1 Runner passes multiple modules to Builder (sync)
+- [x] 2.2 Runner passes multiple modules to Builder (async)
+- [x] 2.3 Import resolution: string success and helpful ImportError on failure
+- [x] 2.4 Reload reloads all modules
+- [x] 2.5 Config merge: kwargs + config `additional_modules` merge and dedupe
+
+## 3. Docs & Release
+- [x] 3.1 README usage example (`pm.run("pipeline_1", additional_modules=["setup"])`)
+- [x] 3.2 CHANGELOG entry under Added
+- [ ] 3.3 Version bump as needed

openspec/changes/enhance-executor-overrides/proposal.md

@@ -0,0 +1,39 @@
+# pipeline-execution Specification
+
+## Purpose
+TBD - created by archiving change update-run-allow-additional-modules. Update Purpose after archive.
+## Requirements
+### Requirement: Run With Additional Modules
+The system SHALL support executing a pipeline while loading additional Python modules into the Hamilton driver so that DAG nodes can be composed across modules.
+
+#### Scenario: Execute with string module names
+- WHEN a user runs `pm.run("pipeline_1", additional_modules=["setup"])`
+- THEN the system imports both `setup` and `pipeline_1` modules and executes the composed DAG
+- AND functions in all loaded modules are available for dependency resolution
+
+#### Scenario: Execute with module objects
+- WHEN a user passes actual module objects in `additional_modules`
+- THEN the system uses these module objects directly without re-importing
+- AND execution succeeds identically to using string names
+
+#### Scenario: Missing module raises helpful error
+- WHEN any `additional_modules` entry cannot be imported
+- THEN the system raises an ImportError that indicates the attempted name and suggests `pipelines.<name>` and `'-'→'_'` formatting
+
+#### Scenario: Precedence order determines node override
+- WHEN multiple modules define the same node name
+- THEN the last module in the `.with_modules(*modules)` order SHALL take precedence (Hamilton standard behavior)
+
+#### Scenario: Reload reloads all loaded modules
+- GIVEN `reload=True` is set
+- WHEN execution starts
+- THEN the system reloads the main pipeline module and each additional module
+
+#### Scenario: Works for async execution
+- WHEN running with `run_async(...)` and `additional_modules` provided
+- THEN the system composes the same set of modules and executes asynchronously
+
+#### Scenario: Cross-module final_vars
+- WHEN `final_vars` references nodes defined in any loaded module
+- THEN the system resolves and returns those nodes successfully
+

openspec/changes/enhance-executor-overrides/tasks.md

@@ -4,7 +4,7 @@ description = "A simple workflow framework for building and managing data proces
 authors = [{ name = "Volker L.", email = "[email protected]" }]
 readme = "README.md"
 requires-python = ">= 3.11"
-version = "0.33.1"
+version = "0.34.0"
 keywords = ["hamilton", "workflow", "pipeline", "scheduler", "dask", "ray"]
 
 dependencies = [