fix(planner): dataset_plan_analysis must say how to LOAD the dataset

Verifying the previous commit on dev showed rule 7 worked — both runs now open
with dataset_plan_analysis — but the plan they got back had no loading step, so
the agent still had to guess an entry point and still burned steps on it.

_workflow_for_intent emitted a load step only for expression_source.type ==
"geo_series_matrix". Most of the registry is an h5ad over `url`, so those
datasets returned a plan starting at dataset_describe and jumping straight to
collapse/DE, with `required_inputs: ["Expression file ..."]` as if the user
supplied it. Loading now comes from _build_loading_plan — already the single
source of truth for how to open a dataset, and it knows the modality-first Path
P routing and the precomputed `collapsed_url`.

Two details that matter:
- Only the get-the-data-open prefix is spliced in. _build_loading_plan also
carries an analysis tail (validate_contrast -> DE -> scoring) which the
intent-specific steps already own; splicing both duplicated it (caught by two
existing planner tests).
- The plan's own decoupler_inspect_data step is dropped for a registered
dataset. That step is why the agent kept calling it despite efficiency rule 4
forbidding it — the plan was telling it to.

Measured on gse28735_pdac: the plan now names decoupler_load_url_counts with the
_collapsed.h5ad URL, so the annotate+collapse pair disappears entirely.

Suite 1325 passed / 60 skipped.

src/tools/dataset_tools/_base.py

@@ -594,6 +594,23 @@ def _detect_intent(question: str) -> tuple[str, str, list[str]]:
     return best_intent, confidence, matched
 
 
+# The prefix of _build_loading_plan that is about OPENING the dataset. Anything
+# else it emits (validate_contrast, DE, scoring) is the analysis tail, which the
+# intent-specific steps in _workflow_for_intent own instead.
+_LOADING_PLAN_TOOLS = frozenset(
+    {
+        "decoupler_load_url_counts",
+        "decoupler_load_geo_series_matrix",
+        "decoupler_load_gdc_star_counts",
+        "decoupler_load_and_visualize_data",
+        "decoupler_annotate_probes_with_gpl",
+        "decoupler_collapse_probes_to_genes",
+        "decoupler_join_clinical_metadata",
+        "dataset_filter_to_curated_samples",
+    }
+)
+
+
 def _workflow_for_intent(
     intent: str,
     dataset_id: str,
@@ -632,26 +649,50 @@ def _workflow_for_intent(
     ]
     step_n += 1
 
-    # Optional loading steps from manifest
+    # Loading steps come from _build_loading_plan — the single source of truth
+    # for HOW to open a dataset. This used to emit a load step only for
+    # `geo_series_matrix`, so every h5ad-over-`url` dataset (most of the
+    # registry) got a plan with NO loading step at all and the agent had to
+    # guess an entry point, which cost 3-5 steps per run and often failed
+    # outright. The loading plan also knows about `collapsed_url`, which skips
+    # the annotate+collapse pair below.
+    loading_steps: list[dict] = []
     if manifest is not None and hasattr(manifest, "expression_source"):
-        if manifest.expression_source.get("type") == "geo_series_matrix":
-            steps.append(
-                {
-                    "step": step_n,
-                    "tool": "decoupler_load_geo_series_matrix",
-                    "purpose": "Download and parse GEO series matrix to h5ad.",
-                    "status": "available",
-                    "args_hint": {
-                        "url_or_path": manifest.expression_source.get("url"),
-                        "condition_column": (
-                            manifest.group_columns[0] if manifest.group_columns else None
-                        ),
-                    },
-                }
-            )
-            step_n += 1
+        try:
+            loading_steps = _build_loading_plan(manifest)
+        except Exception:  # a malformed manifest must not sink the whole plan
+            loading_steps = []
+
+    collapse_precomputed = False
+    for raw in loading_steps:
+        tool = raw.get("tool")
+        # Take only the GET-THE-DATA-OPEN prefix. _build_loading_plan also
+        # carries an analysis tail (validate_contrast → DE → scoring); the
+        # intent-specific steps below own that, and splicing both in would
+        # duplicate it.
+        if tool not in _LOADING_PLAN_TOOLS:
+            continue
+        # The loading plan's own inspect step is redundant for a REGISTERED
+        # dataset — data_level and analysis_path are manifest facts (efficiency
+        # rule 4). Keeping it here is what kept the agent calling it.
+        if tool == "decoupler_inspect_data":
+            continue
+        if tool == "decoupler_collapse_probes_to_genes":
+            collapse_precomputed = True
+        steps.append(
+            {
+                "step": step_n,
+                "tool": tool,
+                "purpose": raw.get("note", "Load the dataset as the manifest specifies."),
+                "status": "available",
+                "args_hint": raw.get("key_args", {}),
+            }
+        )
+        step_n += 1
 
-    if collapse_required:
+    # Only add a collapse step if the loading plan did not already cover it
+    # (a precomputed `collapsed_url` makes it unnecessary).
+    if collapse_required and not collapse_precomputed and not loading_steps:
         steps.append(
             {
                 "step": step_n,

tests/test_planner_tool.py

@@ -327,3 +327,66 @@ class TestManifestDrivenArgs:
         result = dataset_plan_analysis("nonexistent_dataset_xyz", "compare groups")
         assert result["detected_intent"] == "compare_groups"
         assert len(result["recommended_tools"]) > 0
+
+
+class TestPlanIncludesLoading:
+    """The plan must say HOW to open the dataset, not just what to run.
+
+    Measured on dev 2026-08-04: `_workflow_for_intent` emitted a load step only
+    for `geo_series_matrix`, so every h5ad-over-`url` dataset (most of the
+    registry) returned a plan with no loading step. The agent then guessed an
+    entry point — 3-5 wasted steps per run, and for the DE query it never found
+    a working one. Loading now comes from `_build_loading_plan`.
+    """
+
+    def test_h5ad_url_dataset_gets_a_load_step(self):
+        plan = dataset_plan_analysis(
+            dataset_id="gse28735_pdac",
+            user_question="Compare tumor vs normal for differential expression",
+        )
+        tools = [s["tool"] for s in plan["recommended_tools"]]
+        assert "decoupler_load_url_counts" in tools, tools
+        # ...and it must come before the analysis it feeds.
+        assert tools.index("decoupler_load_url_counts") < tools.index(
+            "decoupler_differential_expression"
+        )
+
+    def test_load_step_uses_the_precollapsed_url(self):
+        """The uncollapsed URL costs two extra steps (annotate + collapse)."""
+        plan = dataset_plan_analysis(
+            dataset_id="gse28735_pdac", user_question="differential expression tumor vs normal"
+        )
+        load = next(
+            s for s in plan["recommended_tools"] if s["tool"] == "decoupler_load_url_counts"
+        )
+        assert "collapsed" in load["args_hint"]["url_or_path"]
+        tools = [s["tool"] for s in plan["recommended_tools"]]
+        assert "decoupler_annotate_probes_with_gpl" not in tools
+        assert "decoupler_collapse_probes_to_genes" not in tools
+
+    def test_registered_dataset_plan_omits_inspect_data(self):
+        """Efficiency rule 4: data_level/analysis_path are manifest facts."""
+        for did in ("gse28735_pdac", "paca_au_rnaseq"):
+            plan = dataset_plan_analysis(dataset_id=did, user_question="compare groups")
+            tools = [s["tool"] for s in plan["recommended_tools"]]
+            assert "decoupler_inspect_data" not in tools, (did, tools)
+
+    def test_single_cell_plan_uses_the_sc_loader(self):
+        """Path P must not be handed a bulk flat-file loader (ADR-0006)."""
+        plan = dataset_plan_analysis(
+            dataset_id="gse155698_steele", user_question="compare tumor vs normal"
+        )
+        tools = [s["tool"] for s in plan["recommended_tools"]]
+        assert "decoupler_load_and_visualize_data" in tools, tools
+        assert "decoupler_load_url_counts" not in tools
+
+    def test_steps_are_numbered_consecutively(self):
+        plan = dataset_plan_analysis(dataset_id="gse28735_pdac", user_question="compare groups")
+        nums = [s["step"] for s in plan["recommended_tools"]]
+        assert nums == list(range(1, len(nums) + 1)), nums
+
+    def test_unknown_dataset_still_returns_a_plan(self):
+        """A missing manifest must degrade to a warning, not an exception."""
+        plan = dataset_plan_analysis(dataset_id="not_a_dataset", user_question="compare groups")
+        assert plan["recommended_tools"]
+        assert plan["warnings"]