feat(datasets): dataset_load — execute the loading plan by dataset_id in one call

Closes the High step-budget item: both measured 12-step runs burned 3-5
steps picking a loading entry point and 2 more collapsing probes because
they used the uncollapsed URL. The plan-side fix (7944fee) told the
agent HOW to load; this makes loading ONE tool call.

dataset_load(dataset_id) resolves the manifest, builds the loading plan
(_build_loading_plan stays the single source of truth), and executes its
get-the-data-open prefix — correct loader per source type, precomputed
collapsed_url, clinical join, curated-sample filter — threading the h5ad
path between steps. Returns the analysis-ready adata_path plus
analysis_path/data_level/default_contrast and per-step provenance; a
mid-plan failure surfaces the executed steps and last good intermediate.
Loader executors import lazily (scanpy only for Path P manifests).

Prompt efficiency rule 7 now mandates dataset_load as the first call for
any adata_path tool (dataset_plan_analysis/describe reserved for 'how
would you analyze' questions), and the GEO loader's next_step no longer
unconditionally tells the agent to run decoupler_inspect_data — the
last holdout of the rule-4 contradiction (Med item, same lane).

Live-verified: dataset_load('gse71989_chen') → one step, authenticated
collapsed-h5ad download, 22x21655, Path B + limma contrast.
7 tests in tests/test_dataset_load_tool.py; suite 1426 passed / 60
skipped.

prompts.yaml

@@ -203,7 +203,7 @@ The ONLY datasets registered in this system are listed below with everything\
   \ ## Available Datasets. Skip straight to DE or the relevant analysis tool.\n\n5. **Always pass `subset_query` when the default contrast specifies one.** The ## Available\
   \ Datasets section lists each dataset's default contrasts, including any `subset_query`. When running DE,\
   \ read that value and pass it directly — never pre-filter the AnnData manually. Example: if the contrast\
-  \ entry shows `subset_query=\"tumor_subtype != ''\"`, pass that exact string to `decoupler_differential_expression`.\n\n6. **Never import from `tools`, `src.tools`, `server`, or any `_mcp` module, and never inspect `FunctionTool`/`.fn`/`.run()` internals.** All functions listed below are already pre-loaded as plain callables in your namespace and are re-verified before every step. If a call to one of them raises `NameError` or `'FunctionTool' object is not callable`, do NOT start importing or introspecting — simply retry the exact same call with the parameters documented below. If it still fails after one retry, move on to a different approach rather than reverse-engineering the tool wrapper.\n\n7. **Never guess how to load a registered dataset — ask for its loading plan.** FIRST check whether the tool you need already takes a `dataset_id` (e.g. `dataset_score_bulk_samples`, `dataset_compare_activity_by_group`, `dataset_score_signature`, `dataset_get_integration_plan`). **Those tools load the dataset themselves — call them directly. Do NOT call `dataset_plan_analysis` and do NOT load the data first; that wastes two steps.** Only when you need a tool that takes an `adata_path` (e.g. `decoupler_differential_expression`) do you need the data on disk — and then your first call is `dataset_plan_analysis(dataset_id=..., user_question=...)` (or `dataset_describe`). It returns the exact loader, the exact arguments, and the correct URL to use. Do NOT open a dataset by pattern-matching its URL out of ## Available Datasets and picking a loader that looks right: `decoupler_load_geo_series_matrix` does not read h5ad, a bare `urllib`/`requests` download cannot authenticate to the private data repo, and `ad.read_h5ad` on a URL is not a local path. Each wrong guess costs a step, and the plan you skipped names the right call. In particular, when a dataset has a **collapsed** variant of its file the plan points at that URL — using the uncollapsed one forces two extra steps (`decoupler_annotate_probes_with_gpl` + `decoupler_collapse_probes_to_genes`) that the plan avoids entirely.\n\n8. **Load each dataset exactly once per turn.** The loaded AnnData is written to `output_path` and cached. If you already loaded a dataset this turn, reuse that path — never re-download or re-load it to 'check' something.\n\n## Available Functions\n\nYou have access to the following functions. These functions are already available\
+  \ entry shows `subset_query=\"tumor_subtype != ''\"`, pass that exact string to `decoupler_differential_expression`.\n\n6. **Never import from `tools`, `src.tools`, `server`, or any `_mcp` module, and never inspect `FunctionTool`/`.fn`/`.run()` internals.** All functions listed below are already pre-loaded as plain callables in your namespace and are re-verified before every step. If a call to one of them raises `NameError` or `'FunctionTool' object is not callable`, do NOT start importing or introspecting — simply retry the exact same call with the parameters documented below. If it still fails after one retry, move on to a different approach rather than reverse-engineering the tool wrapper.\n\n7. **Never guess how to load a registered dataset — ask for its loading plan.** FIRST check whether the tool you need already takes a `dataset_id` (e.g. `dataset_score_bulk_samples`, `dataset_compare_activity_by_group`, `dataset_score_signature`, `dataset_get_integration_plan`). **Those tools load the dataset themselves — call them directly. Do NOT call `dataset_plan_analysis` and do NOT load the data first; that wastes two steps.** Only when you need a tool that takes an `adata_path` (e.g. `decoupler_differential_expression`, `decoupler_pseudobulk_aggregate`, the metadata tools) do you need the data on disk — and then your first call is `dataset_load(dataset_id=...)`. It EXECUTES the whole loading plan in one step (the correct loader for the source type, the precomputed collapsed URL, the clinical join, the curated-sample filter) and returns the analysis-ready `adata_path` plus the analysis path, data_level and default contrast — pass that path straight to the analysis tool. Do NOT hand-pick loaders, do NOT call `dataset_plan_analysis`/`dataset_describe` first just to load (use those only when the user asks how a dataset WOULD be analyzed), and do NOT re-load a dataset already loaded this turn. Do NOT open a dataset by pattern-matching its URL out of ## Available Datasets and picking a loader that looks right: `decoupler_load_geo_series_matrix` does not read h5ad, a bare `urllib`/`requests` download cannot authenticate to the private data repo, and `ad.read_h5ad` on a URL is not a local path. Each wrong guess costs a step, and `dataset_load` already makes the right calls — including using a dataset's **collapsed** file variant when one exists (loading the uncollapsed one instead forces two extra steps, `decoupler_annotate_probes_with_gpl` + `decoupler_collapse_probes_to_genes`).\n\n8. **Load each dataset exactly once per turn.** The loaded AnnData is written to `output_path` and cached. If you already loaded a dataset this turn, reuse that path — never re-download or re-load it to 'check' something.\n\n## Available Functions\n\nYou have access to the following functions. These functions are already available\
   \ in your Python environment and can be called directly:\n\n{% for func_name, schema in functions.items() %}\n**{{ schema.function.name\
   \ }}({% for param_name in schema.function.parameters.properties.keys() %}{{ param_name }}{{ \", \" if not loop.last }}{%\
   \ endfor %})**\n- Description: {{ schema.function.description }}\n- Parameters:\n  {% for param_name, param_info in schema.function.parameters.properties.items()\

src/tools/dataset_tools/catalog.py

@@ -3,6 +3,7 @@
 
 from ._base import *  # noqa: F401
 from ._base import (  # noqa: F401
+    _LOADING_PLAN_TOOLS,
     _build_loading_plan,
     _classify_metadata_values,
     _col_semantics,
@@ -202,6 +203,169 @@ def dataset_get_integration_plan(
         }
 
 
+def _loading_step_executors() -> dict:
+    """Map loading-plan tool names to their callables, imported lazily.
+
+    Lazy so this module does not pull in the heavy loader stacks (scanpy for
+    the sc loader) at import time; the tool decorators leave these as plain
+    callable functions.
+    """
+    from src.tools.bulk_rnaseq.tools import (
+        decoupler_join_clinical_metadata,
+        decoupler_load_gdc_star_counts,
+        decoupler_load_url_counts,
+    )
+    from src.tools.rna.loaders import (
+        decoupler_annotate_probes_with_gpl,
+        decoupler_collapse_probes_to_genes,
+        decoupler_load_geo_series_matrix,
+    )
+
+    executors = {
+        "decoupler_load_url_counts": decoupler_load_url_counts,
+        "decoupler_load_gdc_star_counts": decoupler_load_gdc_star_counts,
+        "decoupler_join_clinical_metadata": decoupler_join_clinical_metadata,
+        "decoupler_load_geo_series_matrix": decoupler_load_geo_series_matrix,
+        "decoupler_annotate_probes_with_gpl": decoupler_annotate_probes_with_gpl,
+        "decoupler_collapse_probes_to_genes": decoupler_collapse_probes_to_genes,
+        "dataset_filter_to_curated_samples": dataset_filter_to_curated_samples,
+    }
+    try:  # scanpy stack — only needed for Path P manifests
+        from src.tools.rna_sc import decoupler_load_and_visualize_data
+
+        executors["decoupler_load_and_visualize_data"] = decoupler_load_and_visualize_data
+    except Exception:
+        pass
+    return executors
+
+
+@dataset_mcp.tool
+def dataset_load(
+    dataset_id: Annotated[
+        str,
+        "The registered dataset to load, e.g. 'gse71729_moffitt' or 'tcga_paad'. "
+        "Call dataset_list_available() first only if you are unsure of the ID.",
+    ],
+) -> dict:
+    """
+    Load a registered dataset end-to-end and return the analysis-ready h5ad path.
+
+    This EXECUTES the manifest's loading plan in one call — the correct loader
+    for the source type (url / GEO series matrix / GDC / hosted h5ad), the
+    precomputed collapsed_url when one exists, the clinical-metadata join, and
+    the curated-sample filter — so you never pick an entry point yourself.
+
+    Call this as the FIRST step whenever a registered dataset must be on disk
+    for a tool that takes an adata_path (decoupler_differential_expression,
+    decoupler_pseudobulk_aggregate, metadata tools). Do NOT use it for tools
+    that already take a dataset_id (dataset_score_bulk_samples,
+    dataset_compare_activity_by_group) — those load the data themselves.
+
+    Returns adata_path (pass it straight to the next tool), the analysis path
+    (A = DESeq2 on raw counts, B = limma/ttest on normalized data, P =
+    single-cell: pseudobulk before any DE), the manifest's default contrast,
+    and the executed steps for provenance.
+    """
+    try:
+        manifest = load_manifest(dataset_id)
+    except KeyError as exc:
+        return {
+            "error": str(exc),
+            "available_datasets": [d["dataset_id"] for d in list_available_datasets()],
+        }
+
+    try:
+        plan = _build_loading_plan(manifest)
+    except Exception as exc:
+        return {"error": f"Could not build the loading plan for '{dataset_id}': {exc}"}
+
+    executors = _loading_step_executors()
+    executed: list[dict] = []
+    adata_path: str | None = None
+
+    for raw in plan:
+        tool = raw.get("tool")
+        if tool not in _LOADING_PLAN_TOOLS:
+            continue  # analysis tail — owned by the caller, not the loader
+        fn = executors.get(tool)
+        if fn is None:
+            return {
+                "error": (
+                    f"No executor for loading step '{tool}' — follow the "
+                    f"loading_plan manually from dataset_describe('{dataset_id}')."
+                ),
+                "steps_executed": executed,
+            }
+        args = dict(raw.get("key_args") or {})
+        # Thread the running h5ad path through placeholder args
+        # ("<output_path from step N>") emitted by _build_loading_plan.
+        for k, v in args.items():
+            if isinstance(v, str) and v.startswith("<"):
+                args[k] = adata_path
+        try:
+            res = fn(**args)
+        except Exception as exc:
+            return {
+                "error": f"Loading step '{tool}' failed: {exc}",
+                "steps_executed": executed,
+                "adata_path": adata_path,
+                "note": (
+                    "Steps executed so far succeeded; adata_path (if set) is the "
+                    "last good intermediate. Consult dataset_describe for the "
+                    "remaining plan rather than guessing a loader."
+                ),
+            }
+        if isinstance(res, dict) and res.get("error"):
+            return {
+                "error": f"Loading step '{tool}' failed: {res['error']}",
+                "steps_executed": executed,
+                "adata_path": adata_path,
+            }
+        out = res.get("output_path") if isinstance(res, dict) else None
+        if out:
+            adata_path = out
+        executed.append({"tool": tool, "output_path": out})
+
+    if adata_path is None:
+        return {
+            "error": (
+                f"The loading plan for '{dataset_id}' produced no h5ad — "
+                f"source type '{manifest.expression_source.get('type')}' may be "
+                "unsupported. Use dataset_describe and load manually."
+            ),
+            "steps_executed": executed,
+        }
+
+    contrast = get_contrast_groups(manifest)
+    path = manifest.analysis_path
+    de_method = {
+        "A": "deseq2 (raw integer counts)",
+        "B": "limma or ttest (pre-normalized data — never deseq2)",
+        "P": "NONE directly — pseudobulk first (decoupler_pseudobulk_aggregate), then Path A/B",
+    }.get(path, "see data_level")
+
+    return {
+        "dataset_id": dataset_id,
+        "adata_path": adata_path,
+        "analysis_path": path,
+        "data_level": manifest.data_level,
+        "recommended_de_method": de_method,
+        "default_contrast": contrast,
+        "steps_executed": executed,
+        "next_step": (
+            "The dataset is loaded and analysis-ready — pass adata_path directly "
+            "to the analysis tool. Do NOT call decoupler_inspect_data (the "
+            "manifest already fixes data_level) and do NOT reload the data."
+            + (
+                " Path P: aggregate to pseudobulk before any DE — never "
+                "DESeq2/limma on per-cell counts."
+                if path == "P"
+                else ""
+            )
+        ),
+    }
+
+
 @dataset_mcp.tool
 def dataset_filter_to_curated_samples(
     adata_path: Annotated[

src/tools/rna/loaders.py

@@ -265,8 +265,10 @@ def decoupler_load_geo_series_matrix(
             "var_index_sample": list(adata.var.index[:5]),
             "output_path": str(output_path.resolve()),
             "next_step": (
-                "Run decoupler_inspect_data(adata_path=output_path) to determine data type "
-                "and recommended analysis path."
+                "If this file is NOT a registered dataset, run "
+                "decoupler_inspect_data(adata_path=output_path) to determine data type and "
+                "analysis path. For a REGISTERED dataset, skip it — data_level and "
+                "analysis_path are manifest facts; go straight to the analysis tool."
             ),
             "artifacts": [
                 {
@@ -348,8 +350,10 @@ def decoupler_load_geo_series_matrix(
         "var_index_sample": list(adata.var.index[:5]),
         "output_path": str(output_path.resolve()),
         "next_step": (
-            "Run decoupler_inspect_data(adata_path=output_path) to determine data type "
-            "and recommended analysis path."
+            "If this file is NOT a registered dataset, run "
+            "decoupler_inspect_data(adata_path=output_path) to determine data type and "
+            "analysis path. For a REGISTERED dataset, skip it — data_level and "
+            "analysis_path are manifest facts; go straight to the analysis tool."
         ),
         "artifacts": [
             {

tests/test_dataset_load_tool.py

@@ -0,0 +1,135 @@
+"""
+Tests for dataset_load — the registry-aware load-by-dataset_id executor.
+
+The 2026-08-04 step-budget measurement showed runs burning 3-5 steps picking a
+loading entry point (and 2 more collapsing probes when the uncollapsed URL was
+used despite a precomputed collapsed_url). dataset_load executes the manifest's
+loading plan in ONE tool call: correct loader for the source type, collapsed
+URL when present, clinical join, curated-sample filter — and returns the
+analysis-ready h5ad path plus routing facts.
+
+Loader executors are monkeypatched (no network); the plans are built from the
+real registry manifests so the wiring under test is the deployed wiring.
+"""
+
+from __future__ import annotations
+
+import sys
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).parent.parent))
+
+import src.tools.dataset_tools.catalog as catalog  # noqa: E402
+from src.datasets.registry import load_manifest  # noqa: E402
+from src.tools.dataset_tools._base import _LOADING_PLAN_TOOLS, _build_loading_plan  # noqa: E402
+
+
+def _fake_executors(calls, fail_on=None):
+    """Executor map whose loaders record calls and return sequential paths."""
+
+    def make(tool):
+        def run(**kwargs):
+            if tool == fail_on:
+                raise RuntimeError("boom")
+            calls.append((tool, kwargs))
+            return {"output_path": f"/tmp/fake/{tool}_{len(calls)}.h5ad"}
+
+        return run
+
+    return {t: make(t) for t in _LOADING_PLAN_TOOLS}
+
+
+def test_unknown_dataset_returns_available_list():
+    res = catalog.dataset_load("no_such_dataset")
+    assert "error" in res
+    assert "gse71729_moffitt" in res["available_datasets"]
+
+
+def test_executes_plan_in_order_and_threads_paths(monkeypatch):
+    calls = []
+    monkeypatch.setattr(catalog, "_loading_step_executors", lambda: _fake_executors(calls))
+    res = catalog.dataset_load("gse71729_moffitt")
+
+    assert "error" not in res
+    plan_tools = [
+        s["tool"]
+        for s in _build_loading_plan(load_manifest("gse71729_moffitt"))
+        if s["tool"] in _LOADING_PLAN_TOOLS
+    ]
+    assert [t for t, _ in calls] == plan_tools
+    assert [s["tool"] for s in res["steps_executed"]] == plan_tools
+    # The final adata_path is the last step's output.
+    assert res["adata_path"] == f"/tmp/fake/{plan_tools[-1]}_{len(calls)}.h5ad"
+    # No placeholder strings may reach a loader.
+    for _, kwargs in calls:
+        for v in kwargs.values():
+            assert not (isinstance(v, str) and v.startswith("<")), (
+                f"placeholder leaked into loader args: {v}"
+            )
+    # Routing facts come from the manifest.
+    assert res["analysis_path"] in ("A", "B", "P")
+    assert res["data_level"]
+    assert "design_factor" in res["default_contrast"]
+    assert "decoupler_inspect_data" in res["next_step"]  # explicitly discouraged
+
+
+def test_curated_dataset_includes_curation_step(monkeypatch):
+    """tcga_paad has a curated_sample_list — the executor must run the filter."""
+    calls = []
+    monkeypatch.setattr(catalog, "_loading_step_executors", lambda: _fake_executors(calls))
+    res = catalog.dataset_load("tcga_paad")
+    tools = [t for t, _ in calls]
+    assert "dataset_filter_to_curated_samples" in tools
+    assert res["analysis_path"] == "A"
+    assert "deseq2" in res["recommended_de_method"]
+
+
+def test_path_b_dataset_recommends_non_deseq2(monkeypatch):
+    calls = []
+    monkeypatch.setattr(catalog, "_loading_step_executors", lambda: _fake_executors(calls))
+    res = catalog.dataset_load("gse71729_moffitt")
+    assert res["analysis_path"] == "B"
+    assert "deseq2" not in res["recommended_de_method"].split(" ")[0]
+    assert "limma" in res["recommended_de_method"]
+
+
+def test_step_failure_returns_partial_progress(monkeypatch):
+    calls = []
+    manifest = load_manifest("tcga_paad")
+    plan_tools = [
+        s["tool"] for s in _build_loading_plan(manifest) if s["tool"] in _LOADING_PLAN_TOOLS
+    ]
+    fail_tool = plan_tools[-1]  # fail the last loading step
+    monkeypatch.setattr(
+        catalog, "_loading_step_executors", lambda: _fake_executors(calls, fail_on=fail_tool)
+    )
+    res = catalog.dataset_load("tcga_paad")
+    assert "error" in res and fail_tool in res["error"]
+    # earlier steps are reported, and the last good intermediate is surfaced
+    assert [s["tool"] for s in res["steps_executed"]] == plan_tools[:-1]
+    assert res["adata_path"] == f"/tmp/fake/{plan_tools[-2]}_{len(calls)}.h5ad"
+
+
+def test_error_dict_from_loader_is_surfaced(monkeypatch):
+    def bad_executors():
+        ex = _fake_executors([])
+        first = [
+            s["tool"]
+            for s in _build_loading_plan(load_manifest("gse71729_moffitt"))
+            if s["tool"] in _LOADING_PLAN_TOOLS
+        ][0]
+        ex[first] = lambda **kw: {"error": "404 not found"}
+        return ex
+
+    monkeypatch.setattr(catalog, "_loading_step_executors", bad_executors)
+    res = catalog.dataset_load("gse71729_moffitt")
+    assert "error" in res and "404 not found" in res["error"]
+
+
+def test_single_cell_dataset_flags_pseudobulk(monkeypatch):
+    calls = []
+    monkeypatch.setattr(catalog, "_loading_step_executors", lambda: _fake_executors(calls))
+    res = catalog.dataset_load("gse155698_steele")
+    assert res["analysis_path"] == "P"
+    assert "pseudobulk" in res["recommended_de_method"].lower()
+    assert "pseudobulk" in res["next_step"].lower()