diff --git a/_typos.toml b/_typos.toml
new file mode 100644
index 00000000..82f6f264
--- /dev/null
+++ b/_typos.toml
@@ -0,0 +1,3 @@
+[default.extend-words]
+# "aas" appears in rose-aas.md / "ROSE as a Service" links — not a typo of "as"/"ass".
+aas = "aas"
diff --git a/docs/getting-started/faq.md b/docs/getting-started/faq.md
index c0011c2e..3f2a63c5 100644
--- a/docs/getting-started/faq.md
+++ b/docs/getting-started/faq.md
@@ -9,6 +9,9 @@ Refer to the [Installation Guide](installation.md) for detailed instructions bas
 A dry run allows you to simulate the execution of your tasks locally without making any changes to the behavior of your ML workflow.
 It's a way to verify your setup is working on HPC before applying it to prevent time and resource waste.
 
+!!! tip
+    See the [Dry Run](dry-run.md) guide for a full step-by-step walkthrough.
+
 ## Q3: What if I have an issue?
 If you have any additional questions or an issue, feel free to reach out to our support team by opening a Github [ticket](https://github.com/radical-cybertools/ROSE/issues).
 
diff --git a/docs/getting-started/installation.md b/docs/getting-started/installation.md
index b3c2cad5..7000283f 100644
--- a/docs/getting-started/installation.md
+++ b/docs/getting-started/installation.md
@@ -7,7 +7,7 @@ machines, please refer to the following link: [RADICAL-Pilot Supported HPC Machi
 ## For Linux and macOS 🐧
 
 1. Clone the latest version from the [official website](https://github.com/radical-cybertools/ROSE).
-   ```
+   ```bash
    git clone https://github.com/radical-cybertools/ROSE.git
    ```
 2. Run the following commands to install ROSE and its dependencies:
@@ -21,19 +21,19 @@ machines, please refer to the following link: [RADICAL-Pilot Supported HPC Machi
 1. Download the Windows WSL installer from the [official website](https://apps.microsoft.com/detail/9pdxgncfsczv?hl=en-US&gl=US).
 2. Setup you WSL user name and password.
 3. Make sure you have Python 3.9 or higher in your WSL as follows:
-    ```
+    ```bash
     python --version
     ```
 4. create new pip virtual env:
-    ```
+    ```bash
     python3 -m venv rose_env
     ```
 5. Activate the env:
-   ```
+   ```bash
    source rose_env/bin/activate
    ```
 6. Clone the latest version from the [official website](https://github.com/radical-cybertools/ROSE).
-   ```
+   ```bash
    git clone https://github.com/radical-cybertools/ROSE.git
    ```
 7. Run the following commands to install ROSE and its dependencies:
@@ -42,4 +42,5 @@ machines, please refer to the following link: [RADICAL-Pilot Supported HPC Machi
     pip install .
     ```
 
-If you encounter any issues, refer to the [Issues Section](https://github.com/radical-cybertools/ROSE/issues).
+!!! note
+    If you encounter any issues, refer to the [Issues Section](https://github.com/radical-cybertools/ROSE/issues).
diff --git a/docs/index.md b/docs/index.md
index 76823901..800f1b89 100644
--- a/docs/index.md
+++ b/docs/index.md
@@ -1,6 +1,6 @@
 ## What is ROSE? 🌹
 
-ROSE: RADICAL Orchestrator for Surrogate Exploration (ROSE) toolkit is a framework designed to enable the concurrent and adaptive execution of simulation, surrogate training, and selection tasks on High-Performance Computing (HPC) resources. ROSE is a Python package that provides tools for developing active and reinforcement learning (AL and RL) methods for scientific applications. It enables users to define simulation and surrogate training tasks and automatically manage their execution on HPC resources via a predefined set of Learning Policies (Learners).
+ROSE: RADICAL Orchestrator for Surrogate Exploration (ROSE) toolkit is a framework designed to enable the distributed, concurrent and adaptive execution of simulation, surrogate training, and selection tasks on High-Performance Computing (HPC) resources at a scale. ROSE is a Python package that provides tools for developing active and reinforcement learning (AL and RL) methods for scientific applications. It enables users to define simulation and surrogate training tasks and automatically manage their execution on HPC resources via a predefined set of Learning Policies (Learners).
 
 ROSE also includes tools to facilitate the selection of the most effective surrogate model for a given simulation based on performance metrics.
 
@@ -9,7 +9,7 @@ ROSE leverages [**RADICAL-Cybertools**](https://radical-cybertools.github.io), a
 
 ## Why ROSE? 🚀🚀🚀
 ROSE allows you to enable, scale, and accelerate your learning workflows across thousands of CPU cores and GPUs effectively and efficiently with just a few lines of code.
-ROSE is built on the [**RADICAL-AsyncFlow**](https://radical-cybertools.github.io/radical.asyncflow/) and [**RADICAL-Pilot**](https://github.com/radical-cybertools/radical.pilot) runtime system, a powerful execution engine that enables the distributed execution of millions of scientific tasks and applications such as executables, functions and containers effortlessly.
+ROSE is built on the [**RADICAL-AsyncFlow**](https://radical-cybertools.github.io/radical.asyncflow/) and [**RHAPSODY**](https://radical-cybertools.github.io/rhapsody/landing.html) runtime system, a powerful execution engine that enables the distributed execution of millions of scientific tasks and applications such as executables, functions and containers effortlessly.
 
 **Preemption-safe on HPC.** Every completed iteration is written to disk before the next one starts. If the job is killed mid-run, all completed iterations are already on disk — inspect them, resume from the last checkpoint, and never rerun a finished iteration.
 
diff --git a/docs/integrations/clearml.md b/docs/integrations/clearml.md
index d5b1ee79..0eb2a461 100644
--- a/docs/integrations/clearml.md
+++ b/docs/integrations/clearml.md
@@ -61,6 +61,10 @@ For **parallel learners**, `state.learner_id` is included automatically. The tra
 each state as a separate `series` inside the same scalar title, making per-learner curves
 directly comparable without any user code.
 
+!!! note
+    No user code is required to get this overlay — `state.learner_id` is already populated
+    by the parallel learner framework before the tracker ever sees the state.
+
 ### Task tags — logged in `on_stop`
 
 | ClearML tag | Value |
diff --git a/docs/scripts/custom.js b/docs/scripts/custom.js
new file mode 100644
index 00000000..7e385009
--- /dev/null
+++ b/docs/scripts/custom.js
@@ -0,0 +1,38 @@
+// ROSE docs — subtle scroll-reveal for a more fluid reading experience.
+// Pairs with the .rose-reveal / .rose-revealed rules in styles/custom.css.
+(() => {
+  const reduceMotion = window.matchMedia("(prefers-reduced-motion: reduce)").matches;
+  if (reduceMotion || !("IntersectionObserver" in window)) return;
+
+  const SELECTOR =
+    ".md-content__inner > h2, .md-content__inner > h3, .md-content__inner > p, " +
+    ".md-content__inner > table, .md-content__inner > .admonition, " +
+    ".md-content__inner > pre, .md-content__inner > .highlight";
+
+  const observer = new IntersectionObserver(
+    (entries) => {
+      entries.forEach((entry) => {
+        if (entry.isIntersecting) {
+          entry.target.classList.add("rose-revealed");
+          observer.unobserve(entry.target);
+        }
+      });
+    },
+    { threshold: 0.05, rootMargin: "0px 0px -40px 0px" }
+  );
+
+  const observe = () => {
+    document.querySelectorAll(SELECTOR).forEach((el) => {
+      el.classList.add("rose-reveal");
+      observer.observe(el);
+    });
+  };
+
+  // mkdocs-material's instant navigation swaps content without a full page
+  // load, so re-run on each page change via its document$ observable.
+  if (typeof document$ !== "undefined") {
+    document$.subscribe(observe);
+  } else {
+    document.addEventListener("DOMContentLoaded", observe);
+  }
+})();
diff --git a/docs/styles/custom.css b/docs/styles/custom.css
index c21edcfb..df3e7221 100644
--- a/docs/styles/custom.css
+++ b/docs/styles/custom.css
@@ -1,15 +1,186 @@
-/* Change the banner background color to dark rose */
+/* ============================================================
+   ROSE docs — fluid theme refinement
+   Cohesive rose/pink brand palette, softened rounded surfaces,
+   smooth motion, and viewport-responsive (fluid) type & spacing.
+   ============================================================ */
+
+/* ---- 1. Brand palette --------------------------------------- */
+:root {
+  --rose-50:  #fdf2f6;
+  --rose-100: #fbe4ee;
+  --rose-300: #f06ea6;
+  --rose-500: #d81b60;
+  --rose-600: #c2185b;
+  --rose-700: #a3134c;
+  --rose-900: #6e0e35;
+
+  --md-primary-fg-color:             var(--rose-600);
+  --md-primary-fg-color--light:      var(--rose-300);
+  --md-primary-fg-color--dark:       var(--rose-900);
+  --md-accent-fg-color:              var(--rose-500);
+  --md-accent-fg-color--transparent: rgba(216, 27, 96, 0.1);
+
+  --rose-radius:       0.6rem;
+  --rose-radius-lg:    1rem;
+  --rose-shadow:       0 2px 10px rgba(194, 24, 91, 0.08), 0 1px 2px rgba(0, 0, 0, 0.04);
+  --rose-shadow-hover: 0 6px 20px rgba(194, 24, 91, 0.16), 0 2px 4px rgba(0, 0, 0, 0.06);
+  --rose-transition:   180ms cubic-bezier(0.4, 0, 0.2, 1);
+}
+
+[data-md-color-scheme="slate"] {
+  --md-primary-fg-color:        var(--rose-700);
+  --md-primary-fg-color--dark:  #4a0a26;
+  --md-accent-fg-color:         var(--rose-300);
+  --rose-shadow:       0 2px 12px rgba(0, 0, 0, 0.35);
+  --rose-shadow-hover: 0 8px 24px rgba(0, 0, 0, 0.45);
+}
+
+/* ---- 2. Fluid typography & layout ----------------------------- */
+html {
+  font-size: clamp(18px, 0.6vw + 16px, 22px);
+  scroll-behavior: smooth;
+}
+
+.md-typeset h1 { font-size: clamp(1.8rem, 1.2rem + 1.6vw, 2.6rem); }
+.md-typeset h2 { font-size: clamp(1.4rem, 1.1rem + 1vw, 2rem); }
+.md-typeset h3 { font-size: clamp(1.15rem, 1rem + 0.5vw, 1.5rem); }
+
+.md-typeset {
+  line-height: 1.65;
+}
+
+.md-content__inner {
+  padding: clamp(0.8rem, 1vw + 0.5rem, 1.8rem) clamp(0.6rem, 2vw, 2rem);
+}
+
+.md-grid {
+  max-width: clamp(960px, 90vw, 1440px);
+}
+
+/* ---- 3. Header & tabs: subtle gradient instead of flat fill --- */
 .md-header {
-  background-color: #c2185b !important; /* Dark rose color */
+  background: linear-gradient(135deg, var(--rose-600), var(--rose-700));
+  box-shadow: 0 2px 12px rgba(194, 24, 91, 0.25);
+  transition: box-shadow var(--rose-transition), transform var(--rose-transition);
+}
+
+.md-tabs {
+  background: linear-gradient(135deg, var(--rose-700), var(--rose-900));
+}
+
+/* ---- 4. Motion: links, nav, tabs ------------------------------ */
+a,
+.md-nav__link,
+.md-tabs__link,
+.md-typeset a {
+  transition: color var(--rose-transition), background-color var(--rose-transition),
+    opacity var(--rose-transition);
 }
 
-/* Ensure the active navigation link is distinct */
 .md-nav__link--active {
-  color: #c2185b !important; /* Slightly darker black for active link */
-  font-weight: bold;
+  color: var(--rose-600) !important;
+  font-weight: 600;
+}
+
+[data-md-color-scheme="slate"] .md-nav__link--active {
+  color: var(--rose-300) !important;
+}
+
+.md-nav__link:hover {
+  opacity: 0.75;
 }
 
+.md-typeset a:hover {
+  text-decoration: underline;
+}
+
+/* ---- 5. Softened, elevated surfaces ---------------------------- */
+.md-typeset .admonition,
+.md-typeset details {
+  border-radius: var(--rose-radius);
+  box-shadow: var(--rose-shadow);
+  border-width: 0;
+  border-left: 4px solid var(--md-accent-fg-color);
+  overflow: hidden;
+  transition: box-shadow var(--rose-transition);
+}
+
+.md-typeset .admonition:hover,
+.md-typeset details:hover {
+  box-shadow: var(--rose-shadow-hover);
+}
+
+.md-typeset pre,
+.md-typeset .highlight {
+  border-radius: var(--rose-radius);
+  box-shadow: var(--rose-shadow);
+}
 
+.md-typeset code {
+  border-radius: 0.3rem;
+}
+
+.md-typeset table:not([class]) {
+  border-radius: var(--rose-radius);
+  overflow: hidden;
+  box-shadow: var(--rose-shadow);
+  border-collapse: separate;
+  border-spacing: 0;
+}
+
+.md-typeset table:not([class]) th {
+  background-color: var(--md-accent-fg-color--transparent);
+}
+
+.md-typeset table:not([class]) tr {
+  transition: background-color var(--rose-transition);
+}
+
+.md-typeset table:not([class]) tr:hover {
+  background-color: var(--md-accent-fg-color--transparent);
+}
+
+.md-typeset img,
+.md-typeset figure {
+  border-radius: var(--rose-radius-lg);
+}
+
+/* ---- 6. Search box: rounded, smooth focus ---------------------- */
+.md-search__form {
+  border-radius: 2rem;
+  transition: background-color var(--rose-transition), box-shadow var(--rose-transition);
+}
+
+.md-search__input:focus ~ .md-search__form,
+.md-search__form:hover {
+  box-shadow: 0 0 0 3px var(--md-accent-fg-color--transparent);
+}
+
+/* ---- 7. Buttons / copy button ----------------------------------- */
+.md-typeset .md-button,
+.md-clipboard {
+  border-radius: var(--rose-radius);
+  transition: transform var(--rose-transition), box-shadow var(--rose-transition);
+}
+
+.md-typeset .md-button:hover {
+  transform: translateY(-1px);
+  box-shadow: var(--rose-shadow);
+}
+
+/* ---- 8. Scroll-reveal (paired with scripts/custom.js) ----------- */
+.rose-reveal {
+  opacity: 0;
+  transform: translateY(8px);
+  transition: opacity 420ms ease, transform 420ms ease;
+}
+
+.rose-reveal.rose-revealed {
+  opacity: 1;
+  transform: translateY(0);
+}
+
+/* ---- 9. Config-reference table (existing component, restyled) --- */
 table#config {
   font-family: monospace;
   font-size: 13px;
@@ -22,20 +193,26 @@ table#config span.value {
   font-size: 0.9em;
   filter: opacity(70%);
   display: inline-block;
-  padding: 0px 4px;
-  border-radius: 5px;
+  padding: 1px 6px;
+  border-radius: 999px;
 }
 table#config span.example {
-  background-color: lightBlue;
+  background-color: var(--rose-100);
+  color: var(--rose-700);
 }
 table#config span.value {
-  background-color: lightGray;
+  background-color: rgba(0, 0, 0, 0.06);
 }
 table#config td.comment:nth-child(3) {
-  border-left: 2px solid whiteSmoke;
+  border-left: 2px solid rgba(0, 0, 0, 0.08);
 }
 table#config td.type {
   font-size: 0.8em;
+  filter: opacity(70%);
+  color: #999999;
+}
+table#config tr {
+  transition: background-color var(--rose-transition);
 }
 table#config tr:hover code {
   background-color: hsla(0, 0%, 100%, 1);
@@ -53,10 +230,21 @@ table#config td .anchor:target::before {
   content: "";
   display: block;
 }
-table#config td.type {
-  filter: opacity(70%);
-  color: #999999;
-}
 .md-typeset__scrollwrap {
   overflow-x: inherit;
 }
+
+/* ---- 10. Respect reduced-motion preference ----------------------- */
+@media (prefers-reduced-motion: reduce) {
+  html {
+    scroll-behavior: auto;
+  }
+  * {
+    transition-duration: 0.01ms !important;
+    animation-duration: 0.01ms !important;
+  }
+  .rose-reveal {
+    opacity: 1;
+    transform: none;
+  }
+}
diff --git a/docs/user-guide/acl-metrics.md b/docs/user-guide/acl-metrics.md
index d8d71dc9..4a303f7d 100644
--- a/docs/user-guide/acl-metrics.md
+++ b/docs/user-guide/acl-metrics.md
@@ -4,7 +4,7 @@ ROSE supports different Machine Learning (ML) Metrics such as `RMSE`, `MAE`, and
 
 
 ## Standard Metrics
-For a full list of the supported metrics please refer to the following link [ROSE Standard Metrics](https://github.com/radical-cybertools/ROSE/blob/feature/al_algo_selector/rose/metrics.py)
+For a full list of the supported metrics please refer to the following link [ROSE Standard Metrics](https://github.com/radical-cybertools/ROSE/blob/main/rose/metrics.py)
 
 
 ## Custom Metrics
@@ -26,3 +26,8 @@ async def check_metric(*args):
 ```
 
 In this way, ROSE will understand the relation between the custom metric and the target threshold value.
+
+!!! note
+    `metric_name` is just a label used for logging and tracking — ROSE does not validate it
+    against a fixed list. Any string is accepted as long as the decorated function returns a
+    numerical value to compare against `threshold`.
diff --git a/docs/user-guide/advanced-acl-workflow.md b/docs/user-guide/advanced-acl-workflow.md
index 6c501bcc..64412395 100644
--- a/docs/user-guide/advanced-acl-workflow.md
+++ b/docs/user-guide/advanced-acl-workflow.md
@@ -1,26 +1,39 @@
+# Advanced Active Learning Workflow
+
 To support the rapid advancement of AL techniques, ROSE offers an additional approach to building and executing complex AL workflows.
 
 In this example, we demonstrate how to express an AL workflow with different levels of parallelism. What does that mean?
 
 In some cases, AL workflows may require the execution of N simulation or training tasks **concurrently**. But not only that—additionally, they may also require the submission of M AL workflows concurrently. This introduces two levels of parallelism: one at the task level and another at the AL workflow level. Such an approach is possible and can be easily expressed and executed using ROSE's **custom AL policy**.
 
-```sh
-                             (N AL WFs in Parallel)
-          +-------------------+               +-------------------+
-          |      AL WF 1      |               |      AL WF 2      |
-          +-------------------+               +-------------------+
-                   │                                    │
-  +----------------+-----------------+  +----------------+-----------------+
-  |       (N tasks Parallel)         |  |       (N AL tasks Parallel)      |
-  +---------------+  +---------------+  +---------------+  +---------------+
-  | Simulation 1  |  | Simulation 2  |  | Simulation 1  |  | Simulation 2  |
-  +---------------+  +---------------+  +---------------+  +---------------+
-          |                |                    |                 |
-  +---------------+  +---------------+  +---------------+  +---------------+
-  |  Training 1   |  |  Training 2   |  |  Training 1   |  |  Training 2   |
-  +---------------+  +---------------+  +---------------+  +---------------+
-          |                |                    |                 |
-        (...)            (...)                (...)             (...)
+!!! note
+    Task-level parallelism (N simulation/training tasks per iteration) and workflow-level
+    parallelism (M AL loops running side by side) are independent dimensions — you can scale
+    either one without changing the other.
+
+```mermaid
+graph TD
+    N["N AL WFs in Parallel"]
+    N --> WF1["AL WF 1"]
+    N --> WF2["AL WF 2"]
+
+    subgraph G1[" "]
+        WF1 --> S1a["Simulation 1"]
+        WF1 --> S1b["Simulation 2"]
+        S1a --> T1a["Training 1"]
+        S1b --> T1b["Training 2"]
+        T1a --> E1a["..."]
+        T1b --> E1b["..."]
+    end
+
+    subgraph G2[" "]
+        WF2 --> S2a["Simulation 1"]
+        WF2 --> S2b["Simulation 2"]
+        S2a --> T2a["Training 1"]
+        S2b --> T2b["Training 2"]
+        T2a --> E2a["..."]
+        T2b --> E2b["..."]
+    end
 ```
 
 Since we have already learned how to deploy and load ROSE, and how to instruct it to use different resources, we will skip this part and focus only on expressing the AL workflow.
diff --git a/docs/user-guide/advanced-rl-workflow.md b/docs/user-guide/advanced-rl-workflow.md
index 25323987..6c55920f 100644
--- a/docs/user-guide/advanced-rl-workflow.md
+++ b/docs/user-guide/advanced-rl-workflow.md
@@ -1,39 +1,27 @@
+# Advanced Reinforcement Learning Workflow
+
 In addition to basic reinforcement learning (RL) workflows, ROSE supports advanced RL workflows that can run multiple environment instances in parallel.
 
 The 'ParallelLearner' gives you the ability to run multiple environment tasks simultaneously, each with different parameters, and then merge their experiences for training.
 
 This is particularly useful for scenarios where you want to explore different configurations or hyperparameters in parallel, speeding up the learning process.
-```sh
-
-                +-------------------+
-                |        RL WF      |
-                +-------------------+
-                            │
-  +-------------------------+---------------------------+
-  |             (N Environment Tasks Parallel)          |
-  +---------------+  +---------------+  +---------------+
-  | Environment 1 |  | Environment 2 |  | Environment 3 |
-  +---------------+  +---------------+  +---------------+
-          |                |                    |
-          └────────────────┼────────────────────┘
-                           │
-                    +------v------+
-                    |    Merge    |
-                    +------+------+
-                           │
-                    +------v------+
-                    |   Update    |
-                    +------+------+
-                           │
-                    +------v------+
-                    |    Test     |
-                    +-------------+
+```mermaid
+graph TD
+    WF["RL WF"]
+    WF --> E1["Environment 1"]
+    WF --> E2["Environment 2"]
+    WF --> E3["Environment 3"]
+    E1 --> M["Merge"]
+    E2 --> M
+    E3 --> M
+    M --> U["Update"]
+    U --> T["Test"]
 ```
 Import ROSE parallel RL modules:
 
 ```python
 from radical.asyncflow import WorkflowEngine
-from rhapsody.backends import RadicalExecutionBackend
+from rhapsody.backends import DragonExecutionBackendV3
 from rose.rl.reinforcement_learner import SequentialReinforcementLearner
 ```
 
@@ -44,10 +32,7 @@ from rose.rl.reinforcement_learner import SequentialReinforcementLearner
 
 async def main():
 
-    execution_engine = await RadicalExecutionBackend(
-        {'runtime': 30,
-        'resource': 'local.localhost'}
-        )
+    execution_engine = await DragonExecutionBackendV3()
 
     asyncflow = await WorkflowEngine.create(execution_engine)
 
@@ -80,8 +65,7 @@ async def main():
 Now that each environment task is defined, we define the rest of the workflow components:
 
 !!! note
-
-This snippet of code must be inside an async context or inside `main` function
+    This snippet of code must be inside an async context or inside `main` function
 
 ```python
 @pe.update_task
diff --git a/docs/user-guide/basic-acl-workflow.md b/docs/user-guide/basic-acl-workflow.md
index 4ff8d701..91b88146 100644
--- a/docs/user-guide/basic-acl-workflow.md
+++ b/docs/user-guide/basic-acl-workflow.md
@@ -5,15 +5,17 @@ Import ROSE main modules:
 from rose.metrics import MEAN_SQUARED_ERROR_MSE
 from rose.al.active_learner import SequentialActiveLearner
 
+from concurrent.futures import ProcessPoolExecutor
+
 from radical.asyncflow import WorkflowEngine
-from rhapsody.backends import RadicalExecutionBackend
+from rhapsody.backends import ConcurrentExecutionBackend
 ```
 
 
 Define your resource engine, as we described in our previous [Target Resources](target-resources.md) step:
 
 ```python
-engine = await RadicalExecutionBackend({'resource': 'local.localhost'})
+engine = await ConcurrentExecutionBackend(ProcessPoolExecutor())
 asyncflow = await WorkflowEngine.create(engine)
 
 acl = SequentialActiveLearner(asyncflow)
@@ -38,8 +40,8 @@ async def active_learn(*args):
 ```
 
 !!! tip
-ROSE supports defining tasks with python code instead of executables (i.e., python scripts, shell scripts, etc.). To do that, the user have to
-pass the `as_executable=False` argument to the decorator as follows:
+    ROSE supports defining tasks with python code instead of executables (i.e., python scripts, shell scripts, etc.). To do that, the user have to
+    pass the `as_executable=False` argument to the decorator as follows:
 
 ```python
 @acl.simulation_task(as_executable=False)
@@ -92,7 +94,7 @@ async def check_mse(*args):
     return f'python3 check_mse.py'
 ```
 
-!!! Warning
+!!! warning
     For any metric function like `@acl.as_stop_criterion` the invoked script like `check_mse.py` must return a numerical value.
 
 
diff --git a/docs/user-guide/basic-rl-workflow.md b/docs/user-guide/basic-rl-workflow.md
index 1eeb2711..dbee4a4b 100644
--- a/docs/user-guide/basic-rl-workflow.md
+++ b/docs/user-guide/basic-rl-workflow.md
@@ -1,17 +1,22 @@
+# Basic Reinforcement Learning Workflow
+
 ## Define your target machine to run on
 
 Import ROSE main modules:
 ```python
+from concurrent.futures import ProcessPoolExecutor
+
 from radical.asyncflow import WorkflowEngine
-from rhapsody.backends import RadicalExecutionBackend
+from rhapsody.backends import ConcurrentExecutionBackend
 
 from rose.metrics import GREATER_THAN_THRESHOLD
 from rose.rl.reinforcement_learner import SequentialReinforcementLearner
 ```
 
-Define your resource engine, as we described in our previous [Target Resources](target-resources.md) step:
+You can use and setup an HPC engine as we described in our previous [Target Resources](target-resources.md) step:
+
 ```python
-engine = await RadicalExecutionBackend({'resource': 'local.localhost'})
+engine = await ConcurrentExecutionBackend(ProcessPoolExecutor())
 asyncflow = await WorkflowEngine.create(engine)
 
 rl = SequentialReinforcementLearner(asyncflow)
@@ -33,7 +38,7 @@ async def check_reward(*args):
     return 'python3 check_reward.py'
 ```
 
-!!! Warning
+!!! warning
     For any metric function like `@rl.as_stop_criterion` the invoked script like `check_reward.py` must return a numerical value.
 
 Finally invoke the tasks and register them with the reinforcement learner as a workflow:
diff --git a/docs/user-guide/experience.md b/docs/user-guide/experience.md
index 59155772..696e7d02 100644
--- a/docs/user-guide/experience.md
+++ b/docs/user-guide/experience.md
@@ -39,6 +39,10 @@ batch = bank.sample(batch_size=32, replace=True)
 
 ROSE assigns experience banks a unique session ID automatically, or session IDs can be assigned manually. The session ID is used to identify the bank and can be used to save/load the bank to/from disk:
 
+!!! note
+    Automatic session IDs require no bookkeeping on your part — pass `session_id` explicitly
+    only when you need a stable, predictable filename across runs (e.g. resuming a bank by name).
+
 ```python
 # Custom session ID
 bank = ExperienceBank(session_id="rose_session")
diff --git a/docs/user-guide/parallel_learners_docs.md b/docs/user-guide/parallel_learners_docs.md
index b5cfe5f0..e70e1060 100644
--- a/docs/user-guide/parallel_learners_docs.md
+++ b/docs/user-guide/parallel_learners_docs.md
@@ -10,8 +10,7 @@ This tutorial demonstrates how to configure and run multiple learning pipelines
 ---
 
 !!! note
-
-This approach can be applied for both Active and Reinforcement learners (Sequential and Parallel).
+    This approach can be applied for both Active and Reinforcement learners (Sequential and Parallel).
 
 ## Example Overview
 
@@ -79,14 +78,12 @@ from rose import LearnerConfig
 from rose.al import ParallelActiveLearner
 from rose.metrics import MEAN_SQUARED_ERROR_MSE
 
+from concurrent.futures import ProcessPoolExecutor
+
 from radical.asyncflow import WorkflowEngine
-from rhapsody.backends import RadicalExecutionBackend
+from rhapsody.backends import ConcurrentExecutionBackend
 
-engine = await RadicalExecutionBackend(
-    {'runtime': 30,
-     'resource': 'local.localhost'
-     }
-     )
+engine = await ConcurrentExecutionBackend(ProcessPoolExecutor())
 asyncflow = await WorkflowEngine.create(engine)
 acl = ParallelActiveLearner(asyncflow)
 code_path = f'{sys.executable} {os.getcwd()}'
diff --git a/docs/user-guide/rose-aas.md b/docs/user-guide/rose-aas.md
new file mode 100644
index 00000000..3d24f575
--- /dev/null
+++ b/docs/user-guide/rose-aas.md
@@ -0,0 +1,82 @@
+# ROSE as a Service
+
+This page describes ROSE's service model: how the active-learning/RL loop you define stays under your control while the actual simulation and training work executes on HPC, and what that does and does not require from you today.
+
+---
+
+## Core idea: BYOF — Bring Your Own Workflow
+
+ROSE does not own the science. It owns the **orchestration loop**: submit a task, wait for it, decide what runs next, check a stop criterion, repeat — until `max_iter` or the criterion is met.
+
+You bring:
+
+- A `simulate` / `train` / `active_learn` (or `environment` / `update`, or `prediction` / `uncertainty`) implementation, as a plain Python function or a shell command.
+- The environment that implementation needs (packages, data, scripts).
+- The decision of where it should run — a laptop, a login node, or a leadership-class HPC allocation.
+
+ROSE brings the orchestration: dependency-correct task submission, iteration bookkeeping, stop-criterion evaluation, checkpoint-safe state, and tracking integration. The [YAML Spec API](spec-api.md) is the declarative form of this contract — `function: tasks:simulate` is a pointer to *your* code, not a hook into ROSE's.
+
+This is the same boundary that makes ROSE usable across domains without ROSE having to understand any of them: it sees `*args, **kwargs` in, a return value out, nothing about what happened in between.
+
+---
+
+## Two places work can happen
+
+A ROSE learner doesn't run anything itself — it submits tasks through whichever `asyncflow` execution backend you hand it. Two backends matter for the "as a Service" framing:
+
+| Backend | Where the orchestration loop runs | Where tasks execute | Documented at |
+|---|---|---|---|
+| `DragonExecutionBackendV3` (RHAPSODY) | Wherever your script runs — it submits a pilot job and blocks inside it | Inside the pilot job it just submitted | [Target Resources](target-resources.md) |
+| Orbit backend (RHAPSODY, `bridge_url` + `endpoint_name`) | Wherever your script runs — does **not** need to be the HPC machine | Inside a separate, already-running orbit endpoint — possibly on a different machine entirely | this page |
+
+The first model is "submit a job, then run my loop inside it." The second is "run my loop here; dispatch tasks to a job running somewhere else." That second model is what makes ROSE service-like: the loop's control plane (your `learner.start()` call, deciding when to stop, talking to MLflow/ClearML) is decoupled from the compute plane (the HPC allocation actually executing `simulate`/`train`).
+
+---
+
+## How ROSE operates within a job
+
+The orbit backend connects two things over a bridge:
+
+1. **An orbit endpoint**, running inside an HPC job allocation. The job itself is requested through whatever your site normally uses to get an allocation — it doesn't have to be requested by ROSE. Once the job starts, the endpoint comes up inside it and stays alive for the allocation's lifetime, ready to accept task descriptions.
+2. **Your orchestration process**, running anywhere — your laptop, a long-lived service host, a CI runner. It builds the learner from your spec (`LearnerBuilder(cfg, asyncflow).build()`), starts the loop (`learner.start(...)`), and for every `simulation`/`training`/`active_learn`/... task it submits, the asyncflow engine forwards that task description over the bridge to the endpoint, waits for the result, and resumes the loop.
+
+The practical effect: you submit a ROSE workflow to HPC without an interactive session on the cluster, and without your laptop needing to stay connected to the scheduler — only to the bridge. The job allocation is what's expensive and scheduler-queued; your control process is cheap and can be restarted independently of it. Stop-criterion checks, `set_next_config()` decisions, and tracking calls all happen on your side of the bridge, on every iteration, with no per-iteration job resubmission.
+
+This is additive to the model in [Target Resources](target-resources.md), not a replacement — both go through the same `WorkflowEngine`/`LearnerBuilder` plumbing. Which backend you choose only changes *where* the edge lives; the spec, the learner, and the loop semantics are identical either way.
+
+---
+
+## What ROSE automates for you today
+
+- **The loop itself** — iterate, await, check criterion, repeat, with no boilerplate beyond defining your tasks and the threshold.
+- **Preemption-safe state** — every completed iteration is durable before the next starts, so a killed job loses at most the in-flight iteration.
+- **Tracking** — `tracking.backend: mlflow | clearml` in the spec wires a tracker once; every iteration's metrics, params, and lifecycle events are reported automatically, with no tracking code inside your task functions.
+- **Heterogeneous dispatch** — the same loop drives CPU-only, GPU, MPI, or shell-executable tasks, and (via the `learners:` block) distinct task implementations per parallel learner.
+
+## What ROSE does **not** yet automate: data movement
+
+Today, getting a simulation's output into the training task's hands is entirely your task code's responsibility:
+
+- `type: python` slots pass the previous task's return value in-memory, as the first positional argument — this is implicit in the task type, not something you declare.
+- `type: shell` slots pass nothing automatically; your command's stdout becomes the next task's input only if your scripts agree on a file path or convention outside the spec (this is exactly the pattern in the M3DC1 use case, where `simulate`/`train`/`active_learn` hand-roll a namespaced directory of files to communicate).
+
+**Planned:** a `DataExchangeProtocol` field in the YAML spec — `FileBased` or `MemoryBased` — that makes this an explicit, ROSE-managed choice instead of an implicit consequence of task type:
+
+- `MemoryBased` formalizes what `type: python` already does today — in-process object passing between tasks on the same worker.
+- `FileBased` would let ROSE own staging a per-iteration (and per-learner, for parallel runs) working directory, instead of every use case reimplementing its own namespace/path convention by hand inside task code.
+
+This isn't implemented yet — it's the next piece of surface area on the spec, and the natural place to close the "no environment/output contract" gap without ROSE needing to know anything about the science moving through it.
+
+---
+
+## What you bring, concretely
+
+| You supply | ROSE does not check this for you (today) |
+|---|---|
+| Task implementations (`simulate`, `train`, `active_learn`, criterion evaluator, ...) matching the `*args, **kwargs` convention | Whether return shapes match what the next task expects |
+| The runtime environment those implementations need on the worker | No environment/dependency manifest in the spec |
+| Access to wherever the edge agent's job runs (account, queue, allocation) | Out of scope for the spec layer entirely |
+| Your data, and a convention for how tasks find it (today: `parameters:` + your own file paths) | No declared output/data contract yet — see `DataExchangeProtocol` above |
+| Correct `remote.pythonpath` if task modules aren't already importable where the edge runs | `validate_imports=True` only checks importability in *your local* environment |
+
+None of this is unique to the service model — it's the same BYOF boundary as running ROSE locally. What the service model changes is *where* that boundary sits physically: your task code and its dependencies need to be reachable from the edge's job allocation, not from your laptop.
diff --git a/docs/user-guide/spec-api.md b/docs/user-guide/spec-api.md
new file mode 100644
index 00000000..0029ca7c
--- /dev/null
+++ b/docs/user-guide/spec-api.md
@@ -0,0 +1,650 @@
+# YAML Spec API
+
+The YAML spec API lets you declare a ROSE workflow as a data file instead of Python code. The spec is schema-validated when loaded — missing slots, unknown keys, and type errors are caught before any infrastructure starts. Task functions are plain Python functions with no decorators required. The same spec file works with any ROSE learner type: Active Learning, Reinforcement Learning, or UQ.
+
+---
+
+
+!!! note
+    The YAML spec currently supports only the four built-in `learner.type` values listed below:
+
+    - `SequentialActiveLearner`
+    - `ParallelActiveLearner`
+    - `SequentialReinforcementLearner`
+    - `ParallelReinforcementLearner`
+    - `SequentialUQLearner`
+    - `ParallelUQLearner`
+
+    Custom `Learner` subclasses are not yet expressible in YAML — `LearnerBuilder` raises `ValueError` for any other `learner.type` value. Use the Python API (decorator-based, e.g. `SequentialActiveLearner(asyncflow)`) if you need a custom
+    learner implementation.
+
+
+## Loading a Spec
+
+For the common case — build a learner and run it — `LearnerBuilder` loads and validates the YAML itself; you never need to touch a config object:
+
+```python
+from rose.spec.builder import LearnerBuilder
+
+builder = LearnerBuilder("workflow.yaml", asyncflow)  # validates schema on load
+cfg     = builder.config                              # typed WorkflowConfig, if you need it
+```
+
+Reach for `load_spec` instead when you need `WorkflowSpec`-level features: spec variants via [`workflow_with()`](#spec-variants-with-workflow_with), import validation, or the `.workflow` coroutine used by [ROSE as a Service](rose-aas.md):
+
+```python
+from rose.spec import load_spec
+
+spec = load_spec("workflow.yaml")  # validates schema on load
+cfg  = spec.config                 # typed WorkflowConfig object
+```
+
+Both raise `ValueError` with a precise message on any schema violation. Neither imports task modules by default — see [Import Validation](#import-validation) to enable that check.
+
+---
+
+## Resource Specification via `task_description`
+
+Every task — `simulation`, `training`, `active_learn`, `environment`, `update`, … — can
+carry an optional `task_description`. In the Python API this is a parameter on the
+decorated function itself, with a default value:
+
+```python
+@acl.simulation_task(as_executable=False)
+async def simulate(*args, task_description={"process_templates": [(4, {})]}, **kwargs):
+    ...
+```
+
+The accepted keys depend entirely on which execution backend your `asyncflow` session is
+using — see asyncflow's
+[Execution Backends guide](https://radical-cybertools.github.io/radical.asyncflow/exec_backends/?h=task_desc#assign-resources-for-your-application-task)
+for the full per-backend reference.
+
+The YAML spec exposes the same dict as a `task_description:` key on the corresponding task
+slot — see the side-by-side example in [Task Types](#task-types) below.
+
+---
+
+## Sequential Active Learner
+
+<div style="display: grid; grid-template-columns: 1fr 1fr; gap: 1.5rem; overflow: hidden;" markdown>
+<div style="min-width: 0; overflow-x: auto;" markdown>
+
+**Python API**
+
+```python
+from rose.al.active_learner import SequentialActiveLearner
+
+acl = SequentialActiveLearner(asyncflow)
+
+@acl.simulation_task(as_executable=False)
+async def simulate(*args, **kwargs):
+    ...  # return simulation result
+
+@acl.training_task(as_executable=False)
+async def train(sim_result, **kwargs):
+    ...  # return trained model
+
+@acl.active_learn_task(as_executable=False)
+async def active_learn(sim_result, model, **kwargs):
+    ...  # return updated dataset
+
+@acl.as_stop_criterion(
+    metric_name="mse",
+    threshold=0.01,
+    operator="<",
+)
+async def check_mse(*args, **kwargs):
+    ...  # return float metric
+
+async for state in acl.start(max_iter=5):
+    print(f"iter {state.iteration}  "
+          f"mse={state.metric_value:.4f}")
+
+await asyncflow.shutdown()
+```
+
+</div>
+<div style="min-width: 0; overflow-x: auto;" markdown>
+
+**YAML Spec**
+
+```yaml
+learner:
+  type: sequential_active_learner
+  max_iter: 5
+
+simulation:
+  type: python
+  function: tasks:simulate
+
+training:
+  type: python
+  function: tasks:train
+
+active_learn:
+  type: python
+  function: tasks:active_learn
+
+stop_criterion:
+  metric: mse
+  threshold: 0.01
+  operator: "<"
+  evaluator:
+    type: python
+    function: tasks:check_mse
+```
+
+```python
+# run it
+from rose.spec.builder import LearnerBuilder
+
+builder = LearnerBuilder("workflow.yaml", asyncflow)
+cfg     = builder.config
+learner = builder.build()
+
+async for state in learner.start(
+    max_iter=cfg.learner.max_iter
+):
+    print(f"iter {state.iteration}  "
+          f"mse={state.metric_value:.4f}")
+```
+
+</div>
+</div>
+
+Task functions referenced by `function: tasks:simulate` are ordinary Python callables in a `tasks.py` module — no ROSE decorators required:
+
+```python
+# tasks.py
+def simulate(*args, **kwargs):
+    ...  # return simulation result
+
+def train(sim_result, **kwargs):
+    ...  # return trained model
+
+def active_learn(sim_result, model, **kwargs):
+    ...  # return updated dataset
+
+def check_mse(*args, **kwargs):
+    ...  # return float metric value
+```
+
+---
+
+## Parallel Learner — Shared Tasks
+
+When all parallel learners run the same task implementations, declare the task slots at the top level exactly as in the sequential case.
+
+<div style="display: grid; grid-template-columns: 1fr 1fr; gap: 1.5rem; overflow: hidden;" markdown>
+<div style="min-width: 0; overflow-x: auto;" markdown>
+
+**Python API**
+
+```python
+from rose.al.active_learner import ParallelActiveLearner
+
+acl = ParallelActiveLearner(asyncflow)
+
+@acl.simulation_task(as_executable=False)
+async def simulate(*args, **kwargs):
+    ...
+
+@acl.training_task(as_executable=False)
+async def train(sim_result, **kwargs):
+    ...
+
+@acl.active_learn_task(as_executable=False)
+async def active_learn(sim_result, model, **kwargs):
+    ...
+
+@acl.as_stop_criterion(
+    metric_name="mse",
+    threshold=0.01,
+)
+async def check_mse(*args, **kwargs):
+    ...
+
+async for state in acl.start(
+    parallel_learners=2,
+    max_iter=5,
+):
+    print(f"learner {state.learner_id}  "
+          f"iter {state.iteration}")
+```
+
+</div>
+<div style="min-width: 0; overflow-x: auto;" markdown>
+
+**YAML Spec**
+
+```yaml
+learner:
+  type: parallel_active_learner
+  max_iter: 5
+  parallel_learners: 2
+
+simulation:
+  type: python
+  function: tasks:simulate
+
+training:
+  type: python
+  function: tasks:train
+
+active_learn:
+  type: python
+  function: tasks:active_learn
+
+stop_criterion:
+  metric: mse
+  threshold: 0.01
+  operator: "<"
+  evaluator:
+    type: python
+    function: tasks:check_mse
+```
+
+```python
+# run it
+builder = LearnerBuilder("workflow.yaml", asyncflow)
+cfg     = builder.config
+learner = builder.build()
+
+async for state in learner.start(
+    max_iter=cfg.learner.max_iter,
+    parallel_learners=cfg.learner.parallel_learners,
+):
+    print(f"learner {state.learner_id}  "
+          f"iter {state.iteration}")
+```
+
+</div>
+</div>
+
+Each learner receives a unique `learner_id` kwarg (0, 1, …) in its task calls. Task functions can use it to write to isolated files or namespaces.
+
+---
+
+## Parallel Learner — Distinct Tasks (`learners:` block)
+
+When each parallel learner needs its own task implementations — different models, different executables, different command-line flags — use the `learners:` block. Each entry gets a `label` that is injected as `learner_label` into every task kwarg.
+
+<div style="display: grid; grid-template-columns: 1fr 1fr; gap: 1.5rem; overflow: hidden;" markdown>
+<div style="min-width: 0; overflow-x: auto;" markdown>
+
+**Python API**
+
+```python
+acl = ParallelActiveLearner(asyncflow)
+
+# Manual routing by learner_id
+TRAIN_CMD = {
+    0: "python train.py --model linear",
+    1: "python train.py --model ridge",
+}
+
+@acl.simulation_task
+async def simulate(*args, **kwargs):
+    return "python sim.py"
+
+@acl.training_task
+async def train(*args, **kwargs):
+    return TRAIN_CMD[kwargs["learner_id"]]
+
+@acl.active_learn_task
+async def active_learn(*args, **kwargs):
+    return "python active_learn.py"
+
+@acl.as_stop_criterion(
+    metric_name="mse", threshold=0.01
+)
+async def check_mse(*args, **kwargs):
+    ...
+
+# Build per-learner configs manually
+from rose.learner import LearnerConfig, TaskConfig
+lcs = [
+    LearnerConfig(
+        simulation={i: TaskConfig(kwargs={"learner_id": i})
+                    for i in range(6)},
+        ...
+    )
+    for lid in range(2)
+]
+
+async for state in acl.start(
+    parallel_learners=2,
+    max_iter=5,
+    learner_configs=lcs,
+):
+    ...
+```
+
+</div>
+<div style="min-width: 0; overflow-x: auto;" markdown>
+
+**YAML Spec**
+
+```yaml
+learner:
+  type: parallel_active_learner
+  max_iter: 5
+
+learners:
+  - label: linear_regression
+    simulation:
+      type: shell
+      command: python sim.py --label a
+    training:
+      type: shell
+      command: python train.py --label a --model linear
+    active_learn:
+      type: shell
+      command: python active_learn.py --label a
+
+  - label: ridge_regression
+    simulation:
+      type: shell
+      command: python sim.py --label b
+    training:
+      type: shell
+      command: python train.py --label b --model ridge
+    active_learn:
+      type: shell
+      command: python active_learn.py --label b
+
+stop_criterion:
+  metric: mse
+  threshold: 0.01
+  operator: "<"
+  evaluator:
+    type: python
+    function: tasks:check_mse
+```
+
+```python
+# run it — builder handles routing + LearnerConfig
+builder = LearnerBuilder("workflow.yaml", asyncflow)
+cfg     = builder.config
+learner = builder.build()
+lcs     = builder.build_learner_configs()
+
+async for state in learner.start(
+    max_iter=cfg.learner.max_iter,
+    parallel_learners=len(lcs),
+    learner_configs=lcs,
+):
+    label = cfg.learners[state.learner_id].label
+    print(f"[{label}]  iter {state.iteration}")
+```
+
+</div>
+</div>
+
+The builder creates a single dispatch closure per slot that routes to the correct command or function based on `learner_id`. The `learner_id` kwarg is consumed by the routing layer and never reaches the user function.
+
+!!! note
+    All learners in a `learners:` block must use the same task type (`python` or `shell`) for each slot. Mixed types within a slot are rejected at load time.
+
+---
+
+## Task Types
+
+<div style="display: grid; grid-template-columns: 1fr 1fr; gap: 1.5rem; overflow: hidden;" markdown>
+<div style="min-width: 0; overflow-x: auto;" markdown>
+
+**`type: python`**
+
+Calls a Python function directly on the worker. Data flows in-memory between tasks. Function must be importable via `module:callable` syntax.
+
+```yaml
+simulation:
+  type: python
+  function: my_package.tasks:simulate
+```
+
+```python
+# tasks.py
+def simulate(*args, **kwargs):
+    # receives positional results from
+    # prior tasks as *args
+    return {"X": ..., "y": ...}
+```
+
+The function must be async. The return value is passed as the first positional argument to the next task in the chain.
+
+</div>
+<div style="min-width: 0; overflow-x: auto;" markdown>
+
+**`type: shell`**
+
+Runs in a single or multi-process on the worker. Data flows through files on disk. The command string is the return value — ROSE submits it as an executable.
+
+```yaml
+simulation:
+  type: shell
+  command: python sim.py
+```
+
+With `parameters:` placeholders:
+
+```yaml
+simulation:
+  type: shell
+  command: python sim.py --dataset {dataset}
+```
+
+`{dataset}` is filled from `parameters.dataset` at runtime via `str.format_map(kwargs)`.
+
+The criterion evaluator must print a single float to stdout — ROSE captures stdout as the metric value.
+
+</div>
+</div>
+
+### `task_description` example
+
+<div style="display: grid; grid-template-columns: 1fr 1fr; gap: 1.5rem; overflow: hidden;" markdown>
+<div style="min-width: 0; overflow-x: auto;" markdown>
+
+**Python API**
+
+```python
+@acl.simulation_task(as_executable=False)
+async def simulate(*args, task_description={"process_templates": [(4, {})]}, **kwargs):
+    ...
+```
+
+</div>
+<div style="min-width: 0; overflow-x: auto;" markdown>
+
+**YAML Spec**
+
+```yaml
+simulation:
+  type: python
+  function: tasks:simulate
+  task_description:
+    process_templates:
+      - [4, {}]
+```
+
+</div>
+</div>
+
+!!! note
+    For parallel learners using the `learners:` block, all entries must use the same `task_description` for each slot — the backend registers it once at task-registration time.
+
+---
+
+## Stop Criterion
+
+<div style="display: grid; grid-template-columns: 1fr 1fr; gap: 1.5rem; overflow: hidden;" markdown>
+<div style="min-width: 0; overflow-x: auto;" markdown>
+
+**Python API**
+
+```python
+@acl.as_stop_criterion(
+    metric_name="mse",
+    threshold=0.01,
+    operator="<",
+    as_executable=False,
+)
+async def check_mse(*args, **kwargs):
+    ...  # return float
+```
+
+</div>
+<div style="min-width: 0; overflow-x: auto;" markdown>
+
+**YAML Spec**
+
+```yaml
+stop_criterion:
+  metric: mse
+  threshold: 0.01
+  operator: "<"        # default; also: >, ==, <=, >=
+  evaluator:
+    type: python
+    function: tasks:check_mse
+```
+
+</div>
+</div>
+
+The evaluator function follows the same `*args, **kwargs` convention as task functions. ROSE stops the loop when `metric_value <operator> threshold` is satisfied, or when `max_iter` is reached — whichever comes first.
+
+---
+
+## Parameters
+
+The `parameters:` block defines key-value pairs that are injected into every task's `**kwargs` at every iteration:
+
+```yaml
+parameters:
+  dataset: my_dataset
+  batch_size: 32
+  growing_pool: false
+```
+
+```python
+def simulate(*args, **kwargs):
+    dataset      = kwargs["dataset"]      # "my_dataset"
+    batch_size   = kwargs["batch_size"]   # 32
+    growing_pool = kwargs["growing_pool"] # False
+    iteration    = kwargs["iteration"]    # 0, 1, 2, ...
+    ...
+```
+
+### Reserved keys
+
+The following keys are injected automatically by the builder and must not appear in `parameters:`:
+
+| Key | Available in | Description |
+|-----|-------------|-------------|
+| `iteration` | all tasks | Current loop counter (0-based) |
+| `pythonpath` | all tasks | Contents of `remote.pythonpath` as a list |
+| `learner_id` | parallel tasks | Integer index of the learner (0, 1, …) |
+| `learner_label` | parallel tasks (when `label` set) | Human-readable learner name from `learners[].label` |
+
+---
+
+## Remote Config
+
+```yaml
+remote:
+  pythonpath:
+    - /path/to/my/task/modules
+    - /path/to/shared/utilities
+```
+
+`remote.pythonpath` entries are added to `sys.path` on the remote worker before any task module is imported. They are also injected into every task's `kwargs["pythonpath"]` (as a list), so task functions can construct file paths without duplicating the value in `parameters:`:
+
+```python
+def train(sim_result, **kwargs):
+    pythonpath = kwargs.get("pythonpath", [])
+    base = pythonpath[0] if pythonpath else ""
+    script = Path(base) / "scripts" / "train_model.py"
+    ...
+```
+
+`remote.pythonpath` is the single edit point for the remote worker path — changing it updates both the import path and the kwarg.
+
+---
+
+## Tracking
+
+```yaml
+tracking:
+  backend: mlflow      # mlflow | clearml | none (default)
+  experiment: my-exp   # experiment/project name
+  run_name: run-01     # optional run label
+```
+
+See [MLflow integration](../integrations/mlflow.md) and [ClearML integration](../integrations/clearml.md) for configuration details.
+
+---
+
+## Spec Variants with `workflow_with()`
+
+`workflow_with()` returns a new `WorkflowSpec` with selective overrides applied. The original spec is never mutated:
+
+```python
+base_spec  = load_spec("workflow.yaml")
+test_spec  = base_spec.workflow_with(max_iter=2, parameters={"dataset": "test_ds"})
+large_spec = base_spec.workflow_with(max_iter=50, parameters={"batch_size": 128})
+```
+
+Accepted override keys:
+
+- `parameters` — merged (not replaced) into the existing `parameters:` block
+- Any `learner` field: `max_iter`, `parallel_learners`
+- Any other top-level spec field
+
+Unknown keys raise `ValueError` immediately.
+
+---
+
+## Import Validation
+
+By default, `load_spec` does not import task modules — this is intentional when `remote.pythonpath` points to paths that only exist on the remote worker. To catch `function:` typos locally during development:
+
+```python
+spec = load_spec("workflow.yaml", validate_imports=True)
+```
+
+With `validate_imports=True`, every `module:callable` string is resolved in the current environment. All failures are collected and reported in a single `ValueError` before any infrastructure starts. Use this during development when task files are locally accessible.
+
+---
+
+## Spec Reference
+
+### Top-level keys
+
+| Key | Type | Required | Default | Description |
+|-----|------|----------|---------|-------------|
+| `learner` | object | yes | — | Learner type and loop settings |
+| `learner.type` | string | yes | — | `sequential_active_learner`, `parallel_active_learner`, `sequential_reinforcement_learner`, `uq_active_learner` |
+| `learner.max_iter` | int | no | `0` | Maximum iterations |
+| `learner.parallel_learners` | int | no | `2` | Parallel learner count when `learners:` is absent |
+| `simulation` / `training` / `active_learn` | TaskDef | AL yes | — | Task slots for Active Learning |
+| `environment` / `update` | TaskDef | RL yes | — | Task slots for Reinforcement Learning |
+| `prediction` / `active_learn` / `uncertainty` | TaskDef | UQ yes | — | Task slots for UQ-based AL |
+| `learners` | list | no | — | Per-learner task definitions for heterogeneous parallel |
+| `stop_criterion` | object | yes | — | Stopping condition |
+| `parameters` | dict | no | `{}` | User-defined kwargs injected into all tasks |
+| `remote.pythonpath` | list[str] | no | `[]` | Paths added to `sys.path` on worker; injected as `pythonpath` kwarg |
+| `tracking.backend` | string | no | `none` | `mlflow` / `clearml` / `none` |
+| `tracking.experiment` | string | no | `ROSE-Spec` | Experiment name |
+| `tracking.run_name` | string | no | `null` | Run label |
+
+
+### TaskDef fields
+
+| Key | Type | Required | Description |
+|-----|------|----------|-------------|
+| `type` | `python` \| `shell` | yes | Task execution mode |
+| `function` | string | if `type: python` | `module:callable` — dotted module path and function name |
+| `command` | string | if `type: shell` | Shell command; supports `{param}` placeholders filled from `parameters:` |
+| `task_description` | dict | no | Resource hints forwarded to the execution backend |
diff --git a/docs/user-guide/target-resources.md b/docs/user-guide/target-resources.md
index eb6add8c..7f4c523e 100644
--- a/docs/user-guide/target-resources.md
+++ b/docs/user-guide/target-resources.md
@@ -1,19 +1,19 @@
 # Target Machines for Executing AL Workflows
-ROSE enables the orchestration of ML Surrogate building workflows on diverse computing resources using [radical.asyncflow](https://github.com/radical-cybertools/radical.asyncflow). Below, we will show how you can specify your `local computer` and `remote HPC machine` as target resources using the `RadicalExecutionBackend` from [RHAPSODY](https://github.com/radical-cybertools/rhapsody).
+ROSE enables the orchestration of ML Surrogate building workflows on diverse computing resources using [radical.asyncflow](https://github.com/radical-cybertools/radical.asyncflow). Below, we will show how you can specify your `local computer` and `remote HPC machine` as target resources using the `DragonExecutionBackendV3` from [RHAPSODY](https://radical-cybertools.github.io/rhapsody/getting-started/advanced-usage/?h=hpc#multiple-execution-backends).
 
 ## Local Computer
 For local execution, user can use their desktops, laptops, and their own small clusters to execute their AL workflows as follows:
 ```python
 import os
 
+from concurrent.futures import ProcessPoolExecutor
+
 from radical.asyncflow import WorkflowEngine
-from rhapsody.backends import RadicalExecutionBackend
+from rhapsody.backends import ConcurrentExecutionBackend
 
 from rose.al.active_learner import SequentialActiveLearner
 
-engine = await RadicalExecutionBackend(
-    {'runtime': 30,
-    'resource': 'local.localhost'})
+engine = await ConcurrentExecutionBackend(ProcessPoolExecutor())
 
 asyncflow = await WorkflowEngine.create(engine)
 
@@ -21,20 +21,24 @@ acl = SequentialActiveLearner(asyncflow)
 ```
 
 ## HPC Resources
-To execute AL workflows on HPC machines, users must have an active allocation on the target machine and specify their resource requirements, as well as the time needed to execute their workflows. Remember, ROSE uses `RadicalExecutionBackend` from [RHAPSODY](https://github.com/radical-cybertools/rhapsody) (`rhapsody-py`) which is an interface for RADICAL-Pilot runtime system. For more information on how to access, set up, and execute workflows on HPC machines, refer to the following link [RADICAL-Pilot Job Submission](https://radicalpilot.readthedocs.io/en/stable/tutorials/submission.html):
+
+To execute AL workflows on HPC machines, users must have an active allocation on the target machine and specify their resource requirements to execute their workflows. Remember, ROSE uses `DragonExecutionBackendV3` from [RHAPSODY](https://github.com/radical-cybertools/rhapsody) (`rhapsody-py`) which is an interface for multip execution and AI runtime system. For more information on how to access, set up, and execute workflows on HPC machines, refer to the following link [ROSE with RHAPSODY on HPC](https://radical-cybertools.github.io/rhapsody/getting-started/advanced-usage/?h=hpc#hpc-workloads-with-dragon):
+
+
+!!! note
+    For any ROSE script that uses `DragonExecutionBackendV3`, user must run the
+    rose_script.py with `dragon` binary instead of `python`. Please refer to the following link for more information: [use ROSE with RHAPSODY-Dragon on HPC](https://radical-cybertools.github.io/rhapsody/hpc-machines/#backend-compatibility)
 
 ```python
 import os
 
 from radical.asyncflow import WorkflowEngine
-from rhapsody.backends import RadicalExecutionBackend
+from rhapsody.backends import DragonExecutionBackendV3
 
 from rose.al.active_learner import SequentialActiveLearner
 
 
-hpc_engine = await RadicalExecutionBackend(
-    {'runtime': 30, 'cores': 4096,
-     'gpus' : 4, 'resource': 'tacc.frontera'})
+hpc_engine = await DragonExecutionBackendV3()
 
 asyncflow = await WorkflowEngine.create(hpc_engine)
 
diff --git a/docs/user-guide/uq_based-acl-workflow.md b/docs/user-guide/uq_based-acl-workflow.md
index 24cfc90f..449bbce4 100644
--- a/docs/user-guide/uq_based-acl-workflow.md
+++ b/docs/user-guide/uq_based-acl-workflow.md
@@ -1,3 +1,5 @@
+# UQ-Based Active Learning Workflow
+
 To accelerate the development of UQ-driven active learning methods, ROSE provides a flexible approach for composing and executing complex AL workflows.
 
 In this example, we illustrate how to define a UQ–AL workflow that supports multiple levels of parallelism.
@@ -10,25 +12,23 @@ This introduces **two levels of parallelism**:
 * **Workflow-level parallelism** for executing multiple UQ–AL loops side by side.
 
 Both levels can be naturally expressed and efficiently executed using ROSE’s **custom AL policy**, enabling scalable and adaptive uncertainty-aware learning.
-```sh
-                             (N AL WFs in Parallel)
-          +-------------------+               +-------------------+
-          |      UQ WF 1      |               |      UQ WF 2      |
-          +-------------------+               +-------------------+
-                   │                                    │
-  +----------------+-----------------+  +----------------+-----------------+
-  |       (N tasks Parallel)         |  |       (N AL tasks Parallel)      |
-  +---------------+  +---------------+  +---------------+  +---------------+
-  | Simulation 1,2,..............n   |  | Simulation 1,2,..............n   |
-  +---------------+  +---------------+  +---------------+  +---------------+
-          |                |                    |                 |
-  +---------------+  +---------------+  +---------------+  +---------------+
-  |  Train Model  1,2,...........m   |  |  Train Model  1,2,...........m   |
-  +---------------+  +---------------+  +---------------+  +---------------+
-          |                |                    |                 |
-    +-----------------------------+        +-----------------------------+
-    |      AL based on UQ WF 1    |        |    AL based on UQ WF 1      |
-    +-----------------------------+        +-----------------------------+
+```mermaid
+graph TD
+    N["N AL WFs in Parallel"]
+    N --> WF1["UQ WF 1"]
+    N --> WF2["UQ WF 2"]
+
+    subgraph G1[" "]
+        WF1 --> S1["Simulation 1, 2, ..., n (parallel)"]
+        S1 --> T1["Train Model 1, 2, ..., m (parallel)"]
+        T1 --> A1["AL based on UQ WF 1"]
+    end
+
+    subgraph G2[" "]
+        WF2 --> S2["Simulation 1, 2, ..., n (parallel)"]
+        S2 --> T2["Train Model 1, 2, ..., m (parallel)"]
+        T2 --> A2["AL based on UQ WF 2"]
+    end
 ```
 
 
@@ -56,7 +56,7 @@ To support this approach, two new task types have been introduced:
 code_path = f'{sys.executable} {os.getcwd()}'
 
 # Define and register the prediction task
-@learner.prediction_task()
+@uql.prediction_task()
 async def prediction(*args):
     return f'{code_path}/predict.py'
 
@@ -71,11 +71,11 @@ async def prediction(*args):
 ```python
 
 # Defining the uncertainty quantification with a metric (PREDICTIVE_ENTROPY in this case)
-@learner.uncertainty_quantification(uq_metric_name=PREDICTIVE_ENTROPY,
+@uql.uncertainty_quantification(uq_metric_name=PREDICTIVE_ENTROPY,
                                     threshold=1.0,
                                     query_size=10)
 async def check_uq(*args):
-    return f'{code_path}/check_uq.py'xs
+    return f'{code_path}/check_uq.py'
 
 ```
 
@@ -95,7 +95,7 @@ Import and Initialize the UQ Learner
 
 ```python
 from rose.uq.uq_active_learner import ParallelUQLearner
-learner = ParallelUQLearner(asyncflow)
+uql = ParallelUQLearner(asyncflow)
 ```
 
 Run the learning (Active Learning Loop)
@@ -103,7 +103,7 @@ Run the learning (Active Learning Loop)
 ```python
 PIPELINES = ['UQ_learner1', 'UQ_learner2']
 
-results = await learner.start(
+results = await uql.start(
     learner_names=PIPELINES,
     model_names=MODELS,
     learner_configs=learner_configs,
@@ -122,7 +122,7 @@ print(results)
 with open(Path(os.getcwd(), 'UQ_training_results.json'), 'w') as f:
     json.dump(results, f, indent=4)
 
-await learner.shutdown()
+await uql.shutdown()
 ```
 
 #### Defining costom UQ metric
diff --git a/docs/user-guide/visualization.md b/docs/user-guide/visualization.md
index e69de29b..881189e6 100644
--- a/docs/user-guide/visualization.md
+++ b/docs/user-guide/visualization.md
@@ -0,0 +1,72 @@
+# Visualization
+
+ROSE does not ship a plotting module — visualizing a run's progress is left to whichever
+tool already fits your workflow: raw `IterationState` fields, the native file tracker, or
+an MLflow/ClearML dashboard if `tracking.backend` is already wired up.
+
+!!! note
+    There is no `rose.plot` or similar utility. The three approaches below all read data
+    ROSE already produces — none requires changes to your task code.
+
+---
+
+## Real-time, in-loop
+
+Every iteration yields an `IterationState` with `iteration`, `metric_value`, and
+`metric_history` (the full list of past metric values so far):
+
+```python
+import matplotlib.pyplot as plt
+
+async for state in learner.start(max_iter=20):
+    print(f"[iter {state.iteration}] mse={state.metric_value:.4f}")
+
+plt.plot(state.metric_history)
+plt.xlabel("iteration")
+plt.ylabel("mse")
+plt.show()
+```
+
+This is the fastest path when you just want a quick look right after a local run.
+
+---
+
+## Native file tracker
+
+For HPC runs where you want a durable, preemption-safe record, attach `HPC_FileTracker`
+(see `examples/integrations/tracking/basic.py`) — it appends one JSON line per iteration:
+
+```python
+learner.add_tracker(HPC_FileTracker("run.jsonl"))
+```
+
+Replay and plot after the run, or from a different machine entirely:
+
+```python
+import pandas as pd
+
+df = pd.read_json("run.jsonl", lines=True)
+iterations = df[df.event == "iteration"]
+iterations.plot(x="iteration", y="mse", logy=True)
+```
+
+Any numeric value your tasks return as a `dict` (e.g. `n_labeled`, `train_mse`) is captured
+automatically and available as its own column.
+
+---
+
+## MLflow / ClearML dashboards
+
+If `tracking.backend: mlflow` or `tracking.backend: clearml` is set in your spec (or you
+attach `MLflowTracker`/`ClearMLTracker` directly in the Python API), every iteration's
+metrics are already logged with no extra code. The respective web UIs are the recommended
+way to compare runs and overlay parallel-learner series:
+
+- [MLflow integration](../integrations/mlflow.md) — `mlflow ui --port 5000`, metric curves
+  per run, run comparison.
+- [ClearML integration](../integrations/clearml.md) — **Scalars** tab, parallel learners
+  shown as separate series under the same title.
+
+!!! tip
+    Prefer the dashboards over building your own plots once you have more than a couple of
+    runs to compare — both tools already handle multi-run overlay and filtering.
diff --git a/mkdocs.yml b/mkdocs.yml
index c35fd6a4..5efc9a7c 100644
--- a/mkdocs.yml
+++ b/mkdocs.yml
@@ -6,10 +6,6 @@ edit_uri: ""
 
 theme:
   name: material
-  features:
-    - navigation.tabs
-    - navigation.sections
-    - toc.integrate
   palette:
       - media: "(prefers-color-scheme: light)"
         scheme: default
@@ -17,14 +13,14 @@ theme:
           icon: material/weather-night
           name: Switch to dark mode
         primary: black
-        accent: indigo
+        accent: pink
       - media: "(prefers-color-scheme: dark)"
         scheme: slate
         toggle:
           icon: material/weather-sunny
           name: Switch to light mode
         primary: black
-        accent: indigo
+        accent: pink
 
   highlightjs: true
   hljs_languages:
@@ -37,6 +33,7 @@ theme:
       - navigation.top
       - navigation.tabs
       - navigation.tabs.sticky
+      - toc.integrate
       - search.highlight
       - search.share
       - search.suggest
@@ -66,7 +63,11 @@ markdown_extensions:
   - admonition
   - codehilite:
       linenums: true
-  - pymdownx.superfences
+  - pymdownx.superfences:
+      custom_fences:
+        - name: mermaid
+          class: mermaid
+          format: !!python/name:pymdownx.superfences.fence_code_format
   - pymdownx.details
   - pymdownx.tasklist:
       custom_checkbox: true
@@ -75,6 +76,8 @@ markdown_extensions:
       emoji_generator: !!python/name:material.extensions.emoji.to_svg
   - pymdownx.snippets:
       check_paths: true
+  - pymdownx.tabbed:
+      alternate_style: true
   - attr_list
   - md_in_html
 
@@ -86,15 +89,20 @@ nav:
       - 3.FAQ: getting-started/faq.md
   - User Guide:
       - 1.Target Resources: user-guide/target-resources.md
-      - 2.Basic AL workflow: user-guide/basic-acl-workflow.md
-      - 3.AL Metric: user-guide/acl-metrics.md
-      - 4.Building Parallel Learners: user-guide/parallel_learners_docs.md
-      - 5.Advanced AL workflow: user-guide/advanced-acl-workflow.md
-      - 6.Visualization: user-guide/visualization.md
-      - 7.Reinforcement Learning: user-guide/basic-rl-workflow.md
-      - 8.Experience Banks: user-guide/experience.md
-      - 9.Advanced RL workflow: user-guide/advanced-rl-workflow.md
-      - 10.Tracking & Observability: user-guide/tracking.md
+      - 2.Active Learners:
+          - 1.Sequential Active Learning Example: user-guide/basic-acl-workflow.md
+          - 2.Active Learning Metrics: user-guide/acl-metrics.md
+          - 3.Parallel Active Learning Example: user-guide/parallel_learners_docs.md
+          - 4.Custom Active Learning Example: user-guide/advanced-acl-workflow.md
+          - 5.UQ-Based Active Learning Example: user-guide/uq_based-acl-workflow.md
+      - 3.Reinforcement Learners:
+          - 1.Sequential Reinforcement Learning Example: user-guide/basic-rl-workflow.md
+          - 2.Experience Banks: user-guide/experience.md
+          - 3.Parallel Reinforcement Learning Example: user-guide/advanced-rl-workflow.md
+      - 4.Visualization: user-guide/visualization.md
+      - 5.Tracking & Observability: user-guide/tracking.md
+      - 6.YAML Spec API: user-guide/spec-api.md
+      - 7.ROSE as a Service: user-guide/rose-aas.md
   - Integrations:
       - MLflow: integrations/mlflow.md
       - ClearML: integrations/clearml.md
diff --git a/pyproject.toml b/pyproject.toml
index e8fcca05..f57c4226 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -36,6 +36,8 @@ mlflow = ["mlflow>=2.0"]
 clearml = ["clearml>=1.14"]
 tracking = ["mlflow>=2.0", "clearml>=1.14"]
 
+spec = ["pyyaml>=6.0"]
+
 # All test deps
 dev = [
     "pytest",
diff --git a/rose/__init__.py b/rose/__init__.py
index 0038267b..db05a14b 100644
--- a/rose/__init__.py
+++ b/rose/__init__.py
@@ -2,6 +2,7 @@
 from rose.learner import IterationState, Learner, LearnerConfig, TaskConfig
 from rose.metrics import *  # noqa: F403
 from rose.rl import reinforcement_learner
+from rose.spec import WorkflowSpec, load_spec
 from rose.tracking import PipelineManifest, TrackerBase
 from rose.uq import uq_active_learner, uq_learner, uq_scorer
 
@@ -21,4 +22,7 @@
     # Tracking
     "TrackerBase",
     "PipelineManifest",
+    # YAML spec layer
+    "load_spec",
+    "WorkflowSpec",
 ]
diff --git a/rose/spec/__init__.py b/rose/spec/__init__.py
new file mode 100644
index 00000000..75a5ee72
--- /dev/null
+++ b/rose/spec/__init__.py
@@ -0,0 +1,149 @@
+from __future__ import annotations
+
+from collections.abc import Callable, Coroutine
+from pathlib import Path
+from typing import Any
+
+from .builder import LearnerBuilder
+from .schema import WorkflowConfig
+
+__all__ = ["load_spec", "WorkflowSpec"]
+
+
+def load_spec(path: str | Path, validate_imports: bool = False) -> WorkflowSpec:
+    """Load and validate a YAML workflow spec. Raises ValueError on schema errors.
+
+    Args:
+        path: Path to the YAML file.
+        validate_imports: If True, verify that every python task's ``module:callable``
+            is importable from the current environment (sys.path extended with
+            ``remote.pythonpath``). Use during development when task files are
+            locally accessible. Leave False (default) when ``remote.pythonpath``
+            points to paths that only exist on the remote worker.
+    """
+    spec = WorkflowSpec(WorkflowConfig.from_yaml(path))
+    if validate_imports:
+        _validate_task_imports(spec.config)
+    return spec
+
+
+def _collect_python_specs(cfg: WorkflowConfig) -> list[str]:
+    """Return deduplicated 'module:callable' strings for all Python tasks in the spec."""
+    specs: list[str] = []
+    if cfg.learners is not None:
+        for ld in cfg.learners:
+            for td in ld.tasks.values():
+                if td.type == "python":
+                    specs.append(td.function)
+    else:
+        for td in cfg.tasks.values():
+            if td.type == "python":
+                specs.append(td.function)
+    if cfg.stop_criterion.evaluator.type == "python":
+        specs.append(cfg.stop_criterion.evaluator.function)
+    return list(dict.fromkeys(specs))  # deduplicate, preserve order
+
+
+def _validate_task_imports(cfg: WorkflowConfig) -> None:
+    """Try to import every Python task callable.
+
+    Raises ValueError listing all failures.
+    """
+    import importlib
+    import sys as _sys
+
+    added = [p for p in cfg.remote.pythonpath if p not in _sys.path]
+    for p in added:
+        _sys.path.insert(0, p)
+    try:
+        failures: list[str] = []
+        for spec_str in _collect_python_specs(cfg):
+            mod_path, fn_name = spec_str.rsplit(":", 1)
+            try:
+                mod = importlib.import_module(mod_path)
+                if not hasattr(mod, fn_name):
+                    failures.append(f"  {spec_str!r}: module has no attribute '{fn_name}'")
+            except ImportError as exc:
+                failures.append(f"  {spec_str!r}: {exc}")
+        if failures:
+            raise ValueError(
+                "validate_imports failed — the following function specs are not importable:\n"
+                + "\n".join(failures)
+            )
+    finally:
+        for p in added:
+            if p in _sys.path:
+                _sys.path.remove(p)
+
+
+class WorkflowSpec:
+    """Validated workflow spec that produces a coroutine compatible with service_utils.run()."""
+
+    def __init__(self, config: WorkflowConfig) -> None:
+        self.config = config
+
+    def workflow_with(self, **overrides: Any) -> WorkflowSpec:
+        """Return a new WorkflowSpec with selective overrides applied.
+
+        Accepted keys:
+        - ``parameters``: dict merged (not replaced) into the existing parameters block
+        - Any ``LearnerSpec`` field: ``max_iter``, ``parallel_learners``
+        - Any other top-level ``WorkflowConfig`` field
+
+        Example::
+
+            spec.workflow_with(max_iter=3, parameters={"dataset": "test_ds"})
+        """
+        data = self.config.model_dump()
+        learner_fields = set(data["learner"].keys())
+        for key, value in overrides.items():
+            if key == "parameters" and isinstance(value, dict):
+                data["parameters"] = {**data["parameters"], **value}
+            elif key in learner_fields:
+                data["learner"][key] = value
+            elif key in data:
+                data[key] = value
+            else:
+                raise ValueError(f"workflow_with: unknown spec field '{key}'")
+        return WorkflowSpec(WorkflowConfig.model_validate(data))
+
+    @property
+    def workflow(self) -> Callable[..., Coroutine[Any, Any, None]]:
+        cfg = self.config
+
+        async def _workflow(bridge_url: str, endpoint_name: str) -> None:
+            import rhapsody
+            from radical.asyncflow import WorkflowEngine
+
+            engine = await rhapsody.get_backend("orbit", bridge_url=bridge_url, endpoint_name=endpoint_name)
+            asyncflow = await WorkflowEngine.create(engine)
+
+            builder = LearnerBuilder(cfg, asyncflow)
+            learner = builder.build()
+
+            start_kwargs: dict[str, Any] = {"max_iter": cfg.learner.max_iter}
+            if cfg.learner.type == "parallel_active_learner":
+                lcs = builder.build_learner_configs()
+                if lcs is not None:
+                    start_kwargs["parallel_learners"] = len(lcs)
+                    start_kwargs["learner_configs"] = lcs
+                else:
+                    start_kwargs["parallel_learners"] = cfg.learner.parallel_learners
+            else:
+                # Sequential learners accept initial_config — the same ROSE-native
+                # mechanism used by the parallel path, giving tasks access to
+                # parameters and iteration via kwargs at every iteration.
+                ic = builder.build_learner_config()
+                if ic is not None:
+                    start_kwargs["initial_config"] = ic
+
+            try:
+                async for state in learner.start(**start_kwargs):
+                    print(
+                        f"[iter {state.iteration}]  metric={state.metric_value}",
+                        flush=True,
+                    )
+            finally:
+                await asyncflow.shutdown()
+
+        return _workflow
diff --git a/rose/spec/adapters.py b/rose/spec/adapters.py
new file mode 100644
index 00000000..e8910108
--- /dev/null
+++ b/rose/spec/adapters.py
@@ -0,0 +1,111 @@
+from __future__ import annotations
+
+from collections.abc import Callable
+
+from .schema import RemoteConfig, TaskDef
+
+
+class TaskAdapterFactory:
+    @staticmethod
+    def make_closure(task_def: TaskDef, remote: RemoteConfig) -> Callable:
+        td = dict(task_def.task_description or {})
+        if task_def.type == "shell":
+            return _make_shell_closure(task_def.command, td)
+        return _make_python_closure(task_def.function, remote.pythonpath, td)
+
+    @staticmethod
+    def as_executable(task_def: TaskDef) -> bool:
+        return task_def.type == "shell"
+
+    @staticmethod
+    def make_dispatch_closure(
+        slot_name: str,
+        task_defs: list[TaskDef],
+        remote: RemoteConfig,
+    ) -> Callable:
+        """Dispatch closure for parallel learners: routes per-learner based on learner_id kwarg.
+
+        task_description uses the first learner's value — asyncflow reads it once at registration.
+        """
+        td = dict(task_defs[0].task_description or {})
+        if task_defs[0].type == "shell":
+            return _make_shell_dispatch(
+                {i: tdi.command for i, tdi in enumerate(task_defs)}, slot_name, td
+            )
+        return _make_python_dispatch(
+            {i: tdi.function for i, tdi in enumerate(task_defs)},
+            list(remote.pythonpath),
+            slot_name,
+            td,
+        )
+
+
+def _make_shell_closure(command: str, task_description: dict) -> Callable:
+    _cmd = command
+    _td = task_description
+
+    async def _task(*args, task_description=_td, **kwargs) -> str:
+        return _cmd.format_map(kwargs)
+
+    _task.__name__ = "shell_task"
+    return _task
+
+
+def _make_python_closure(spec: str, remote_paths: list[str], task_description: dict) -> Callable:
+    _spec = spec
+    _paths = list(remote_paths)
+    _td = task_description
+
+    async def _task(*args, task_description=_td, **kwargs):
+        import importlib as _il
+        import inspect as _ins
+        import sys as _sys
+
+        for p in _paths:
+            if p not in _sys.path:
+                _sys.path.insert(0, p)
+        mod_path, fn_name = _spec.rsplit(":", 1)
+        fn = getattr(_il.import_module(mod_path), fn_name)
+        result = fn(*args, **kwargs)
+        return (await result) if _ins.iscoroutine(result) else result
+
+    _task.__name__ = spec.split(":")[-1]
+    return _task
+
+
+def _make_shell_dispatch(cmds: dict[int, str], slot_name: str, task_description: dict) -> Callable:
+    _cmds = dict(cmds)
+    _td = task_description
+
+    async def _dispatch(*args, task_description=_td, **kwargs) -> str:
+        cmd = _cmds[kwargs.get("learner_id", 0)]
+        return cmd.format_map(kwargs)
+
+    _dispatch.__name__ = f"{slot_name}_dispatch"
+    return _dispatch
+
+
+def _make_python_dispatch(
+    specs: dict[int, str], remote_paths: list[str], slot_name: str, task_description: dict
+) -> Callable:
+    _specs = dict(specs)
+    _paths = list(remote_paths)
+    _td = task_description
+
+    async def _dispatch(*args, task_description=_td, **kwargs):
+        import importlib as _il
+        import inspect as _ins
+        import sys as _sys
+
+        lid = kwargs.pop("learner_id", 0)  # strip internal routing key before calling user fn
+        spec = _specs[lid]
+        for p in _paths:
+            if p not in _sys.path:
+                _sys.path.insert(0, p)
+        mod_path, fn_name = spec.rsplit(":", 1)
+        fn = getattr(_il.import_module(mod_path), fn_name)
+        result = fn(*args, **kwargs)
+        return (await result) if _ins.iscoroutine(result) else result
+
+    _dispatch.__name__ = f"{slot_name}_dispatch"
+    return _dispatch
diff --git a/rose/spec/builder.py b/rose/spec/builder.py
new file mode 100644
index 00000000..7c158faa
--- /dev/null
+++ b/rose/spec/builder.py
@@ -0,0 +1,162 @@
+from __future__ import annotations
+
+from pathlib import Path
+
+from rose.learner import Learner
+
+from .adapters import TaskAdapterFactory
+from .schema import _REQUIRED_SLOTS, TrackingConfig, WorkflowConfig
+
+# Slots that exist as fields on LearnerConfig; uncertainty is required by uq_active_learner
+# but is not a LearnerConfig field — filter it out when constructing LearnerConfig.
+_LEARNER_CONFIG_SLOTS: frozenset[str] = frozenset(
+    {"simulation", "training", "prediction", "active_learn", "environment", "update"}
+)
+
+
+def _get_learner_class(learner_type: str):
+    if learner_type == "sequential_active_learner":
+        from rose.al.active_learner import SequentialActiveLearner
+
+        return SequentialActiveLearner
+    if learner_type == "parallel_active_learner":
+        from rose.al.active_learner import ParallelActiveLearner
+
+        return ParallelActiveLearner
+    if learner_type == "sequential_reinforcement_learner":
+        from rose.rl.reinforcement_learner import SequentialReinforcementLearner
+
+        return SequentialReinforcementLearner
+    if learner_type == "uq_active_learner":
+        from rose.uq.uq_active_learner import SeqUQLearner
+
+        return SeqUQLearner
+    raise ValueError(f"No learner class registered for type '{learner_type}'")
+
+
+class LearnerBuilder:
+    def __init__(self, source: str | Path | WorkflowConfig, asyncflow) -> None:
+        if isinstance(source, (str, Path)):
+            source = WorkflowConfig.from_yaml(source)
+        elif hasattr(source, "config"):  # duck-typed: also accepts a WorkflowSpec
+            source = source.config
+        self.config = source
+        self.asyncflow = asyncflow
+
+    def build(self) -> Learner:
+        cfg = self.config
+        learner = _get_learner_class(cfg.learner.type)(self.asyncflow)
+
+        if cfg.learners is not None:
+            self._register_dispatched_tasks(learner, cfg)
+        else:
+            self._register_flat_tasks(learner, cfg)
+
+        crit = cfg.stop_criterion
+        c_closure = TaskAdapterFactory.make_closure(crit.evaluator, cfg.remote)
+        c_as_exec = TaskAdapterFactory.as_executable(crit.evaluator)
+        learner.as_stop_criterion(
+            metric_name=crit.metric,
+            threshold=crit.threshold,
+            operator=crit.operator,
+            as_executable=c_as_exec,
+        )(c_closure)
+
+        _attach_tracker(learner, cfg.tracking)
+        return learner
+
+    def _register_flat_tasks(self, learner: Learner, cfg: WorkflowConfig) -> None:
+        for slot_name, task_def in cfg.tasks.items():
+            closure = TaskAdapterFactory.make_closure(task_def, cfg.remote)
+            as_exec = TaskAdapterFactory.as_executable(task_def)
+            getattr(learner, f"{slot_name}_task")(as_executable=as_exec)(closure)
+
+    def _register_dispatched_tasks(self, learner: Learner, cfg: WorkflowConfig) -> None:
+        required = _REQUIRED_SLOTS[cfg.learner.type]
+        for slot_name in required:
+            task_defs = [ld.tasks[slot_name] for ld in cfg.learners]
+            as_exec = task_defs[0].type == "shell"
+            closure = TaskAdapterFactory.make_dispatch_closure(slot_name, task_defs, cfg.remote)
+            getattr(learner, f"{slot_name}_task")(as_executable=as_exec)(closure)
+
+    def build_learner_config(self):
+        """Single LearnerConfig for sequential learners.
+
+        Passed as initial_config to learner.start() so that parameters and iteration reach task
+        kwargs via the same ROSE-native mechanism used by the parallel path.  Returns None when no
+        parameters are defined.
+        """
+        cfg = self.config
+        if not cfg.parameters and not cfg.remote.pythonpath:
+            return None
+        from rose.learner import LearnerConfig, TaskConfig
+
+        required = _REQUIRED_SLOTS[cfg.learner.type]
+        max_iter = cfg.learner.max_iter
+        params = dict(cfg.parameters)
+        params["pythonpath"] = list(cfg.remote.pythonpath)
+        schedule = {n: TaskConfig(kwargs={**params, "iteration": n}) for n in range(max_iter + 1)}
+        schedule[-1] = TaskConfig(kwargs={**params, "iteration": max_iter})
+        lc_slots = required & _LEARNER_CONFIG_SLOTS
+        return LearnerConfig(**{slot: schedule for slot in lc_slots}, criterion=schedule)
+
+    def build_learner_configs(self):
+        """LearnerConfig list for parallel learners.
+
+        Returns None for non-parallel learner types (use build_learner_config
+        instead) or when there's nothing to inject: no per-learner `learners:`
+        block, no `parameters`, and no `remote.pythonpath`.
+
+        When `learners:` is absent (shared tasks across learners), the list
+        length falls back to `learner.parallel_learners` so that `parameters`/
+        `pythonpath` still reach every learner's task kwargs.
+        """
+        cfg = self.config
+        if cfg.learner.type != "parallel_active_learner":
+            return None
+        if cfg.learners is None and not cfg.parameters and not cfg.remote.pythonpath:
+            return None
+        from rose.learner import LearnerConfig, TaskConfig
+
+        required = _REQUIRED_SLOTS[cfg.learner.type]
+        configs = []
+        max_iter = cfg.learner.max_iter
+        params = dict(cfg.parameters)
+        params["pythonpath"] = list(cfg.remote.pythonpath)
+        lc_slots = required & _LEARNER_CONFIG_SLOTS
+        learner_defs = cfg.learners or [None] * cfg.learner.parallel_learners
+        for i, learner_def in enumerate(learner_defs):
+            # learner_id    → dispatch routing key (popped by closure, never reaches user fn)
+            # iteration     → per-entry so get_task_config(slot, n) returns the right value
+            # learner_label → human-readable learner name; only injected when non-empty
+            # parameters    → user-defined values from the YAML parameters: block
+            base = {"learner_id": i, **params}
+            if learner_def is not None and learner_def.label:
+                base["learner_label"] = learner_def.label
+            schedule = {n: TaskConfig(kwargs={**base, "iteration": n}) for n in range(max_iter + 1)}
+            schedule[-1] = TaskConfig(kwargs={**base, "iteration": max_iter})
+            configs.append(
+                LearnerConfig(**{slot: schedule for slot in lc_slots}, criterion=schedule)
+            )
+        return configs
+
+
+def _attach_tracker(learner: Learner, tracking: TrackingConfig) -> None:
+    if tracking.backend == "mlflow":
+        from rose.integrations.mlflow_tracker import MLflowTracker
+
+        learner.add_tracker(
+            MLflowTracker(
+                experiment_name=tracking.experiment,
+                run_name=tracking.run_name,
+            )
+        )
+    elif tracking.backend == "clearml":
+        from rose.integrations.clearml_tracker import ClearMLTracker
+
+        learner.add_tracker(
+            ClearMLTracker(
+                project_name=tracking.experiment,
+                task_name=tracking.run_name or "rose-spec-run",
+            )
+        )
diff --git a/rose/spec/schema.py b/rose/spec/schema.py
new file mode 100644
index 00000000..99641920
--- /dev/null
+++ b/rose/spec/schema.py
@@ -0,0 +1,181 @@
+from __future__ import annotations
+
+from pathlib import Path
+from typing import Any, Literal
+
+from pydantic import BaseModel, Field, model_validator
+
+
+# ── Task definition ───────────────────────────────────────────────────────────
+class TaskDef(BaseModel):
+    type: Literal["shell", "python"]
+    command: str | None = None  # required when type=="shell"
+    function: str | None = None  # required when type=="python"; "module:callable"
+    task_description: dict[str, Any] | None = None  # resource hints forwarded to asyncflow backend
+
+    model_config = {"extra": "forbid"}
+
+    @model_validator(mode="after")
+    def _check_fields(self) -> TaskDef:
+        if self.type == "shell" and not self.command:
+            raise ValueError("shell task requires 'command'")
+        if self.type == "python":
+            if not self.function:
+                raise ValueError("python task requires 'function'")
+            if ":" not in self.function:
+                raise ValueError("'function' must use 'module:callable' syntax")
+        return self
+
+
+# ── Stop criterion ────────────────────────────────────────────────────────────
+class StopCriterionDef(BaseModel):
+    metric: str
+    threshold: float
+    operator: Literal["<", ">", "==", "<=", ">="] = "<"
+    evaluator: TaskDef
+
+    model_config = {"extra": "forbid"}
+
+
+# ── Learner spec (YAML key: "learner") ───────────────────────────────────────
+_REQUIRED_SLOTS: dict[str, frozenset[str]] = {
+    "sequential_active_learner": frozenset({"simulation", "training", "active_learn"}),
+    "parallel_active_learner": frozenset({"simulation", "training", "active_learn"}),
+    "sequential_reinforcement_learner": frozenset({"environment", "update"}),
+    "uq_active_learner": frozenset(
+        {"simulation", "training", "prediction", "active_learn", "uncertainty"}
+    ),
+}
+_ALL_SLOTS: frozenset[str] = frozenset().union(*_REQUIRED_SLOTS.values())
+
+
+class LearnerSpec(BaseModel):
+    type: str
+    max_iter: int = 0
+    parallel_learners: int = 2  # used only when learners: is absent for parallel types
+
+    model_config = {"extra": "forbid"}
+
+
+# ── Per-learner definition (parallel learners only) ───────────────────────────
+class LearnerDef(BaseModel):
+    label: str = ""
+    simulation: TaskDef | None = None
+    training: TaskDef | None = None
+    active_learn: TaskDef | None = None
+    environment: TaskDef | None = None
+    update: TaskDef | None = None
+    prediction: TaskDef | None = None
+    uncertainty: TaskDef | None = None
+
+    model_config = {"extra": "forbid"}
+
+    @property
+    def tasks(self) -> dict[str, TaskDef]:
+        return {s: getattr(self, s) for s in _ALL_SLOTS if getattr(self, s) is not None}
+
+
+# ── Remote / tracking ────────────────────────────────────────────────────────
+class RemoteConfig(BaseModel):
+    pythonpath: list[str] = []
+
+    model_config = {"extra": "forbid"}
+
+
+class TrackingConfig(BaseModel):
+    backend: Literal["mlflow", "clearml", "none"] = "none"
+    experiment: str = "ROSE-Spec"
+    run_name: str | None = None
+
+    model_config = {"extra": "forbid"}
+
+
+# Keys the builder always injects — users must not put these in parameters:
+_RESERVED_PARAMETER_KEYS: frozenset[str] = frozenset(
+    {"learner_id", "learner_label", "iteration", "pythonpath"}
+)
+
+
+# ── Top-level spec ────────────────────────────────────────────────────────────
+class WorkflowConfig(BaseModel):
+    learner: LearnerSpec
+    # Task slots — explicit fields keep extra="forbid" and enable IDE autocomplete.
+    # Access non-None slots as a dict via the .tasks property.
+    simulation: TaskDef | None = None
+    training: TaskDef | None = None
+    active_learn: TaskDef | None = None
+    environment: TaskDef | None = None
+    update: TaskDef | None = None
+    prediction: TaskDef | None = None
+    uncertainty: TaskDef | None = None
+    learners: list[LearnerDef] | None = None
+    stop_criterion: StopCriterionDef
+    parameters: dict[str, Any] = Field(default_factory=dict)
+    remote: RemoteConfig = Field(default_factory=RemoteConfig)
+    tracking: TrackingConfig = Field(default_factory=TrackingConfig)
+
+    model_config = {"extra": "forbid"}
+
+    @property
+    def tasks(self) -> dict[str, TaskDef]:
+        return {s: getattr(self, s) for s in _ALL_SLOTS if getattr(self, s) is not None}
+
+    @model_validator(mode="after")
+    def _validate_task_slots(self) -> WorkflowConfig:
+        ltype = self.learner.type
+        required = _REQUIRED_SLOTS.get(ltype)
+        if required is None:
+            raise ValueError(
+                f"Unknown learner type '{ltype}'. Supported: {sorted(_REQUIRED_SLOTS.keys())}"
+            )
+        is_parallel = ltype == "parallel_active_learner"
+
+        if is_parallel and self.learners is not None:
+            for ld in self.learners:
+                missing = required - set(ld.tasks.keys())
+                if missing:
+                    raise ValueError(f"Learner '{ld.label}' missing: {sorted(missing)}")
+                extra = set(ld.tasks.keys()) - required
+                if extra:
+                    raise ValueError(f"Learner '{ld.label}' unexpected fields: {sorted(extra)}")
+            for slot in required:
+                types = {ld.tasks[slot].type for ld in self.learners}
+                if len(types) > 1:
+                    raise ValueError(
+                        f"Slot '{slot}' has mixed types across learners: {types}. "
+                        "All learners must use the same type for a given slot."
+                    )
+                descs = [dict(ld.tasks[slot].task_description or {}) for ld in self.learners]
+                if len(descs) > 1 and any(d != descs[0] for d in descs[1:]):
+                    raise ValueError(
+                        f"Slot '{slot}' has different task_description values across learners. "
+                        "asyncflow registers task_description once per slot at registration time — "
+                        "all learners must use the same value. Use identical task_description "
+                        "across all learners or omit it from all but the first."
+                    )
+        else:
+            present = set(self.tasks.keys())
+            missing = required - present
+            if missing:
+                raise ValueError(f"learner type '{ltype}' requires: {sorted(missing)}")
+            extra = present - required
+            if extra:
+                raise ValueError(f"Unexpected task fields for '{ltype}': {sorted(extra)}")
+        return self
+
+    @model_validator(mode="after")
+    def _validate_parameters(self) -> WorkflowConfig:
+        conflicts = _RESERVED_PARAMETER_KEYS & set(self.parameters.keys())
+        if conflicts:
+            raise ValueError(
+                f"'parameters' must not use reserved keys: {sorted(conflicts)}. "
+                "These are injected automatically by the spec builder."
+            )
+        return self
+
+    @classmethod
+    def from_yaml(cls, path: str | Path) -> WorkflowConfig:
+        import yaml
+
+        raw = yaml.safe_load(Path(path).read_text())
+        return cls.model_validate(raw)
diff --git a/tests/__init__.py b/tests/__init__.py
new file mode 100644
index 00000000..e69de29b
diff --git a/tests/integration/__init__.py b/tests/integration/__init__.py
new file mode 100644
index 00000000..e69de29b
diff --git a/tests/integration/spec/__init__.py b/tests/integration/spec/__init__.py
new file mode 100644
index 00000000..e69de29b
diff --git a/tests/integration/spec/helpers.py b/tests/integration/spec/helpers.py
new file mode 100644
index 00000000..45c60d70
--- /dev/null
+++ b/tests/integration/spec/helpers.py
@@ -0,0 +1,40 @@
+"""Lightweight task helpers used by the YAML spec integration tests."""
+
+# Shared capture list — cleared by tests that need to inspect received kwargs.
+received_kwargs: list[dict] = []
+
+
+async def sim(*args, **kwargs):
+    return [1.0, 2.0, 3.0]
+
+
+async def train(data, **kwargs):
+    return {"mean": sum(data) / len(data)}
+
+
+async def active_learn(sim_result, model, **kwargs):
+    return abs(model["mean"] - 2.0)
+
+
+async def criterion(*args, **kwargs):
+    return 0.05
+
+
+# ── Parameter-capturing variants ──────────────────────────────────────────────
+
+
+async def sim_capture(*args, **kwargs):
+    received_kwargs.append(dict(kwargs))
+    return [1.0, 2.0, 3.0]
+
+
+async def train_capture(data, **kwargs):
+    return {"mean": sum(data) / len(data)}
+
+
+async def active_learn_capture(sim_result, model, **kwargs):
+    return abs(model["mean"] - 2.0)
+
+
+async def criterion_capture(*args, **kwargs):
+    return 0.05
diff --git a/tests/integration/spec/test_yaml_workflow.py b/tests/integration/spec/test_yaml_workflow.py
new file mode 100644
index 00000000..7ce139b3
--- /dev/null
+++ b/tests/integration/spec/test_yaml_workflow.py
@@ -0,0 +1,217 @@
+"""Integration tests for the YAML spec layer — full AL loop, no HPC required."""
+
+import textwrap
+from concurrent.futures import ThreadPoolExecutor
+
+import pytest
+from radical.asyncflow import WorkflowEngine
+from rhapsody.backends import ConcurrentExecutionBackend
+
+from rose.spec import load_spec
+from rose.spec.builder import LearnerBuilder
+from rose.spec.schema import WorkflowConfig
+
+# ── Sequential AL via YAML ────────────────────────────────────────────────────
+
+SEQ_YAML = textwrap.dedent("""\
+    learner:
+      type: sequential_active_learner
+      max_iter: 3
+
+    simulation:
+      type: python
+      function: tests.integration.spec.helpers:sim
+
+    training:
+      type: python
+      function: tests.integration.spec.helpers:train
+
+    active_learn:
+      type: python
+      function: tests.integration.spec.helpers:active_learn
+
+    stop_criterion:
+      metric: mse
+      threshold: 0.01
+      operator: "<"
+      evaluator:
+        type: python
+        function: tests.integration.spec.helpers:criterion
+""")
+
+
+@pytest.mark.asyncio
+@pytest.mark.integration
+async def test_yaml_sequential_workflow(tmp_path):
+    p = tmp_path / "spec.yaml"
+    p.write_text(SEQ_YAML)
+
+    engine = await ConcurrentExecutionBackend(ThreadPoolExecutor())
+    asyncflow = await WorkflowEngine.create(engine)
+
+    cfg = WorkflowConfig.from_yaml(p)
+    builder = LearnerBuilder(cfg, asyncflow)
+    learner = builder.build()
+
+    states = []
+    async for state in learner.start(max_iter=3):
+        states.append(state)
+
+    assert len(states) == 3
+    await asyncflow.shutdown()
+
+
+# ── load_spec convenience wrapper ─────────────────────────────────────────────
+
+
+@pytest.mark.asyncio
+@pytest.mark.integration
+async def test_load_spec_returns_workflow_spec(tmp_path):
+    p = tmp_path / "spec.yaml"
+    p.write_text(SEQ_YAML)
+    spec = load_spec(p)
+    assert hasattr(spec, "workflow")
+    assert callable(spec.workflow)
+
+
+# ── Sequential AL with parameters: block ─────────────────────────────────────
+
+SEQ_WITH_PARAMS_YAML = textwrap.dedent("""\
+    learner:
+      type: sequential_active_learner
+      max_iter: 2
+
+    simulation:
+      type: python
+      function: tests.integration.spec.helpers:sim_capture
+
+    training:
+      type: python
+      function: tests.integration.spec.helpers:train_capture
+
+    active_learn:
+      type: python
+      function: tests.integration.spec.helpers:active_learn_capture
+
+    stop_criterion:
+      metric: mse
+      threshold: 0.01
+      operator: "<"
+      evaluator:
+        type: python
+        function: tests.integration.spec.helpers:criterion_capture
+
+    parameters:
+      dataset: test_ds
+      scale: 1.5
+""")
+
+
+@pytest.mark.asyncio
+@pytest.mark.integration
+async def test_sequential_workflow_parameters_reach_tasks(tmp_path):
+    import tests.integration.spec.helpers as helpers
+
+    helpers.received_kwargs.clear()
+
+    p = tmp_path / "spec.yaml"
+    p.write_text(SEQ_WITH_PARAMS_YAML)
+
+    engine = await ConcurrentExecutionBackend(ThreadPoolExecutor())
+    asyncflow = await WorkflowEngine.create(engine)
+
+    cfg = WorkflowConfig.from_yaml(p)
+    builder = LearnerBuilder(cfg, asyncflow)
+    learner = builder.build()
+
+    ic = builder.build_learner_config()
+    start_kwargs = {"max_iter": 2}
+    if ic is not None:
+        start_kwargs["initial_config"] = ic
+
+    async for _ in learner.start(**start_kwargs):
+        pass
+
+    await asyncflow.shutdown()
+
+    assert len(helpers.received_kwargs) > 0, "sim_capture was never called"
+    for kw in helpers.received_kwargs:
+        assert kw.get("dataset") == "test_ds", f"expected dataset in kwargs, got {kw}"
+        assert kw.get("scale") == 1.5, f"expected scale in kwargs, got {kw}"
+
+
+# ── Parallel AL with parameters: block ───────────────────────────────────────
+
+PAR_WITH_PARAMS_YAML = textwrap.dedent("""\
+    learner:
+      type: parallel_active_learner
+      max_iter: 2
+
+    learners:
+      - label: rf
+        simulation:
+          type: python
+          function: tests.integration.spec.helpers:sim_capture
+        training:
+          type: python
+          function: tests.integration.spec.helpers:train_capture
+        active_learn:
+          type: python
+          function: tests.integration.spec.helpers:active_learn_capture
+      - label: mlp
+        simulation:
+          type: python
+          function: tests.integration.spec.helpers:sim_capture
+        training:
+          type: python
+          function: tests.integration.spec.helpers:train_capture
+        active_learn:
+          type: python
+          function: tests.integration.spec.helpers:active_learn_capture
+
+    stop_criterion:
+      metric: mse
+      threshold: 0.01
+      operator: "<"
+      evaluator:
+        type: python
+        function: tests.integration.spec.helpers:criterion_capture
+
+    parameters:
+      dataset: par_ds
+      lr: 0.01
+""")
+
+
+@pytest.mark.asyncio
+@pytest.mark.integration
+async def test_parallel_workflow_parameters_reach_tasks(tmp_path):
+    import tests.integration.spec.helpers as helpers
+
+    helpers.received_kwargs.clear()
+
+    p = tmp_path / "spec.yaml"
+    p.write_text(PAR_WITH_PARAMS_YAML)
+
+    engine = await ConcurrentExecutionBackend(ThreadPoolExecutor())
+    asyncflow = await WorkflowEngine.create(engine)
+
+    cfg = WorkflowConfig.from_yaml(p)
+    builder = LearnerBuilder(cfg, asyncflow)
+    learner = builder.build()
+
+    lcs = builder.build_learner_configs()
+    start_kwargs = {"max_iter": 2, "parallel_learners": len(lcs), "learner_configs": lcs}
+
+    async for _ in learner.start(**start_kwargs):
+        pass
+
+    await asyncflow.shutdown()
+
+    assert len(helpers.received_kwargs) > 0, "sim_capture was never called"
+    for kw in helpers.received_kwargs:
+        assert kw.get("dataset") == "par_ds", f"expected dataset in kwargs, got {kw}"
+        assert kw.get("lr") == 0.01, f"expected lr in kwargs, got {kw}"
+        # learner_id is stripped by the dispatch closure; learner_label passes through
+        assert "learner_id" not in kw, "learner_id should be stripped by dispatch"
+        assert kw.get("learner_label") in {"rf", "mlp"}
diff --git a/tests/unit/spec/__init__.py b/tests/unit/spec/__init__.py
new file mode 100644
index 00000000..e69de29b
diff --git a/tests/unit/spec/test_adapters.py b/tests/unit/spec/test_adapters.py
new file mode 100644
index 00000000..c6608c27
--- /dev/null
+++ b/tests/unit/spec/test_adapters.py
@@ -0,0 +1,168 @@
+"""Unit tests for rose.spec.adapters — closure factories, no HPC needed."""
+
+import asyncio
+import inspect
+
+from rose.spec.adapters import TaskAdapterFactory
+from rose.spec.schema import RemoteConfig, TaskDef
+
+
+def _remote() -> RemoteConfig:
+    return RemoteConfig()
+
+
+# ── Shell closure ─────────────────────────────────────────────────────────────
+
+
+def test_shell_closure_returns_command():
+    td = TaskDef(type="shell", command="echo hello")
+    fn = TaskAdapterFactory.make_closure(td, _remote())
+    result = asyncio.run(fn())
+    assert result == "echo hello"
+
+
+def test_shell_as_executable_true():
+    td = TaskDef(type="shell", command="x")
+    assert TaskAdapterFactory.as_executable(td) is True
+
+
+# ── Python closure ────────────────────────────────────────────────────────────
+
+
+def test_python_closure_calls_sync_function():
+    td = TaskDef(type="python", function="os.path:join")
+    fn = TaskAdapterFactory.make_closure(td, _remote())
+    result = asyncio.run(fn("a", "b"))
+    assert result == "a/b"
+
+
+def test_python_closure_calls_async_function():
+    async def _async_fn(x):
+        return x * 2
+
+    import types
+
+    mod = types.ModuleType("_test_async_mod")
+    mod.double = _async_fn
+    import sys
+
+    sys.modules["_test_async_mod"] = mod
+
+    td = TaskDef(type="python", function="_test_async_mod:double")
+    fn = TaskAdapterFactory.make_closure(td, _remote())
+    result = asyncio.run(fn(21))
+    assert result == 42
+
+    del sys.modules["_test_async_mod"]
+
+
+def test_python_as_executable_false():
+    td = TaskDef(type="python", function="os:getcwd")
+    assert TaskAdapterFactory.as_executable(td) is False
+
+
+def test_python_closure_injects_remote_path(tmp_path):
+    (tmp_path / "myutil.py").write_text("def add(a, b): return a + b\n")
+    td = TaskDef(type="python", function="myutil:add")
+    fn = TaskAdapterFactory.make_closure(td, RemoteConfig(pythonpath=[str(tmp_path)]))
+    result = asyncio.run(fn(3, 4))
+    assert result == 7
+
+
+# ── Shell dispatch ────────────────────────────────────────────────────────────
+
+
+def test_shell_dispatch_routes_by_learner_id():
+    tds = [
+        TaskDef(type="shell", command="cmd_0"),
+        TaskDef(type="shell", command="cmd_1"),
+    ]
+    fn = TaskAdapterFactory.make_dispatch_closure("simulation", tds, _remote())
+    assert asyncio.run(fn(learner_id=0)) == "cmd_0"
+    assert asyncio.run(fn(learner_id=1)) == "cmd_1"
+
+
+def test_shell_dispatch_defaults_to_zero():
+    tds = [TaskDef(type="shell", command="cmd_default")]
+    fn = TaskAdapterFactory.make_dispatch_closure("simulation", tds, _remote())
+    assert asyncio.run(fn()) == "cmd_default"
+
+
+# ── Python dispatch ───────────────────────────────────────────────────────────
+
+
+def test_python_dispatch_routes_by_learner_id():
+    tds = [
+        TaskDef(type="python", function="os.path:basename"),
+        TaskDef(type="python", function="os.path:dirname"),
+    ]
+    fn = TaskAdapterFactory.make_dispatch_closure("training", tds, _remote())
+    assert asyncio.run(fn("/a/b/c", learner_id=0)) == "c"
+    assert asyncio.run(fn("/a/b/c", learner_id=1)) == "/a/b"
+
+
+# ── task_description default kwarg injection ──────────────────────────────────
+
+
+def test_shell_closure_task_description_in_signature():
+    td = TaskDef(type="shell", command="echo hi", task_description={"cpu_count": 4})
+    fn = TaskAdapterFactory.make_closure(td, _remote())
+    sig = inspect.signature(fn)
+    assert "task_description" in sig.parameters
+    assert sig.parameters["task_description"].default == {"cpu_count": 4}
+
+
+def test_shell_closure_no_task_description_defaults_empty():
+    td = TaskDef(type="shell", command="echo hi")
+    fn = TaskAdapterFactory.make_closure(td, _remote())
+    sig = inspect.signature(fn)
+    assert sig.parameters["task_description"].default == {}
+
+
+def test_python_closure_task_description_in_signature():
+    td = TaskDef(type="python", function="os:getcwd", task_description={"gpu_count": 2})
+    fn = TaskAdapterFactory.make_closure(td, _remote())
+    sig = inspect.signature(fn)
+    assert sig.parameters["task_description"].default == {"gpu_count": 2}
+
+
+# ── Shell command interpolation ───────────────────────────────────────────────
+
+
+def test_shell_closure_interpolates_kwargs():
+    td = TaskDef(type="shell", command="python sim.py --dataset {dataset} --n {n}")
+    fn = TaskAdapterFactory.make_closure(td, _remote())
+    result = asyncio.run(fn(dataset="m3dc1", n=42))
+    assert result == "python sim.py --dataset m3dc1 --n 42"
+
+
+def test_shell_closure_no_placeholders_unchanged():
+    td = TaskDef(type="shell", command="python sim.py")
+    fn = TaskAdapterFactory.make_closure(td, _remote())
+    result = asyncio.run(fn(dataset="ignored"))
+    assert result == "python sim.py"
+
+
+def test_shell_dispatch_interpolates_kwargs():
+    tds = [
+        TaskDef(type="shell", command="python sim.py --model rf --dataset {dataset}"),
+        TaskDef(type="shell", command="python sim.py --model mlp --dataset {dataset}"),
+    ]
+    fn = TaskAdapterFactory.make_dispatch_closure("simulation", tds, _remote())
+    assert (
+        asyncio.run(fn(learner_id=0, dataset="m3dc1")) == "python sim.py --model rf --dataset m3dc1"
+    )
+    assert (
+        asyncio.run(fn(learner_id=1, dataset="test_ds"))
+        == "python sim.py --model mlp --dataset test_ds"
+    )
+
+
+def test_dispatch_closure_task_description_uses_first_candidate():
+    tds = [
+        TaskDef(type="shell", command="cmd_0", task_description={"cpu_count": 8}),
+        TaskDef(type="shell", command="cmd_1", task_description={"cpu_count": 4}),
+    ]
+    fn = TaskAdapterFactory.make_dispatch_closure("simulation", tds, _remote())
+    sig = inspect.signature(fn)
+    assert sig.parameters["task_description"].default == {"cpu_count": 8}
diff --git a/tests/unit/spec/test_builder.py b/tests/unit/spec/test_builder.py
new file mode 100644
index 00000000..9a893293
--- /dev/null
+++ b/tests/unit/spec/test_builder.py
@@ -0,0 +1,310 @@
+"""Unit tests for rose.spec.builder — LearnerConfig construction, no HPC needed."""
+
+import textwrap
+from unittest.mock import MagicMock
+
+from rose.spec.schema import WorkflowConfig
+
+SEQ_WITH_PARAMS = textwrap.dedent("""\
+    learner:
+      type: sequential_active_learner
+      max_iter: 3
+    simulation:
+      type: python
+      function: mymod:sim
+    training:
+      type: python
+      function: mymod:train
+    active_learn:
+      type: python
+      function: mymod:select
+    stop_criterion:
+      metric: mse
+      threshold: 0.1
+      evaluator:
+        type: python
+        function: mymod:eval
+    parameters:
+      dataset: test_ds
+      scale: 2.5
+""")
+
+SEQ_NO_PARAMS = textwrap.dedent("""\
+    learner:
+      type: sequential_active_learner
+      max_iter: 3
+    simulation:
+      type: python
+      function: mymod:sim
+    training:
+      type: python
+      function: mymod:train
+    active_learn:
+      type: python
+      function: mymod:select
+    stop_criterion:
+      metric: mse
+      threshold: 0.1
+      evaluator:
+        type: python
+        function: mymod:eval
+""")
+
+PAR_WITH_PARAMS = textwrap.dedent("""\
+    learner:
+      type: parallel_active_learner
+      max_iter: 2
+    learners:
+      - label: rf
+        simulation:
+          type: python
+          function: mymod:sim
+        training:
+          type: python
+          function: mymod:train
+        active_learn:
+          type: python
+          function: mymod:select
+      - label: mlp
+        simulation:
+          type: python
+          function: mymod:sim
+        training:
+          type: python
+          function: mymod:train
+        active_learn:
+          type: python
+          function: mymod:select
+    stop_criterion:
+      metric: r2
+      threshold: 0.9
+      operator: ">"
+      evaluator:
+        type: python
+        function: mymod:eval
+    parameters:
+      dataset: prod_ds
+      lr: 0.001
+""")
+
+
+def _make_builder(yaml_text, tmp_path):
+    p = tmp_path / "spec.yaml"
+    p.write_text(yaml_text)
+    cfg = WorkflowConfig.from_yaml(p)
+    from rose.spec.builder import LearnerBuilder
+
+    return LearnerBuilder(cfg, MagicMock())
+
+
+# ── build_learner_config (sequential path) ────────────────────────────────────
+
+
+def test_build_learner_config_no_parameters_returns_none(tmp_path):
+    builder = _make_builder(SEQ_NO_PARAMS, tmp_path)
+    assert builder.build_learner_config() is None
+
+
+def test_build_learner_config_returns_learner_config(tmp_path):
+    from rose.learner import LearnerConfig
+
+    builder = _make_builder(SEQ_WITH_PARAMS, tmp_path)
+    lc = builder.build_learner_config()
+    assert lc is not None
+    assert isinstance(lc, LearnerConfig)
+
+
+def test_build_learner_config_per_iteration_kwargs(tmp_path):
+    builder = _make_builder(SEQ_WITH_PARAMS, tmp_path)
+    lc = builder.build_learner_config()
+    # iteration 0: parameters + iteration=0
+    kw0 = lc.simulation[0].kwargs
+    assert kw0["dataset"] == "test_ds"
+    assert kw0["scale"] == 2.5
+    assert kw0["iteration"] == 0
+    # iteration 3 (max_iter): parameters + iteration=3
+    kw3 = lc.simulation[3].kwargs
+    assert kw3["dataset"] == "test_ds"
+    assert kw3["iteration"] == 3
+
+
+def test_build_learner_config_criterion_schedule_included(tmp_path):
+    builder = _make_builder(SEQ_WITH_PARAMS, tmp_path)
+    lc = builder.build_learner_config()
+    assert lc.criterion is not None
+    assert lc.criterion[0].kwargs["dataset"] == "test_ds"
+
+
+# ── build_learner_configs (parallel path) ────────────────────────────────────
+
+
+def test_build_learner_configs_no_learners_returns_none(tmp_path):
+    builder = _make_builder(SEQ_WITH_PARAMS, tmp_path)
+    assert builder.build_learner_configs() is None
+
+
+def test_build_learner_configs_with_parameters(tmp_path):
+    from rose.learner import LearnerConfig
+
+    builder = _make_builder(PAR_WITH_PARAMS, tmp_path)
+    lcs = builder.build_learner_configs()
+    assert lcs is not None
+    assert len(lcs) == 2
+    assert all(isinstance(lc, LearnerConfig) for lc in lcs)
+
+
+def test_build_learner_configs_learner_0_kwargs(tmp_path):
+    builder = _make_builder(PAR_WITH_PARAMS, tmp_path)
+    lcs = builder.build_learner_configs()
+    kw = lcs[0].simulation[0].kwargs
+    assert kw["learner_id"] == 0
+    assert kw["learner_label"] == "rf"
+    assert kw["dataset"] == "prod_ds"
+    assert kw["lr"] == 0.001
+    assert kw["iteration"] == 0
+
+
+def test_build_learner_configs_learner_1_kwargs(tmp_path):
+    builder = _make_builder(PAR_WITH_PARAMS, tmp_path)
+    lcs = builder.build_learner_configs()
+    kw = lcs[1].simulation[1].kwargs
+    assert kw["learner_id"] == 1
+    assert kw["learner_label"] == "mlp"
+    assert kw["dataset"] == "prod_ds"
+    assert kw["iteration"] == 1
+
+
+PAR_SHARED_WITH_PARAMS = textwrap.dedent("""\
+    learner:
+      type: parallel_active_learner
+      max_iter: 2
+      parallel_learners: 3
+    simulation:
+      type: python
+      function: mymod:sim
+    training:
+      type: python
+      function: mymod:train
+    active_learn:
+      type: python
+      function: mymod:select
+    stop_criterion:
+      metric: r2
+      threshold: 0.9
+      operator: ">"
+      evaluator:
+        type: python
+        function: mymod:eval
+    parameters:
+      dataset: prod_ds
+      lr: 0.001
+""")
+
+
+def test_build_learner_configs_shared_tasks_with_parameters_falls_back_to_parallel_learners(
+    tmp_path,
+):
+    from rose.learner import LearnerConfig
+
+    builder = _make_builder(PAR_SHARED_WITH_PARAMS, tmp_path)
+    lcs = builder.build_learner_configs()
+    assert lcs is not None
+    assert len(lcs) == 3  # learner.parallel_learners, since learners: is absent
+    assert all(isinstance(lc, LearnerConfig) for lc in lcs)
+    kw0 = lcs[0].simulation[0].kwargs
+    assert kw0["learner_id"] == 0
+    assert kw0["dataset"] == "prod_ds"
+    assert "learner_label" not in kw0  # no per-learner label when learners: is absent
+    kw2 = lcs[2].simulation[1].kwargs
+    assert kw2["learner_id"] == 2
+    assert kw2["iteration"] == 1
+
+
+# ── pythonpath auto-injection ─────────────────────────────────────────────────
+
+SEQ_WITH_PYTHONPATH = textwrap.dedent("""\
+    learner:
+      type: sequential_active_learner
+      max_iter: 2
+    simulation:
+      type: python
+      function: mymod:sim
+    training:
+      type: python
+      function: mymod:train
+    active_learn:
+      type: python
+      function: mymod:select
+    stop_criterion:
+      metric: mse
+      threshold: 0.1
+      evaluator:
+        type: python
+        function: mymod:eval
+    remote:
+      pythonpath:
+        - /remote/path/a
+        - /remote/path/b
+""")
+
+PAR_WITH_PYTHONPATH = textwrap.dedent("""\
+    learner:
+      type: parallel_active_learner
+      max_iter: 2
+    learners:
+      - label: rf
+        simulation:
+          type: python
+          function: mymod:sim
+        training:
+          type: python
+          function: mymod:train
+        active_learn:
+          type: python
+          function: mymod:select
+      - label: mlp
+        simulation:
+          type: python
+          function: mymod:sim
+        training:
+          type: python
+          function: mymod:train
+        active_learn:
+          type: python
+          function: mymod:select
+    stop_criterion:
+      metric: r2
+      threshold: 0.9
+      operator: ">"
+      evaluator:
+        type: python
+        function: mymod:eval
+    remote:
+      pythonpath:
+        - /remote/path/a
+""")
+
+
+def test_build_learner_config_injects_pythonpath(tmp_path):
+    builder = _make_builder(SEQ_WITH_PYTHONPATH, tmp_path)
+    lc = builder.build_learner_config()
+    assert lc is not None
+    assert lc.simulation[0].kwargs["pythonpath"] == ["/remote/path/a", "/remote/path/b"]
+
+
+def test_build_learner_config_pythonpath_empty_when_no_remote(tmp_path):
+    builder = _make_builder(SEQ_WITH_PARAMS, tmp_path)
+    lc = builder.build_learner_config()
+    assert lc.simulation[0].kwargs["pythonpath"] == []
+
+
+def test_build_learner_configs_injects_pythonpath(tmp_path):
+    builder = _make_builder(PAR_WITH_PYTHONPATH, tmp_path)
+    lcs = builder.build_learner_configs()
+    assert lcs[0].simulation[0].kwargs["pythonpath"] == ["/remote/path/a"]
+    assert lcs[1].simulation[0].kwargs["pythonpath"] == ["/remote/path/a"]
+
+
+def test_build_learner_config_no_params_no_pythonpath_returns_none(tmp_path):
+    builder = _make_builder(SEQ_NO_PARAMS, tmp_path)
+    assert builder.build_learner_config() is None
diff --git a/tests/unit/spec/test_schema.py b/tests/unit/spec/test_schema.py
new file mode 100644
index 00000000..690adbf3
--- /dev/null
+++ b/tests/unit/spec/test_schema.py
@@ -0,0 +1,647 @@
+"""Unit tests for rose.spec.schema — YAML validation without any HPC machinery."""
+
+import textwrap
+
+import pytest
+from pydantic import ValidationError
+
+from rose.spec.schema import TaskDef, WorkflowConfig
+
+# ── TaskDef ───────────────────────────────────────────────────────────────────
+
+
+def test_taskdef_shell_valid():
+    t = TaskDef(type="shell", command="python sim.py")
+    assert t.type == "shell"
+    assert t.command == "python sim.py"
+
+
+def test_taskdef_python_valid():
+    t = TaskDef(type="python", function="mymod.sub:fn")
+    assert t.function == "mymod.sub:fn"
+
+
+def test_taskdef_shell_missing_command():
+    with pytest.raises(ValueError, match="shell task requires 'command'"):
+        TaskDef(type="shell")
+
+
+def test_taskdef_python_missing_function():
+    with pytest.raises(ValueError, match="python task requires 'function'"):
+        TaskDef(type="python")
+
+
+def test_taskdef_python_bad_syntax():
+    with pytest.raises(ValueError, match="module:callable"):
+        TaskDef(type="python", function="mymod.fn")
+
+
+def test_taskdef_extra_field_rejected():
+    with pytest.raises(ValidationError):
+        TaskDef(type="shell", command="x", unknown_field="y")
+
+
+def test_taskdef_task_description_roundtrip():
+    t = TaskDef(
+        type="shell", command="python sim.py", task_description={"cpu_count": 4, "gpu_count": 1}
+    )
+    assert t.task_description == {"cpu_count": 4, "gpu_count": 1}
+
+
+def test_taskdef_task_description_defaults_none():
+    t = TaskDef(type="shell", command="x")
+    assert t.task_description is None
+
+
+def test_sequential_spec_task_description_from_yaml(tmp_path):
+    yaml = textwrap.dedent("""\
+        learner:
+          type: sequential_active_learner
+          max_iter: 1
+        simulation:
+          type: shell
+          command: python sim.py
+          task_description:
+            cpu_count: 8
+        training:
+          type: python
+          function: mymod:train
+        active_learn:
+          type: python
+          function: mymod:select
+        stop_criterion:
+          metric: mse
+          threshold: 0.1
+          evaluator:
+            type: python
+            function: mymod:eval
+    """)
+    p = tmp_path / "td.yaml"
+    p.write_text(yaml)
+    cfg = WorkflowConfig.from_yaml(p)
+    assert cfg.simulation.task_description == {"cpu_count": 8}
+    assert cfg.training.task_description is None
+
+
+# ── WorkflowConfig — sequential learner ──────────────────────────────────────────
+
+SEQ_YAML = textwrap.dedent("""\
+    learner:
+      type: sequential_active_learner
+      max_iter: 5
+
+    simulation:
+      type: shell
+      command: python sim.py
+
+    training:
+      type: python
+      function: mymod:train
+
+    active_learn:
+      type: python
+      function: mymod:select
+
+    stop_criterion:
+      metric: mse
+      threshold: 0.1
+      operator: "<"
+      evaluator:
+        type: python
+        function: mymod:eval_mse
+""")
+
+
+def test_sequential_spec_roundtrip(tmp_path):
+    p = tmp_path / "spec.yaml"
+    p.write_text(SEQ_YAML)
+    cfg = WorkflowConfig.from_yaml(p)
+    assert cfg.learner.type == "sequential_active_learner"
+    assert cfg.learner.max_iter == 5
+    assert cfg.simulation.command == "python sim.py"
+    assert cfg.training.function == "mymod:train"
+    assert cfg.stop_criterion.metric == "mse"
+    assert cfg.tasks == {
+        "simulation": cfg.simulation,
+        "training": cfg.training,
+        "active_learn": cfg.active_learn,
+    }
+
+
+def test_sequential_spec_missing_slot(tmp_path):
+    yaml = textwrap.dedent("""\
+        learner:
+          type: sequential_active_learner
+        simulation:
+          type: shell
+          command: python sim.py
+        training:
+          type: python
+          function: mymod:train
+        stop_criterion:
+          metric: mse
+          threshold: 0.1
+          evaluator:
+            type: python
+            function: mymod:eval
+    """)
+    p = tmp_path / "bad.yaml"
+    p.write_text(yaml)
+    with pytest.raises(ValueError, match="active_learn"):
+        WorkflowConfig.from_yaml(p)
+
+
+def test_sequential_spec_extra_slot(tmp_path):
+    yaml = textwrap.dedent("""\
+        learner:
+          type: sequential_active_learner
+        simulation:
+          type: shell
+          command: python sim.py
+        training:
+          type: python
+          function: mymod:train
+        active_learn:
+          type: python
+          function: mymod:select
+        environment:
+          type: shell
+          command: python env.py
+        stop_criterion:
+          metric: mse
+          threshold: 0.1
+          evaluator:
+            type: python
+            function: mymod:eval
+    """)
+    p = tmp_path / "bad.yaml"
+    p.write_text(yaml)
+    with pytest.raises(ValueError, match="Unexpected"):
+        WorkflowConfig.from_yaml(p)
+
+
+def test_unknown_learner_type(tmp_path):
+    yaml = textwrap.dedent("""\
+        learner:
+          type: unknown_learner
+        simulation:
+          type: shell
+          command: x
+        stop_criterion:
+          metric: mse
+          threshold: 0.1
+          evaluator:
+            type: shell
+            command: x
+    """)
+    p = tmp_path / "bad.yaml"
+    p.write_text(yaml)
+    with pytest.raises(ValueError, match="Unknown learner type"):
+        WorkflowConfig.from_yaml(p)
+
+
+# ── WorkflowConfig — parallel learner with learners ──────────────────────────────
+
+PAR_YAML = textwrap.dedent("""\
+    learner:
+      type: parallel_active_learner
+      max_iter: 3
+
+    learners:
+      - label: rf
+        simulation:
+          type: shell
+          command: python sim.py --model rf
+        training:
+          type: python
+          function: mymod:train_rf
+        active_learn:
+          type: python
+          function: mymod:select
+      - label: mlp
+        simulation:
+          type: shell
+          command: python sim.py --model mlp
+        training:
+          type: python
+          function: mymod:train_mlp
+        active_learn:
+          type: python
+          function: mymod:select
+
+    stop_criterion:
+      metric: r2
+      threshold: 0.9
+      operator: ">"
+      evaluator:
+        type: python
+        function: mymod:eval_r2
+""")
+
+
+def test_parallel_learners_roundtrip(tmp_path):
+    p = tmp_path / "par.yaml"
+    p.write_text(PAR_YAML)
+    cfg = WorkflowConfig.from_yaml(p)
+    assert len(cfg.learners) == 2
+    assert cfg.learners[0].label == "rf"
+    assert cfg.learners[1].tasks["training"].function == "mymod:train_mlp"
+
+
+def test_parallel_learner_missing_slot(tmp_path):
+    yaml = textwrap.dedent("""\
+        learner:
+          type: parallel_active_learner
+        learners:
+          - label: rf
+            simulation:
+              type: shell
+              command: python sim.py
+            training:
+              type: python
+              function: mymod:train
+        stop_criterion:
+          metric: r2
+          threshold: 0.9
+          evaluator:
+            type: python
+            function: mymod:eval
+    """)
+    p = tmp_path / "bad.yaml"
+    p.write_text(yaml)
+    with pytest.raises(ValueError, match="active_learn"):
+        WorkflowConfig.from_yaml(p)
+
+
+# ── WorkflowConfig — parameters block ────────────────────────────────────────────
+
+# ── WorkflowConfig — reserved parameter keys ──────────────────────────────────────
+
+# ── WorkflowConfig — task_description consistency across parallel learners ────────
+
+
+def test_parallel_learners_identical_task_description_valid(tmp_path):
+    yaml = textwrap.dedent("""\
+        learner:
+          type: parallel_active_learner
+        learners:
+          - label: a
+            simulation:
+              type: shell
+              command: python sim.py
+              task_description:
+                cpu_count: 4
+            training:
+              type: python
+              function: mymod:train
+            active_learn:
+              type: python
+              function: mymod:select
+          - label: b
+            simulation:
+              type: shell
+              command: python sim.py --model b
+              task_description:
+                cpu_count: 4
+            training:
+              type: python
+              function: mymod:train
+            active_learn:
+              type: python
+              function: mymod:select
+        stop_criterion:
+          metric: r2
+          threshold: 0.9
+          evaluator:
+            type: python
+            function: mymod:eval
+    """)
+    p = tmp_path / "ok.yaml"
+    p.write_text(yaml)
+    cfg = WorkflowConfig.from_yaml(p)
+    assert len(cfg.learners) == 2
+
+
+def test_parallel_learners_nested_task_description_different_key_order_valid(tmp_path):
+    """Nested dicts with the same keys/values in a different insertion order must compare equal —
+    dict equality, not string-sorted serialization."""
+    yaml = textwrap.dedent("""\
+        learner:
+          type: parallel_active_learner
+        learners:
+          - label: a
+            simulation:
+              type: shell
+              command: python sim.py
+              task_description:
+                resources:
+                  gpus: 1
+                  cpus: 4
+            training:
+              type: python
+              function: mymod:train
+            active_learn:
+              type: python
+              function: mymod:select
+          - label: b
+            simulation:
+              type: shell
+              command: python sim.py --model b
+              task_description:
+                resources:
+                  cpus: 4
+                  gpus: 1
+            training:
+              type: python
+              function: mymod:train
+            active_learn:
+              type: python
+              function: mymod:select
+        stop_criterion:
+          metric: r2
+          threshold: 0.9
+          evaluator:
+            type: python
+            function: mymod:eval
+    """)
+    p = tmp_path / "ok_nested.yaml"
+    p.write_text(yaml)
+    cfg = WorkflowConfig.from_yaml(p)
+    assert len(cfg.learners) == 2
+
+
+def test_parallel_learners_different_task_description_raises(tmp_path):
+    yaml = textwrap.dedent("""\
+        learner:
+          type: parallel_active_learner
+        learners:
+          - label: a
+            simulation:
+              type: shell
+              command: python sim.py
+              task_description:
+                cpu_count: 4
+            training:
+              type: python
+              function: mymod:train
+            active_learn:
+              type: python
+              function: mymod:select
+          - label: b
+            simulation:
+              type: shell
+              command: python sim.py
+              task_description:
+                cpu_count: 8
+            training:
+              type: python
+              function: mymod:train
+            active_learn:
+              type: python
+              function: mymod:select
+        stop_criterion:
+          metric: r2
+          threshold: 0.9
+          evaluator:
+            type: python
+            function: mymod:eval
+    """)
+    p = tmp_path / "bad.yaml"
+    p.write_text(yaml)
+    with pytest.raises(ValueError, match="task_description"):
+        WorkflowConfig.from_yaml(p)
+
+
+def test_parallel_learners_one_has_task_description_other_absent_raises(tmp_path):
+    yaml = textwrap.dedent("""\
+        learner:
+          type: parallel_active_learner
+        learners:
+          - label: a
+            simulation:
+              type: shell
+              command: python sim.py
+              task_description:
+                cpu_count: 4
+            training:
+              type: python
+              function: mymod:train
+            active_learn:
+              type: python
+              function: mymod:select
+          - label: b
+            simulation:
+              type: shell
+              command: python sim.py
+            training:
+              type: python
+              function: mymod:train
+            active_learn:
+              type: python
+              function: mymod:select
+        stop_criterion:
+          metric: r2
+          threshold: 0.9
+          evaluator:
+            type: python
+            function: mymod:eval
+    """)
+    p = tmp_path / "bad.yaml"
+    p.write_text(yaml)
+    with pytest.raises(ValueError, match="task_description"):
+        WorkflowConfig.from_yaml(p)
+
+
+def test_parameters_reserved_key_pythonpath(tmp_path):
+    yaml = textwrap.dedent("""\
+        learner:
+          type: sequential_active_learner
+          max_iter: 1
+        simulation:
+          type: python
+          function: mymod:sim
+        training:
+          type: python
+          function: mymod:train
+        active_learn:
+          type: python
+          function: mymod:select
+        stop_criterion:
+          metric: mse
+          threshold: 0.1
+          evaluator:
+            type: python
+            function: mymod:eval
+        parameters:
+          pythonpath: /some/path
+    """)
+    p = tmp_path / "bad.yaml"
+    p.write_text(yaml)
+    with pytest.raises(ValueError, match="reserved keys"):
+        WorkflowConfig.from_yaml(p)
+
+
+def test_parameters_reserved_key_iteration(tmp_path):
+    yaml = textwrap.dedent("""\
+        learner:
+          type: sequential_active_learner
+          max_iter: 1
+        simulation:
+          type: python
+          function: mymod:sim
+        training:
+          type: python
+          function: mymod:train
+        active_learn:
+          type: python
+          function: mymod:select
+        stop_criterion:
+          metric: mse
+          threshold: 0.1
+          evaluator:
+            type: python
+            function: mymod:eval
+        parameters:
+          iteration: 5
+    """)
+    p = tmp_path / "bad.yaml"
+    p.write_text(yaml)
+    with pytest.raises(ValueError, match="reserved keys"):
+        WorkflowConfig.from_yaml(p)
+
+
+def test_parameters_reserved_key_learner_id(tmp_path):
+    yaml = textwrap.dedent("""\
+        learner:
+          type: sequential_active_learner
+          max_iter: 1
+        simulation:
+          type: python
+          function: mymod:sim
+        training:
+          type: python
+          function: mymod:train
+        active_learn:
+          type: python
+          function: mymod:select
+        stop_criterion:
+          metric: mse
+          threshold: 0.1
+          evaluator:
+            type: python
+            function: mymod:eval
+        parameters:
+          learner_id: 0
+    """)
+    p = tmp_path / "bad.yaml"
+    p.write_text(yaml)
+    with pytest.raises(ValueError, match="reserved keys"):
+        WorkflowConfig.from_yaml(p)
+
+
+def test_parameters_non_reserved_key_allowed(tmp_path):
+    yaml = textwrap.dedent("""\
+        learner:
+          type: sequential_active_learner
+          max_iter: 1
+        simulation:
+          type: python
+          function: mymod:sim
+        training:
+          type: python
+          function: mymod:train
+        active_learn:
+          type: python
+          function: mymod:select
+        stop_criterion:
+          metric: mse
+          threshold: 0.1
+          evaluator:
+            type: python
+            function: mymod:eval
+        parameters:
+          dataset: my_ds
+    """)
+    p = tmp_path / "ok.yaml"
+    p.write_text(yaml)
+    cfg = WorkflowConfig.from_yaml(p)
+    assert cfg.parameters == {"dataset": "my_ds"}
+
+
+def test_parameters_roundtrip(tmp_path):
+    yaml = textwrap.dedent("""\
+        learner:
+          type: sequential_active_learner
+          max_iter: 5
+        simulation:
+          type: python
+          function: mymod:sim
+        training:
+          type: python
+          function: mymod:train
+        active_learn:
+          type: python
+          function: mymod:select
+        stop_criterion:
+          metric: mse
+          threshold: 0.1
+          evaluator:
+            type: python
+            function: mymod:eval
+        parameters:
+          dataset: my_ds
+          scale: 1.5
+          growing_pool: false
+    """)
+    p = tmp_path / "params.yaml"
+    p.write_text(yaml)
+    cfg = WorkflowConfig.from_yaml(p)
+    assert cfg.parameters == {"dataset": "my_ds", "scale": 1.5, "growing_pool": False}
+
+
+def test_parameters_defaults_empty(tmp_path):
+    p = tmp_path / "spec.yaml"
+    p.write_text(SEQ_YAML)
+    cfg = WorkflowConfig.from_yaml(p)
+    assert cfg.parameters == {}
+
+
+# ── WorkflowConfig — parallel learners (mixed types) ─────────────────────────────
+
+
+def test_parallel_learners_mixed_types(tmp_path):
+    yaml = textwrap.dedent("""\
+        learner:
+          type: parallel_active_learner
+        learners:
+          - label: a
+            simulation:
+              type: shell
+              command: python sim.py
+            training:
+              type: python
+              function: mymod:train_a
+            active_learn:
+              type: python
+              function: mymod:select
+          - label: b
+            simulation:
+              type: python
+              function: mymod:sim_b
+            training:
+              type: python
+              function: mymod:train_b
+            active_learn:
+              type: python
+              function: mymod:select
+        stop_criterion:
+          metric: r2
+          threshold: 0.9
+          evaluator:
+            type: python
+            function: mymod:eval
+    """)
+    p = tmp_path / "bad.yaml"
+    p.write_text(yaml)
+    with pytest.raises(ValueError, match="mixed types"):
+        WorkflowConfig.from_yaml(p)
diff --git a/tests/unit/spec/test_workflow_spec.py b/tests/unit/spec/test_workflow_spec.py
new file mode 100644
index 00000000..a9e7fab8
--- /dev/null
+++ b/tests/unit/spec/test_workflow_spec.py
@@ -0,0 +1,252 @@
+"""Unit tests for WorkflowSpec — load_spec and workflow_with()."""
+
+import textwrap
+
+import pytest
+
+from rose.spec import WorkflowSpec, load_spec
+
+SEQ_YAML = textwrap.dedent("""\
+    learner:
+      type: sequential_active_learner
+      max_iter: 5
+
+    simulation:
+      type: python
+      function: mymod:sim
+
+    training:
+      type: python
+      function: mymod:train
+
+    active_learn:
+      type: python
+      function: mymod:select
+
+    stop_criterion:
+      metric: mse
+      threshold: 0.1
+      evaluator:
+        type: python
+        function: mymod:eval
+
+    parameters:
+      dataset: base_ds
+      scale: 1.0
+""")
+
+
+def _make_spec(tmp_path, yaml=SEQ_YAML):
+    p = tmp_path / "spec.yaml"
+    p.write_text(yaml)
+    return load_spec(p)
+
+
+# ── workflow_with: learner field override ─────────────────────────────────────
+
+
+def test_workflow_with_max_iter(tmp_path):
+    spec = _make_spec(tmp_path)
+    new = spec.workflow_with(max_iter=2)
+    assert new.config.learner.max_iter == 2
+    assert spec.config.learner.max_iter == 5  # original unchanged
+
+
+def test_workflow_with_does_not_mutate_original(tmp_path):
+    spec = _make_spec(tmp_path)
+    _ = spec.workflow_with(max_iter=1)
+    assert spec.config.learner.max_iter == 5
+
+
+# ── workflow_with: parameters merge ──────────────────────────────────────────
+
+
+def test_workflow_with_parameters_merges(tmp_path):
+    spec = _make_spec(tmp_path)
+    new = spec.workflow_with(parameters={"dataset": "test_ds"})
+    assert new.config.parameters["dataset"] == "test_ds"
+    assert new.config.parameters["scale"] == 1.0  # existing key preserved
+
+
+def test_workflow_with_parameters_adds_new_key(tmp_path):
+    spec = _make_spec(tmp_path)
+    new = spec.workflow_with(parameters={"lr": 0.001})
+    assert new.config.parameters["lr"] == 0.001
+    assert new.config.parameters["dataset"] == "base_ds"
+
+
+def test_workflow_with_parameters_does_not_mutate_original(tmp_path):
+    spec = _make_spec(tmp_path)
+    _ = spec.workflow_with(parameters={"dataset": "other"})
+    assert spec.config.parameters["dataset"] == "base_ds"
+
+
+# ── workflow_with: combined overrides ─────────────────────────────────────────
+
+
+def test_workflow_with_combined(tmp_path):
+    spec = _make_spec(tmp_path)
+    new = spec.workflow_with(max_iter=3, parameters={"dataset": "test_ds", "scale": 2.0})
+    assert new.config.learner.max_iter == 3
+    assert new.config.parameters["dataset"] == "test_ds"
+    assert new.config.parameters["scale"] == 2.0
+
+
+# ── workflow_with: unknown field raises ───────────────────────────────────────
+
+
+def test_workflow_with_unknown_field_raises(tmp_path):
+    spec = _make_spec(tmp_path)
+    with pytest.raises(ValueError, match="unknown spec field"):
+        spec.workflow_with(nonexistent_key=42)
+
+
+# ── workflow_with: returns WorkflowSpec with callable .workflow ───────────────
+
+
+def test_workflow_with_returns_workflow_spec(tmp_path):
+    spec = _make_spec(tmp_path)
+    new = spec.workflow_with(max_iter=1)
+    assert isinstance(new, WorkflowSpec)
+    assert callable(new.workflow)
+
+
+# ── validate_imports ──────────────────────────────────────────────────────────
+
+IMPORTABLE_YAML = textwrap.dedent("""\
+    learner:
+      type: sequential_active_learner
+      max_iter: 1
+
+    simulation:
+      type: python
+      function: os.path:join
+
+    training:
+      type: python
+      function: os.path:dirname
+
+    active_learn:
+      type: python
+      function: os.path:basename
+
+    stop_criterion:
+      metric: mse
+      threshold: 0.1
+      evaluator:
+        type: python
+        function: os.path:exists
+""")
+
+BAD_MODULE_YAML = textwrap.dedent("""\
+    learner:
+      type: sequential_active_learner
+      max_iter: 1
+
+    simulation:
+      type: python
+      function: no_such_module_xyz:fn
+
+    training:
+      type: python
+      function: os.path:dirname
+
+    active_learn:
+      type: python
+      function: os.path:basename
+
+    stop_criterion:
+      metric: mse
+      threshold: 0.1
+      evaluator:
+        type: python
+        function: os.path:exists
+""")
+
+BAD_ATTR_YAML = textwrap.dedent("""\
+    learner:
+      type: sequential_active_learner
+      max_iter: 1
+
+    simulation:
+      type: python
+      function: os.path:no_such_fn_xyz
+
+    training:
+      type: python
+      function: os.path:dirname
+
+    active_learn:
+      type: python
+      function: os.path:basename
+
+    stop_criterion:
+      metric: mse
+      threshold: 0.1
+      evaluator:
+        type: python
+        function: os.path:exists
+""")
+
+MULTI_BAD_YAML = textwrap.dedent("""\
+    learner:
+      type: sequential_active_learner
+      max_iter: 1
+
+    simulation:
+      type: python
+      function: no_such_module_xyz:fn
+
+    training:
+      type: python
+      function: os.path:no_such_fn_xyz
+
+    active_learn:
+      type: python
+      function: os.path:basename
+
+    stop_criterion:
+      metric: mse
+      threshold: 0.1
+      evaluator:
+        type: python
+        function: os.path:exists
+""")
+
+
+def test_validate_imports_succeeds_for_importable_functions(tmp_path):
+    p = tmp_path / "spec.yaml"
+    p.write_text(IMPORTABLE_YAML)
+    spec = load_spec(p, validate_imports=True)
+    assert spec is not None
+
+
+def test_validate_imports_raises_on_bad_module(tmp_path):
+    p = tmp_path / "spec.yaml"
+    p.write_text(BAD_MODULE_YAML)
+    with pytest.raises(ValueError, match="no_such_module_xyz:fn"):
+        load_spec(p, validate_imports=True)
+
+
+def test_validate_imports_raises_on_bad_attribute(tmp_path):
+    p = tmp_path / "spec.yaml"
+    p.write_text(BAD_ATTR_YAML)
+    with pytest.raises(ValueError, match="no_such_fn_xyz"):
+        load_spec(p, validate_imports=True)
+
+
+def test_validate_imports_reports_all_failures(tmp_path):
+    p = tmp_path / "spec.yaml"
+    p.write_text(MULTI_BAD_YAML)
+    with pytest.raises(ValueError) as exc_info:
+        load_spec(p, validate_imports=True)
+    msg = str(exc_info.value)
+    assert "no_such_module_xyz:fn" in msg
+    assert "no_such_fn_xyz" in msg
+
+
+def test_validate_imports_default_false_skips_check(tmp_path):
+    p = tmp_path / "spec.yaml"
+    p.write_text(BAD_MODULE_YAML)
+    spec = load_spec(p)  # validate_imports=False by default — should not raise
+    assert spec is not None