ADR-0001 Phase 2 Mode A: early integration (T7 + T8 + T9)

T7 — src/workflows/integration.py: combined-AnnData builder. Harmonise each
dataset to a gene-symbol axis, intersect features, concat with a `batch` obs key
(=dataset_id). Pure core combine_anndatas() + loader wrapper
build_combined_anndata() (resolves each manifest's hosted h5ad).

T8 — decoupler_differential_expression gains batch_column: DESeq2 fits
~batch+factor; new run_limma_covariate (~batch+group) in microarray.py; ttest+batch
is refused (a t-test cannot model a covariate; no silent drop). Default None keeps
single-dataset behaviour unchanged.

T9 — decoupler_integrate_datasets (integration_mcp, 3rd tool): re-checks
get_integration_plan and pools + runs ONE batch-aware DE only on mode=="early";
refuses/reroutes for late (meta_analyze) / concordance / refuse. Auto-selects
deseq2 (raw counts) or limma from the plan's poolable_data_level.

Tests: test_integration_mode_a (9), test_de_batch_covariate (4),
test_integrate_datasets_tool (9). Full non-live suite: 915 passed. v1 limits:
gene-symbol axis only (no probe collapse / ortholog mapping), ComBat (step 11)
deferred. Branch-local; NOT deployed (origin = prod Space).

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

docs/adr/ADR-0001-implementation-plan.md

@@ -1,7 +1,7 @@
 # ADR-0001 Implementation Plan — Cross-Dataset Analysis
 
 **Companion to:** [`ADR-0001-cross-dataset-integration.md`](./ADR-0001-cross-dataset-integration.md)
-**Status:** Phase 1 (Mode B) complete & shipped to prod 2026-06-19 (T1–T6); Phase 2 (Mode A: T7–T9) not started (`early` falls back to `late`); Phase 3 (routing/interpretation/eval) pending. **Concordance mode** (same-cohort sibling quantifications) added later in biodata-registry 0.1.6 — see the 2026-06-24 update below; agent re-pinned + **deployed to prod** (`fb2091e`, `origin/main`). Reconciled 2026-06-24.
+**Status:** Phase 1 (Mode B) complete & shipped to prod 2026-06-19 (T1–T6); Phase 2 (Mode A: T7–T9) built + tested 2026-06-24 on branch `adr-0001-phase-2-mode-a`, not yet deployed (`early` still falls back to `late` in prod until then); Phase 3 (routing/interpretation/eval) pending. **Concordance mode** (same-cohort sibling quantifications) added later in biodata-registry 0.1.6 — see the 2026-06-24 update below; agent re-pinned + **deployed to prod** (`fb2091e`, `origin/main`). Reconciled 2026-06-24.
 **Scope:** `biodata-registry`, `DecoupleRpy_Agent`, `pdac-analysis-orchestrator`
 
 This plan turns the ADR's three phases into file-level work, grounded in the
@@ -115,17 +115,22 @@ pins `get_integration_plan`'s logic (the D3 poolability rule).
 
 ## Phase 2 — Mode A (early integration)
 
-8. [ ] **`src/workflows/integration.py`** — feature intersection + AnnData concat
+> **Built + tested 2026-06-24** on branch `adr-0001-phase-2-mode-a` (T7–T9; 915
+> non-live tests green). v1 harmonises to a gene-symbol axis (probe collapse /
+> ortholog mapping not handled), ttest+batch is refused (no silent ComBat), and
+> step 11 (ComBat) is deferred. **Not yet deployed** (origin = prod Space).
+
+8. [x] **`src/workflows/integration.py`** — feature intersection + AnnData concat
    → emit one **standard combined AnnData** with a `batch` obs key
    (`dataset_id`). Existing single-dataset tools run on it unchanged.
-9. [ ] **`supports_covariate` capability (D5)** — extend
+9. [x] **`supports_covariate` capability (D5)** — extend
    `decoupler_differential_expression` with an optional `batch_column`:
    DESeq2 `~batch + condition` (one-liner); **refactor `run_limma` to a design
    matrix** including batch; ttest path uses pre-correction instead. Back-compat
    default off.
-10. [ ] **`decoupler_integrate_datasets`** Layer 1 tool, gated behind the
+10. [x] **`decoupler_integrate_datasets`** Layer 1 tool, gated behind the
     batch-confound refusal from `get_integration_plan`.
-11. [ ] *(Optional)* add `inmoose` (ComBat / ComBat-seq) for the pre-correction
+11. [ ] *(Optional — deferred, not in v1)* add `inmoose` (ComBat / ComBat-seq) for the pre-correction
     route; keep the dependency optional.
 
 > **No per-tool integration work.** A future tool gains Mode A by reading the

src/tools/integration_tools.py

@@ -15,8 +15,11 @@ This MCP server exposes the cross-dataset result-combination paths:
    call overlap), because meta-analyzing identical samples double-counts the
    cohort. Routed here by get_integration_plan's mode == "concordance".
 
-The early-integration (pooling) tool `decoupler_integrate_datasets` is Phase 2
-(ADR-0001 T9) and will be added to this same server.
+3. decoupler_integrate_datasets: *early integration* — POOL >=2 datasets into one
+   matrix (feature intersection + a `batch` obs key) and run a single batch-aware
+   DE. Only for datasets the plan rules "early" (shared feature space + poolable
+   data_level); it re-checks the plan and refuses/reroutes otherwise. (ADR-0001
+   T9, Mode A.)
 
 Design (ADR-0001 Mode B):
 - The *math* is generic over the result envelope and lives in
@@ -32,7 +35,7 @@ from __future__ import annotations
 
 from datetime import datetime
 from pathlib import Path
-from typing import Annotated, Optional
+from typing import Annotated, Literal, Optional
 
 import pandas as pd
 from fastmcp import FastMCP
@@ -45,6 +48,11 @@ from src.core.combine import (
 from src.tools.rna._base import OUTPUT_DIR
 from src.workflows.concordance import concordance_metrics
 from src.workflows.meta_analysis import combine_envelopes
+# Mode-A early integration (T9): the plan gate, the combined-AnnData builder, and
+# the batch-aware DE tool. Module-level so tests can monkeypatch them.
+from src.datasets.registry import get_integration_plan
+from src.workflows.integration import build_combined_anndata
+from src.tools.rna.analysis import decoupler_differential_expression
 
 integration_mcp = FastMCP(name="integration")
 
@@ -462,3 +470,154 @@ def decoupler_normalization_concordance(
             }
         ],
     }
+
+
+@integration_mcp.tool
+def decoupler_integrate_datasets(
+    dataset_ids: Annotated[
+        list[str],
+        "Two or more registered dataset IDs to POOL into one matrix and analyze "
+        "together (early integration), e.g. ['tcga_paad', 'paca_au_rnaseq'].",
+    ],
+    design_factor: Annotated[
+        str,
+        "obs column holding the contrast groups; must exist in EVERY dataset "
+        "(e.g. 'tumor_subtype').",
+    ],
+    test_group: Annotated[str, "Test arm of the contrast (a value of design_factor)."],
+    control_group: Annotated[str, "Control / reference arm of the contrast."],
+    method: Annotated[
+        Literal["auto", "deseq2", "limma"],
+        "DE method on the pooled matrix. 'auto' (default) picks deseq2 for pooled "
+        "raw counts and limma otherwise, from the integration plan's poolable "
+        "data_level. `dataset` is always modelled as a batch covariate.",
+    ] = "auto",
+    min_shared_features: Annotated[
+        int,
+        "Minimum shared gene symbols required to pool; below this the datasets "
+        "likely do not share a feature space and the request is refused.",
+    ] = 200,
+    out_prefix: Annotated[Optional[str], "Output file prefix."] = None,
+) -> dict:
+    """
+    Early integration (Mode A): pool >=2 datasets into one matrix + one batch-aware DE.
+
+    Use this ONLY when dataset_get_integration_plan returns mode=="early": the
+    datasets share a poolable feature space AND data_level, so they can be
+    concatenated and analyzed jointly with `dataset` as a batch covariate (more
+    power than meta-analysis when valid). This tool RE-CHECKS the plan itself and
+    refuses (does not pool) when the verdict is not 'early':
+
+    - 'late'        -> run each dataset separately, then decoupler_meta_analyze.
+    - 'concordance' -> same-cohort variants -> decoupler_normalization_concordance.
+    - 'refuse'      -> incompatible (e.g. confounded design / cross-modality).
+
+    On 'early' it builds the combined AnnData (feature intersection + a `batch`
+    obs key = dataset_id), runs decoupler_differential_expression with
+    batch_column='batch' (DESeq2 ~batch+factor for raw counts, limma ~batch+group
+    otherwise), and returns the pooled DE result plus per-dataset composition.
+    Returns an {"error": ..., "refused": True} dict (does not raise) so the agent
+    can surface the reason and reroute.
+
+    Pass design_factor + test_group + control_group: they drive the plan's
+    confound check AND the pooled contrast.
+    """
+    if not isinstance(dataset_ids, list) or len(dataset_ids) < 2:
+        return _refusal(
+            "Early integration needs >=2 dataset_ids; got "
+            f"{len(dataset_ids) if isinstance(dataset_ids, list) else 'a non-list'}."
+        )
+
+    try:
+        plan = get_integration_plan(
+            dataset_ids,
+            design_factor=design_factor,
+            test_group=test_group,
+            control_group=control_group,
+        )
+    except Exception as exc:  # unknown dataset id, etc.
+        return _refusal(f"Could not compute the integration plan: {exc}")
+
+    mode = plan.get("mode")
+    if mode != "early":
+        reason = plan.get("reason", "")
+        reroute = {
+            "late": "Run each dataset separately, then call decoupler_meta_analyze.",
+            "concordance": (
+                "These are same-cohort variants — call "
+                "decoupler_normalization_concordance instead."
+            ),
+        }.get(mode, "")
+        return _refusal(
+            (
+                f"Datasets are not eligible for early pooling (plan mode='{mode}'). "
+                f"{reason} {reroute}"
+            ).strip(),
+            mode=mode,
+            plan_reason=reason,
+            refusal_rules_triggered=plan.get("refusal_rules_triggered", []),
+        )
+
+    # Derive the DE method from the poolable data_level unless explicitly set.
+    if method == "auto":
+        method = "deseq2" if plan.get("poolable_data_level") == "raw_counts" else "limma"
+
+    try:
+        built = build_combined_anndata(
+            dataset_ids,
+            design_factor=design_factor,
+            batch_key="batch",
+            min_shared_features=min_shared_features,
+            out_prefix=(f"{out_prefix}_combined" if out_prefix else None),
+        )
+    except Exception as exc:
+        return _refusal(f"Could not build the pooled dataset: {exc}", mode="early")
+
+    try:
+        de = decoupler_differential_expression(
+            adata_path=built["output_path"],
+            design_factor=design_factor,
+            contrast=[design_factor, test_group, control_group],
+            method=method,
+            batch_column="batch",
+            out_prefix=out_prefix,
+        )
+    except Exception as exc:
+        return _refusal(
+            f"Pooled DE failed: {exc}", mode="early", combined=built, method=method
+        )
+
+    return {
+        "message": (
+            f"Early integration of {len(dataset_ids)} datasets "
+            f"({', '.join(dataset_ids)}): pooled {built['n_obs']} samples over "
+            f"{built['n_vars']} shared genes and ran {de.get('method_used')} with "
+            f"dataset modelled as a batch covariate. "
+            f"{de.get('n_significant')} significant genes."
+        ),
+        "mode": "early",
+        "plan_reason": plan.get("reason", ""),
+        "datasets": list(dataset_ids),
+        "batch_key": "batch",
+        "per_batch_n": built["per_batch_n"],
+        "n_shared_features": built["n_vars"],
+        "n_combined_samples": built["n_obs"],
+        "method_used": de.get("method_used"),
+        "n_significant": de.get("n_significant"),
+        "batch_modeled": True,
+        "combined_path": built["output_path"],
+        "de_results_path": de.get("output_path"),
+        "sanity_warnings": de.get("sanity_warnings"),
+        "interpretation_note": (
+            "Early-integration (pooled) result: the datasets were concatenated and "
+            "analyzed jointly with `dataset` as a batch covariate, so the group "
+            "effect is estimated adjusting for dataset. Report the datasets combined, "
+            "the per-dataset sample counts, and the batch handling. This is more "
+            "powerful than meta-analysis, but valid ONLY because the plan verdict was "
+            "'early' (shared feature space + poolable data_level)."
+        ),
+        "artifacts": [
+            {"description": "Pooled DE results (CSV)", "path": de.get("output_path")},
+            {"description": "Combined AnnData (h5ad)", "path": built["output_path"]},
+        ],
+    }

src/tools/rna/analysis.py

@@ -8,6 +8,7 @@ from ._base import (  # noqa: F401
 )
 from src.core import combinable, make_combine_descriptor
 from src.workflows.sanity_checks import run_sanity_checks
+from src.workflows.microarray import run_limma_covariate
 
 # ── ADR-0001 Mode-B combine conformance (see src/core/combine.py) ─────────
 # Module-level so they validate at import: a bad result_type or envelope value
@@ -59,6 +60,15 @@ def decoupler_differential_expression(
         "when limma is unavailable so the statistical method is never silently changed. "
         "Set to True only to explicitly allow t-test fallback during exploratory analysis.",
     ] = False,
+    batch_column: Annotated[
+        str | None,
+        "Optional obs column to model as a batch covariate for cross-dataset early "
+        "integration (e.g. 'batch' in a combined AnnData from "
+        "decoupler_integrate_datasets). When set: DESeq2 fits ~batch + factor; "
+        "limma fits ~batch + group (the group effect is estimated adjusting for "
+        "batch); method='ttest' is rejected (a t-test cannot model a covariate). "
+        "Default None = single-dataset analysis (behaviour unchanged).",
+    ] = None,
     n_cpus: Annotated[int, "Number of CPUs for parallel processing (DESeq2 only)"] = 8,
     refit_cooks: Annotated[bool, "Whether to refit Cook's distances (DESeq2 only)"] = True,
     top_genes: Annotated[int, "Number of top genes to show in volcano plot"] = 10,
@@ -105,6 +115,13 @@ def decoupler_differential_expression(
             f"Available columns: {list(adata.obs.columns)}"
         )
 
+    # Batch covariate (Mode-A early integration): the column must exist.
+    if batch_column is not None and batch_column not in adata.obs.columns:
+        raise ValueError(
+            f"batch_column '{batch_column}' not found in obs. "
+            f"Available columns: {list(adata.obs.columns)}"
+        )
+
     # Guard: reject deseq2 on pre-normalised data before any expensive computation
     if method == "deseq2":
         _X = adata.X
@@ -135,7 +152,9 @@ def decoupler_differential_expression(
         inference = DefaultInference(n_cpus=n_cpus)
         dds = DeseqDataSet(
             adata=adata,
-            design=f"~{design_factor}",
+            design=(
+                f"~{batch_column} + {design_factor}" if batch_column else f"~{design_factor}"
+            ),
             refit_cooks=refit_cooks,
             inference=inference,
         )
@@ -155,27 +174,47 @@ def decoupler_differential_expression(
             X = X.toarray()
         mask_test = adata.obs[design_factor] == contrast[1]
         mask_ctrl = adata.obs[design_factor] == contrast[2]
-        X_test = X[mask_test.values]
-        X_ctrl = X[mask_ctrl.values]
         genes = list(adata.var.index)
-        if method == "limma":
-            try:
-                results_df = run_limma(X_test, X_ctrl, genes, contrast[1], contrast[2])
-                method_label = "limma"
-            except Exception as e:
-                if fallback_to_ttest:
-                    results_df = run_welch_ttest(X_test, X_ctrl, genes)
-                    method_label = f"ttest (limma unavailable, fallback requested: {e})"
-                else:
-                    raise RuntimeError(
-                        "method='limma' was requested but limma is unavailable: "
-                        f"{e}. "
-                        "Run decoupler_check_limma_available() to diagnose. "
-                        "Set fallback_to_ttest=True to explicitly allow t-test fallback."
-                    ) from e
+
+        if batch_column is not None:
+            # Mode-A early integration: model batch as a covariate (limma only).
+            # Do NOT fall back to ttest here — a t-test cannot honour the
+            # covariate, so silently dropping it would be wrong; surface instead.
+            if method != "limma":
+                raise ValueError(
+                    f"batch_column requires method='limma' (or method='deseq2' for raw "
+                    f"counts); method='{method}' cannot model a batch covariate. Use "
+                    f"limma/deseq2, or batch-correct (e.g. ComBat) before a t-test."
+                )
+            mask_both = (mask_test | mask_ctrl).values
+            X_both = X[mask_both]
+            groups = adata.obs[design_factor][mask_both].astype(str).tolist()
+            batches = adata.obs[batch_column][mask_both].astype(str).tolist()
+            results_df = run_limma_covariate(
+                X_both, groups, batches, genes, contrast[1], contrast[2]
+            )
+            method_label = f"limma (~{batch_column} + {design_factor})"
         else:
-            results_df = run_welch_ttest(X_test, X_ctrl, genes)
-            method_label = "ttest"
+            X_test = X[mask_test.values]
+            X_ctrl = X[mask_ctrl.values]
+            if method == "limma":
+                try:
+                    results_df = run_limma(X_test, X_ctrl, genes, contrast[1], contrast[2])
+                    method_label = "limma"
+                except Exception as e:
+                    if fallback_to_ttest:
+                        results_df = run_welch_ttest(X_test, X_ctrl, genes)
+                        method_label = f"ttest (limma unavailable, fallback requested: {e})"
+                    else:
+                        raise RuntimeError(
+                            "method='limma' was requested but limma is unavailable: "
+                            f"{e}. "
+                            "Run decoupler_check_limma_available() to diagnose. "
+                            "Set fallback_to_ttest=True to explicitly allow t-test fallback."
+                        ) from e
+            else:
+                results_df = run_welch_ttest(X_test, X_ctrl, genes)
+                method_label = "ttest"
 
     # Save results — name the index so pd.read_csv produces a 'gene' column
     results_df.index.name = "gene"

src/workflows/integration.py

@@ -0,0 +1,221 @@
+"""
+Cross-dataset early integration (ADR-0001, Mode A — T7).
+
+Build ONE combined AnnData from >=2 registered datasets by:
+  1. harmonising each to a shared gene-symbol feature axis,
+  2. intersecting features (join="inner"), and
+  3. concatenating samples with a ``batch`` obs column = dataset_id.
+
+The result is a *standard* AnnData that the existing single-dataset tools
+(``decoupler_differential_expression`` and the enrichment tools) consume
+unchanged — Mode A pools the matrices and a downstream DE models ``dataset_id``
+as a batch covariate (see ``decoupler_differential_expression``'s
+``batch_column``, T8). Whether a set of datasets is eligible for early pooling is
+decided upstream by ``get_integration_plan`` (mode == "early"); this module
+performs the mechanics once that verdict is in (the gating lives in the
+``decoupler_integrate_datasets`` tool, T9).
+
+Pure core: ``combine_anndatas(adatas, ...)`` operates on already-loaded AnnData
+objects (unit-testable, no network). ``build_combined_anndata(dataset_ids, ...)``
+is the thin loader wrapper that resolves each manifest's hosted h5ad.
+
+v1 scope / limitations:
+- Harmonises to gene symbols: a dataset whose ``feature_id_type`` is
+  "gene_symbol" uses var_names directly; otherwise a var column named
+  SYMBOL/gene_symbol/gene_name is used if present; anything else raises (probe
+  collapse / ortholog mapping / entrez without a symbol column are NOT handled
+  here — such datasets do not reach an ``early`` verdict in v1).
+- Duplicate symbols within a dataset are collapsed keep-first (preserves raw
+  integer counts for the DESeq2 path — no averaging).
+"""
+from __future__ import annotations
+
+import os
+import tempfile
+import urllib.request
+from pathlib import Path
+from typing import TYPE_CHECKING, Optional, Sequence
+
+import anndata as ad
+import numpy as np
+
+from src.tools.rna._base import OUTPUT_DIR
+
+if TYPE_CHECKING:
+    from anndata import AnnData
+
+# var columns checked, in order, for a gene-symbol axis when feature_id_type
+# is not already "gene_symbol".
+_SYMBOL_VAR_COLUMNS = ("SYMBOL", "gene_symbol", "Gene_Symbol", "symbol", "gene_name")
+
+
+def _symbol_axis(adata: "AnnData", feature_id_type: str) -> list[str]:
+    """Return a per-var list of gene symbols for ``adata``, or raise if unavailable."""
+    if feature_id_type == "gene_symbol":
+        return [str(v) for v in adata.var_names]
+    for col in _SYMBOL_VAR_COLUMNS:
+        if col in adata.var.columns:
+            return [str(v) for v in adata.var[col].tolist()]
+    raise ValueError(
+        f"cannot resolve gene symbols: feature_id_type='{feature_id_type}' and no "
+        f"symbol column {list(_SYMBOL_VAR_COLUMNS)} in var. Mode-A v1 needs a "
+        f"gene-symbol axis (probe collapse / id-mapping is not handled here)."
+    )
+
+
+def _to_symbol_adata(adata: "AnnData", feature_id_type: str) -> "AnnData":
+    """Return a copy of ``adata`` whose var_names are unique gene symbols.
+
+    Drops features with empty/missing symbols; collapses duplicate symbols
+    keep-first (preserves raw counts — no averaging)."""
+    symbols = _symbol_axis(adata, feature_id_type)
+    a = adata.copy()
+    a.var_names = [str(s) for s in symbols]
+    keep = np.array(
+        [bool(s) and s.strip().lower() not in ("nan", "none", "") for s in a.var_names]
+    )
+    a = a[:, keep].copy()
+    # keep-first on duplicate symbols (np.unique returns the first index per value)
+    _, first_idx = np.unique(np.asarray(a.var_names, dtype=object), return_index=True)
+    a = a[:, np.sort(first_idx)].copy()
+    return a
+
+
+def combine_anndatas(
+    adatas: "Sequence[AnnData]",
+    dataset_ids: Sequence[str],
+    feature_id_types: Sequence[str],
+    *,
+    design_factor: Optional[str] = None,
+    batch_key: str = "batch",
+    min_shared_features: int = 200,
+) -> "AnnData":
+    """Pure core: feature-intersection concat of >=2 AnnData into one combined AnnData.
+
+    Each input is harmonised to a unique gene-symbol var axis, the var
+    intersection is taken (``join="inner"``), a ``batch_key`` obs column is set to
+    the dataset_id, and samples are concatenated (obs names disambiguated per
+    batch with a ``-<dataset_id>`` suffix).
+
+    Raises
+    ------
+    ValueError
+        on <2 inputs, mismatched argument lengths, an unresolvable symbol axis,
+        a missing ``design_factor`` column in any dataset, or a feature
+        intersection smaller than ``min_shared_features``.
+    """
+    if len(adatas) < 2:
+        raise ValueError(f"early integration needs >=2 datasets; got {len(adatas)}")
+    if not (len(adatas) == len(dataset_ids) == len(feature_id_types)):
+        raise ValueError(
+            "adatas, dataset_ids, and feature_id_types must be the same length "
+            f"({len(adatas)}, {len(dataset_ids)}, {len(feature_id_types)})"
+        )
+
+    prepared: list["AnnData"] = []
+    for a, did, fit in zip(adatas, dataset_ids, feature_id_types):
+        sa = _to_symbol_adata(a, fit)
+        if design_factor is not None and design_factor not in sa.obs.columns:
+            raise ValueError(
+                f"dataset '{did}' has no obs column '{design_factor}' required for "
+                f"the requested contrast."
+            )
+        prepared.append(sa)
+
+    shared = set(prepared[0].var_names)
+    for sa in prepared[1:]:
+        shared &= set(sa.var_names)
+    if len(shared) < min_shared_features:
+        raise ValueError(
+            f"only {len(shared)} shared gene symbols across {list(dataset_ids)} "
+            f"(minimum {min_shared_features}); the datasets likely do not share a "
+            f"feature space and are not eligible for early pooling."
+        )
+
+    combined = ad.concat(
+        prepared,
+        axis=0,
+        join="inner",  # feature intersection
+        label=batch_key,
+        keys=[str(d) for d in dataset_ids],
+        index_unique="-",  # disambiguate duplicate sample barcodes across cohorts
+        merge="same",
+    )
+    combined.obs[batch_key] = combined.obs[batch_key].astype("category")
+    return combined
+
+
+def _resolve_to_local(url_or_path: str) -> tuple[str, bool]:
+    """Return (local_path, is_temp). Downloads http(s)/ftp URLs to a temp file."""
+    if str(url_or_path).startswith(("http://", "https://", "ftp://")):
+        suffix = Path(url_or_path.split("?", 1)[0]).suffix or ".h5ad"
+        fd, tmp = tempfile.mkstemp(suffix=suffix)
+        os.close(fd)
+        urllib.request.urlretrieve(url_or_path, tmp)
+        return tmp, True
+    return str(url_or_path), False
+
+
+def build_combined_anndata(
+    dataset_ids: Sequence[str],
+    *,
+    design_factor: Optional[str] = None,
+    batch_key: str = "batch",
+    min_shared_features: int = 200,
+    out_prefix: Optional[str] = None,
+) -> dict:
+    """Loader wrapper: resolve each registered dataset's hosted h5ad and combine.
+
+    Resolves each ``dataset_id`` to its manifest's ``expression_source.url``,
+    reads the AnnData, and delegates to :func:`combine_anndatas`. Writes the
+    combined AnnData to ``OUTPUT_DIR`` and returns a summary dict.
+    """
+    import scanpy as sc
+
+    from src.datasets.registry import load_manifest
+
+    if len(dataset_ids) < 2:
+        raise ValueError(f"early integration needs >=2 datasets; got {len(dataset_ids)}")
+
+    adatas: list["AnnData"] = []
+    feature_id_types: list[str] = []
+    temps: list[str] = []
+    try:
+        for did in dataset_ids:
+            m = load_manifest(did)
+            url = (m.expression_source or {}).get("url")
+            if not url:
+                raise ValueError(f"dataset '{did}' has no expression_source.url")
+            local, is_temp = _resolve_to_local(url)
+            if is_temp:
+                temps.append(local)
+            adatas.append(sc.read_h5ad(local))
+            feature_id_types.append(m.feature_id_type)
+
+        combined = combine_anndatas(
+            adatas,
+            dataset_ids,
+            feature_id_types,
+            design_factor=design_factor,
+            batch_key=batch_key,
+            min_shared_features=min_shared_features,
+        )
+    finally:
+        for t in temps:
+            Path(t).unlink(missing_ok=True)
+
+    out_prefix = out_prefix or ("combined_" + "_".join(str(d) for d in dataset_ids))[:80]
+    OUTPUT_DIR.mkdir(parents=True, exist_ok=True)
+    out_path = OUTPUT_DIR / f"{out_prefix}.h5ad"
+    combined.write_h5ad(out_path)
+
+    per_batch = combined.obs[batch_key].value_counts().to_dict()
+    return {
+        "output_path": str(out_path.resolve()),
+        "n_obs": int(combined.n_obs),
+        "n_vars": int(combined.n_vars),
+        "batch_key": batch_key,
+        "shared_feature_space": "gene_symbol",
+        "per_batch_n": {str(k): int(v) for k, v in per_batch.items()},
+        "dataset_ids": list(dataset_ids),
+    }

src/workflows/microarray.py

@@ -926,3 +926,117 @@ cat("OK\\n")
     })
     top_df.index = genes
     return top_df[["log2FoldChange", "stat", "pvalue", "padj"]]
+
+
+def run_limma_covariate(
+    X: "np.ndarray",
+    group_labels: "list[str]",
+    batch_labels: "list[str]",
+    genes: "list[str]",
+    test_group: str,
+    control_group: str,
+) -> "pd.DataFrame":
+    """
+    Limma with a batch covariate via Rscript — ``model.matrix(~ batch + group)``.
+
+    Mode-A early integration (ADR-0001 T8): when several cohorts are pooled into
+    one matrix, the per-cohort ``batch`` is modelled as a covariate so the group
+    effect is estimated *adjusting* for it. Group labels are recoded to ctrl/test
+    (control_group -> "ctrl", test_group -> "test") so the tested coefficient is
+    always "grouptest" regardless of the original label spelling; batch enters as
+    additional factor columns whose names do not matter (only the group
+    coefficient is read back).
+
+    Parameters
+    ----------
+    X : samples x genes matrix (rows = samples, aligned with group/batch labels).
+    group_labels, batch_labels : per-sample labels, length == X.shape[0].
+    genes : gene ids, length == X.shape[1].
+
+    Returns columns: log2FoldChange, stat, pvalue, padj (same schema as run_limma).
+    Raises RuntimeError if Rscript/limma is unavailable or the design is rank-
+    deficient (e.g. batch perfectly confounded with group).
+    """
+    import os
+    import subprocess
+    import tempfile
+
+    X = np.asarray(X, dtype=np.float64)
+    n = X.shape[0]
+    if not (len(group_labels) == len(batch_labels) == n):
+        raise ValueError(
+            f"group_labels ({len(group_labels)}) and batch_labels "
+            f"({len(batch_labels)}) must match X sample count ({n})"
+        )
+    coded = ["test" if str(g) == str(test_group) else "ctrl" for g in group_labels]
+    if len(set(coded)) < 2:
+        raise ValueError(
+            f"need both groups present; got only {set(group_labels)} for "
+            f"test='{test_group}' / control='{control_group}'"
+        )
+
+    expr = X.T  # genes x samples
+    with tempfile.TemporaryDirectory() as tmpdir:
+        expr_csv = os.path.join(tmpdir, "expr.csv")
+        meta_csv = os.path.join(tmpdir, "meta.csv")
+        result_csv = os.path.join(tmpdir, "result.csv")
+
+        pd.DataFrame(expr, index=genes).to_csv(expr_csv, header=False)
+        pd.DataFrame(
+            {"group": coded, "batch": [str(b) for b in batch_labels]}
+        ).to_csv(meta_csv, index=False)
+
+        r_script = f"""
+suppressPackageStartupMessages({{
+    library(limma)
+    library(utils)
+}})
+
+expr  <- as.matrix(read.csv("{expr_csv}", header=FALSE, row.names=1))
+meta  <- read.csv("{meta_csv}", colClasses = "character")
+group <- factor(meta$group, levels = c("ctrl", "test"))
+batch <- factor(meta$batch)
+design <- model.matrix(~ batch + group)
+if (qr(design)$rank < ncol(design)) {{
+    stop("design is rank-deficient (batch likely confounded with group)")
+}}
+
+fit    <- lmFit(expr, design)
+fit    <- eBayes(fit)
+result <- topTable(fit, coef = "grouptest",
+                   number = nrow(expr),
+                   sort.by = "none",
+                   adjust.method = "BH")
+
+write.csv(result, "{result_csv}", row.names = TRUE)
+cat("OK\\n")
+"""
+        proc = subprocess.run(
+            ["Rscript", "--vanilla", "-"],
+            input=r_script,
+            capture_output=True,
+            text=True,
+            timeout=300,
+        )
+
+        if proc.returncode != 0:
+            raise RuntimeError(
+                f"Rscript/limma (covariate) failed (exit {proc.returncode}):\n"
+                f"{proc.stderr.strip()}"
+            )
+        if not os.path.exists(result_csv):
+            raise RuntimeError(
+                f"Rscript ran but produced no output.\nstdout: {proc.stdout}\n"
+                f"stderr: {proc.stderr}"
+            )
+
+        top_df = pd.read_csv(result_csv, index_col=0)
+
+    top_df = top_df.rename(columns={
+        "logFC":     "log2FoldChange",
+        "t":         "stat",
+        "P.Value":   "pvalue",
+        "adj.P.Val": "padj",
+    })
+    top_df.index = genes
+    return top_df[["log2FoldChange", "stat", "pvalue", "padj"]]

tests/test_de_batch_covariate.py

@@ -0,0 +1,115 @@
+"""
+Tests for the batch-covariate DE path (ADR-0001 T8).
+
+Covers:
+- run_limma_covariate output schema (skipped if Rscript/limma is unavailable),
+- the decoupler_differential_expression guards: method='ttest' + batch_column is
+  rejected, and a missing batch_column is rejected,
+- back-compat: batch_column=None still runs the plain two-group ttest path.
+
+Tiny synthetic AnnData only — no network. The DESeq2 covariate path (a one-line
+design change) is exercised by the existing DESeq2 tests + the design string;
+it is not re-run here (needs raw counts + the pydeseq2 stack).
+"""
+from __future__ import annotations
+
+import sys
+from pathlib import Path
+
+import anndata as ad
+import numpy as np
+import pandas as pd
+import pytest
+
+sys.path.insert(0, str(Path(__file__).parent.parent))
+
+from src.workflows.microarray import run_limma_covariate  # noqa: E402
+from src.tools.rna.analysis import decoupler_differential_expression  # noqa: E402
+
+
+def _rscript_limma_available() -> bool:
+    """True if a tiny run_limma_covariate succeeds (Rscript + limma installed)."""
+    rng = np.random.default_rng(0)
+    X = rng.normal(size=(6, 4))
+    X[:3, :] += 3.0  # group effect on the 'test' samples
+    try:
+        run_limma_covariate(
+            X,
+            ["t", "t", "t", "c", "c", "c"],
+            ["b1", "b2", "b1", "b2", "b1", "b2"],
+            ["G1", "G2", "G3", "G4"],
+            "t",
+            "c",
+        )
+        return True
+    except RuntimeError:
+        return False
+    except Exception:
+        return False
+
+
+def _h5ad(tmp_path, *, with_batch=True, n_genes=5):
+    rng = np.random.default_rng(1)
+    n = 6
+    X = rng.normal(loc=5.0, scale=1.0, size=(n, n_genes)).astype(np.float32)
+    obs = pd.DataFrame(
+        {"condition": ["treatment", "treatment", "treatment", "control", "control", "control"]},
+        index=[f"s{i}" for i in range(n)],
+    )
+    if with_batch:
+        obs["batch"] = ["A", "B", "A", "B", "A", "B"]
+    var = pd.DataFrame(index=[f"G{i}" for i in range(n_genes)])
+    a = ad.AnnData(X=X, obs=obs, var=var)
+    p = tmp_path / "ds.h5ad"
+    a.write_h5ad(p)
+    return str(p)
+
+
+@pytest.mark.skipif(not _rscript_limma_available(), reason="Rscript/limma not available")
+def test_run_limma_covariate_schema():
+    rng = np.random.default_rng(2)
+    X = rng.normal(size=(8, 10))
+    X[:4, :] += 2.5  # test-group effect
+    genes = [f"G{i}" for i in range(10)]
+    groups = ["t", "t", "t", "t", "c", "c", "c", "c"]
+    batches = ["b1", "b2", "b1", "b2", "b1", "b2", "b1", "b2"]
+    out = run_limma_covariate(X, groups, batches, genes, "t", "c")
+    assert list(out.columns) == ["log2FoldChange", "stat", "pvalue", "padj"]
+    assert list(out.index) == genes
+    assert out["stat"].notna().all()
+
+
+def test_ttest_with_batch_rejected(tmp_path):
+    path = _h5ad(tmp_path, with_batch=True)
+    with pytest.raises(ValueError, match="cannot model a batch covariate"):
+        decoupler_differential_expression(
+            adata_path=path,
+            design_factor="condition",
+            contrast=["condition", "treatment", "control"],
+            method="ttest",
+            batch_column="batch",
+        )
+
+
+def test_missing_batch_column_rejected(tmp_path):
+    path = _h5ad(tmp_path, with_batch=False)
+    with pytest.raises(ValueError, match="batch_column 'batch' not found"):
+        decoupler_differential_expression(
+            adata_path=path,
+            design_factor="condition",
+            contrast=["condition", "treatment", "control"],
+            method="limma",
+            batch_column="batch",
+        )
+
+
+def test_backcompat_ttest_no_batch(tmp_path):
+    path = _h5ad(tmp_path, with_batch=True)
+    res = decoupler_differential_expression(
+        adata_path=path,
+        design_factor="condition",
+        contrast=["condition", "treatment", "control"],
+        method="ttest",
+    )
+    assert res["method_used"] == "ttest"
+    assert "output_path" in res

tests/test_integrate_datasets_tool.py

@@ -0,0 +1,153 @@
+"""
+Tests for the decoupler_integrate_datasets MCP tool (ADR-0001 T9, Mode A).
+
+The tool's value is the routing gate: it pools + runs a batch-aware DE ONLY on a
+plan verdict of 'early', and refuses/reroutes otherwise. These tests monkeypatch
+the plan, the combined-AnnData builder, and the DE tool so the routing is checked
+deterministically with no network or compute. The actual pooling/DE mechanics
+are covered by test_integration_mode_a.py and test_de_batch_covariate.py.
+"""
+from __future__ import annotations
+
+import asyncio
+import sys
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).parent.parent))
+
+import src.tools.integration_tools as it  # noqa: E402
+from src.tools.integration_tools import (  # noqa: E402
+    decoupler_integrate_datasets,
+    integration_mcp,
+)
+
+
+def _plan(mode, **extra):
+    base = {"mode": mode, "reason": f"{mode} reason.", "refusal_rules_triggered": []}
+    base.update(extra)
+    return lambda *a, **k: base
+
+
+def test_tool_registered():
+    names = {t.name for t in asyncio.run(integration_mcp.list_tools())}
+    assert "decoupler_integrate_datasets" in names
+
+
+def test_arity_guard():
+    out = decoupler_integrate_datasets(
+        dataset_ids=["solo"], design_factor="g", test_group="t", control_group="c"
+    )
+    assert out["refused"] is True
+    assert ">=2" in out["error"]
+
+
+def test_late_reroutes_to_meta_analyze(monkeypatch):
+    monkeypatch.setattr(it, "get_integration_plan", _plan("late"))
+    out = decoupler_integrate_datasets(
+        dataset_ids=["a", "b"], design_factor="g", test_group="t", control_group="c"
+    )
+    assert out["refused"] and out["mode"] == "late"
+    assert "decoupler_meta_analyze" in out["error"]
+
+
+def test_concordance_reroutes(monkeypatch):
+    monkeypatch.setattr(it, "get_integration_plan", _plan("concordance"))
+    out = decoupler_integrate_datasets(
+        dataset_ids=["a", "b"], design_factor="g", test_group="t", control_group="c"
+    )
+    assert out["refused"] and out["mode"] == "concordance"
+    assert "decoupler_normalization_concordance" in out["error"]
+
+
+def test_refuse_passes_reason_and_rules(monkeypatch):
+    monkeypatch.setattr(
+        it, "get_integration_plan",
+        _plan("refuse", reason="CONFOUNDED_DESIGN: ...", refusal_rules_triggered=["CONFOUNDED_DESIGN"]),
+    )
+    out = decoupler_integrate_datasets(
+        dataset_ids=["a", "b"], design_factor="g", test_group="t", control_group="c"
+    )
+    assert out["refused"] and out["mode"] == "refuse"
+    assert "CONFOUNDED_DESIGN" in out["error"]
+    assert out["refusal_rules_triggered"] == ["CONFOUNDED_DESIGN"]
+
+
+def test_plan_exception_refuses(monkeypatch):
+    def boom(*a, **k):
+        raise ValueError("dataset 'nope' not registered")
+    monkeypatch.setattr(it, "get_integration_plan", boom)
+    out = decoupler_integrate_datasets(
+        dataset_ids=["a", "nope"], design_factor="g", test_group="t", control_group="c"
+    )
+    assert out["refused"] and "not registered" in out["error"]
+
+
+def test_early_builds_and_runs_batch_aware_de(monkeypatch, tmp_path):
+    monkeypatch.setattr(
+        it, "get_integration_plan", _plan("early", poolable_data_level="raw_counts")
+    )
+    built = {
+        "output_path": str(tmp_path / "combined.h5ad"),
+        "n_obs": 42, "n_vars": 1500, "per_batch_n": {"a": 20, "b": 22},
+        "dataset_ids": ["a", "b"],
+    }
+    monkeypatch.setattr(it, "build_combined_anndata", lambda *a, **k: built)
+    captured = {}
+
+    def fake_de(**kwargs):
+        captured.update(kwargs)
+        return {
+            "method_used": "DESeq2", "n_significant": 7,
+            "output_path": str(tmp_path / "de.csv"), "sanity_warnings": {"n_warnings": 0},
+        }
+
+    monkeypatch.setattr(it, "decoupler_differential_expression", fake_de)
+    out = decoupler_integrate_datasets(
+        dataset_ids=["a", "b"], design_factor="subtype",
+        test_group="basal", control_group="classical",
+    )
+    assert out["mode"] == "early" and out["batch_modeled"] is True
+    assert out["n_significant"] == 7 and out["method_used"] == "DESeq2"
+    assert out["per_batch_n"] == {"a": 20, "b": 22}
+    # auto -> deseq2 for raw counts; batch covariate + contrast wired correctly
+    assert captured["method"] == "deseq2"
+    assert captured["batch_column"] == "batch"
+    assert captured["contrast"] == ["subtype", "basal", "classical"]
+    assert captured["adata_path"] == built["output_path"]
+
+
+def test_early_auto_method_limma_for_non_raw(monkeypatch, tmp_path):
+    monkeypatch.setattr(
+        it, "get_integration_plan", _plan("early", poolable_data_level="log_expression")
+    )
+    monkeypatch.setattr(
+        it, "build_combined_anndata",
+        lambda *a, **k: {"output_path": str(tmp_path / "c.h5ad"), "n_obs": 10,
+                         "n_vars": 900, "per_batch_n": {"a": 5, "b": 5}},
+    )
+    captured = {}
+
+    def fake_de(**kwargs):
+        captured.update(kwargs)
+        return {"method_used": "limma", "n_significant": 3,
+                "output_path": str(tmp_path / "de.csv")}
+
+    monkeypatch.setattr(it, "decoupler_differential_expression", fake_de)
+    decoupler_integrate_datasets(
+        dataset_ids=["a", "b"], design_factor="g", test_group="t", control_group="c"
+    )
+    assert captured["method"] == "limma"
+
+
+def test_early_build_failure_refuses(monkeypatch):
+    monkeypatch.setattr(
+        it, "get_integration_plan", _plan("early", poolable_data_level="raw_counts")
+    )
+
+    def boom(*a, **k):
+        raise ValueError("only 12 shared gene symbols")
+    monkeypatch.setattr(it, "build_combined_anndata", boom)
+    out = decoupler_integrate_datasets(
+        dataset_ids=["a", "b"], design_factor="g", test_group="t", control_group="c"
+    )
+    assert out["refused"] and "shared gene symbols" in out["error"]

tests/test_integration_mode_a.py

@@ -0,0 +1,118 @@
+"""
+Unit tests for the Mode-A early-integration core (ADR-0001 T7).
+
+Pure tests over constructed AnnData objects — no data, no network. Cover the
+feature-intersection concat, the gene-symbol harmonisation (var.index vs a
+symbol column), keep-first duplicate collapse, empty-symbol dropping, the
+batch obs key, and the arity / shared-feature / design-factor guards.
+"""
+from __future__ import annotations
+
+import sys
+from pathlib import Path
+
+import anndata as ad
+import numpy as np
+import pandas as pd
+import pytest
+
+sys.path.insert(0, str(Path(__file__).parent.parent))
+
+from src.workflows.integration import combine_anndatas  # noqa: E402
+
+
+def _adata(symbols, *, n_obs=4, condition=None, symbol_col=None):
+    """Build a small AnnData. If symbol_col is given, var.index holds opaque ids
+    and the symbols live in that var column (mimics ensembl + SYMBOL)."""
+    X = (np.arange(n_obs * len(symbols), dtype=float).reshape(n_obs, len(symbols)) + 1.0)
+    if symbol_col:
+        var = pd.DataFrame(index=[f"id{i}" for i in range(len(symbols))])
+        var[symbol_col] = list(symbols)
+    else:
+        var = pd.DataFrame(index=list(symbols))
+    obs = pd.DataFrame(index=[f"s{i}" for i in range(n_obs)])
+    if condition is not None:
+        obs["condition"] = list(condition)
+    return ad.AnnData(X=X, obs=obs, var=var)
+
+
+def test_intersection_concat_and_batch_key():
+    a = _adata(["G1", "G2", "G3"], n_obs=3)
+    b = _adata(["G2", "G3", "G4"], n_obs=2)
+    out = combine_anndatas(
+        [a, b], ["A", "B"], ["gene_symbol", "gene_symbol"], min_shared_features=1
+    )
+    assert set(out.var_names) == {"G2", "G3"}  # feature intersection
+    assert out.n_obs == 5
+    assert out.obs["batch"].value_counts().to_dict() == {"A": 3, "B": 2}
+    # obs names disambiguated per batch
+    assert any(name.endswith("-A") for name in out.obs_names)
+    assert any(name.endswith("-B") for name in out.obs_names)
+
+
+def test_symbol_column_used_when_not_gene_symbol():
+    a = _adata(["G1", "G2", "G3"], symbol_col="SYMBOL")  # var.index = id0.. , SYMBOL col
+    b = _adata(["G2", "G3"], symbol_col="SYMBOL")
+    out = combine_anndatas(
+        [a, b], ["A", "B"], ["ensembl_gene_id", "ensembl_gene_id"], min_shared_features=1
+    )
+    assert set(out.var_names) == {"G2", "G3"}
+
+
+def test_duplicate_symbols_collapsed_keep_first():
+    a = _adata(["G1", "G1", "G2"], n_obs=2)  # G1 duplicated
+    b = _adata(["G1", "G2"], n_obs=2)
+    out = combine_anndatas([a, b], ["A", "B"], ["gene_symbol", "gene_symbol"], min_shared_features=1)
+    # G1 deduped within A keep-first, then intersected
+    assert sorted(out.var_names) == ["G1", "G2"]
+
+
+def test_empty_and_nan_symbols_dropped():
+    a = _adata(["G1", "", "G2", "nan"], n_obs=2)
+    b = _adata(["G1", "G2"], n_obs=2)
+    out = combine_anndatas([a, b], ["A", "B"], ["gene_symbol", "gene_symbol"], min_shared_features=1)
+    assert sorted(out.var_names) == ["G1", "G2"]
+
+
+def test_arity_guard():
+    a = _adata(["G1", "G2"])
+    with pytest.raises(ValueError, match="needs >=2"):
+        combine_anndatas([a], ["A"], ["gene_symbol"])
+
+
+def test_too_few_shared_features_raises():
+    a = _adata(["G1", "G2", "G3"])
+    b = _adata(["G4", "G5", "G6"])  # disjoint
+    with pytest.raises(ValueError, match="shared gene symbols"):
+        combine_anndatas([a, b], ["A", "B"], ["gene_symbol", "gene_symbol"], min_shared_features=1)
+
+
+def test_missing_design_factor_raises():
+    a = _adata(["G1", "G2"], condition=["t", "c", "t", "c"])
+    b = _adata(["G1", "G2"])  # no condition column
+    with pytest.raises(ValueError, match="no obs column 'condition'"):
+        combine_anndatas(
+            [a, b], ["A", "B"], ["gene_symbol", "gene_symbol"],
+            design_factor="condition", min_shared_features=1,
+        )
+
+
+def test_design_factor_present_ok():
+    a = _adata(["G1", "G2"], condition=["t", "c", "t", "c"])
+    b = _adata(["G1", "G2"], condition=["t", "c", "t", "c"])
+    out = combine_anndatas(
+        [a, b], ["A", "B"], ["gene_symbol", "gene_symbol"],
+        design_factor="condition", min_shared_features=1,
+    )
+    assert "condition" in out.obs.columns
+    assert "batch" in out.obs.columns
+
+
+def test_unresolvable_symbol_axis_raises():
+    a = _adata(["G1", "G2"], symbol_col="SYMBOL")
+    b = _adata(["G1", "G2"], symbol_col="SYMBOL")
+    # feature_id_type not gene_symbol AND no recognised symbol column -> drop the col
+    a.var = a.var.rename(columns={"SYMBOL": "mystery"})
+    b.var = b.var.rename(columns={"SYMBOL": "mystery"})
+    with pytest.raises(ValueError, match="cannot resolve gene symbols"):
+        combine_anndatas([a, b], ["A", "B"], ["probe_id", "probe_id"], min_shared_features=1)