src/tools/bulk_dataset_tools.py

"""
MCP tools for bulk expression activity scoring and group comparison.

Tools
-----
dataset_score_bulk_samples
    Score pre-normalised bulk expression samples using PROGENy, CollecTRI,
    or Hallmark prior knowledge via decoupleR ULM/MLM/zscore.

dataset_compare_activity_by_group
    Compare per-sample activity scores between two metadata-defined groups
    (Welch's t-test + BH FDR).  Use after dataset_score_bulk_samples when
    the user asks whether TF/pathway/hallmark activity differs between groups.

Design rules
------------
- No dataset-specific logic.  Pass any samples × genes expression file.
- Scoring logic lives in src/workflows/activity_scoring.py.
- Comparison logic lives in src/workflows/activity_stats.py.
- All return values are JSON-serialisable (DataFrames are saved, not returned).
"""

from __future__ import annotations

import sys
from datetime import datetime
from pathlib import Path
from typing import Annotated, Literal

import pandas as pd
from fastmcp import FastMCP

_PROJECT_ROOT = Path(__file__).parent.parent.parent.resolve()
if str(_PROJECT_ROOT) not in sys.path:
    sys.path.insert(0, str(_PROJECT_ROOT))

import os as _os

from src.core.constants import DECOUPLER_DISCLAIMER_PER_SAMPLE
from src.workflows.activity_scoring import (
    plot_activity_landscape,
    score_bulk_samples_with_decoupler,
)
from src.workflows.activity_stats import compare_activity_by_group
from src.workflows.signatures import score_samples_against_signature


def _resolve_output_dir() -> Path:
    """Return a writable output directory: env var > project default > /tmp fallback."""
    env_val = _os.environ.get("RNA_OUTPUT_DIR")
    if env_val:
        p = Path(env_val)
        p.mkdir(parents=True, exist_ok=True)
        return p
    project_default = _PROJECT_ROOT / "tmp" / "outputs"
    try:
        project_default.mkdir(parents=True, exist_ok=True)
        return project_default
    except PermissionError:
        fallback = Path("/tmp/decoupleRpy/outputs")
        fallback.mkdir(parents=True, exist_ok=True)
        return fallback


OUTPUT_DIR = _resolve_output_dir()

_timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")

bulk_dataset_mcp = FastMCP(name="bulk_dataset")


# ---------------------------------------------------------------------------
# Internal helpers
# ---------------------------------------------------------------------------


def _resolve_to_local_path(path_or_url: str) -> tuple[str, bool]:
    """
    Resolve a local path or remote URL to a readable local file.

    Returns ``(local_path, is_temp)``.  When ``is_temp`` is True the caller must
    delete ``local_path`` after reading.  http(s)/ftp URLs (e.g. an h5ad hosted
    on the HuggingFace dataset repo) are streamed to a temp file, since scanpy's
    h5ad reader requires a local path.  Local paths are returned unchanged — the
    caller is responsible for its own existence check.
    """
    from src.core.data_io import resolve_to_local_path

    return resolve_to_local_path(path_or_url)


def _load_expression_df(expression_path: str) -> pd.DataFrame:
    """
    Load a samples × genes expression matrix from .h5ad, .csv, or .tsv.

    Accepts a local path or an http(s)/ftp URL (remote files are downloaded to a
    temp file first).  For .h5ad files the dense matrix is relabelled to gene
    symbols when ``var.index`` holds Ensembl/probe IDs and a symbol column is
    available (see ``resolve_symbols_from_var``).

    Returns ``(dataframe, symbol_info)`` where ``symbol_info`` is the resolution
    report for h5ad inputs and ``None`` for CSV/TSV.
    """
    local_path, is_temp = _resolve_to_local_path(expression_path)
    if not is_temp and not Path(local_path).exists():
        raise FileNotFoundError(f"Expression file not found: {expression_path}")

    suffix = Path(expression_path.split("?", 1)[0]).suffix.lower()
    try:
        if suffix == ".h5ad":
            import scanpy as sc

            from src.workflows.microarray import resolve_symbols_from_var

            adata = sc.read_h5ad(local_path)
            df = adata.to_df()
            # An h5ad may be indexed by Ensembl IDs with the symbol parked in a
            # var column (gse205154_sears: var.index is versioned Ensembl,
            # var['SYMBOL'] holds the symbol). Activity scoring matches network
            # genes by NAME, so without this the tool dies with "zero gene
            # overlap" and the agent has to hand-roll the remap — which is
            # exactly what happened on the live prod run of this query.
            resolved = resolve_symbols_from_var(df, adata.var)
            resolved["obs"] = adata.obs.copy()
            return resolved["dataframe"], resolved
        elif suffix in (".tsv", ".txt"):
            return pd.read_csv(local_path, sep="\t", index_col=0), None
        else:
            return pd.read_csv(local_path, index_col=0), None
    finally:
        if is_temp:
            Path(local_path).unlink(missing_ok=True)


# ---------------------------------------------------------------------------
# MCP tool
# ---------------------------------------------------------------------------


@bulk_dataset_mcp.tool
def dataset_score_bulk_samples(
    expression_path: Annotated[
        str,
        "Path to a pre-normalised samples × genes expression file "
        "(.h5ad, .csv, or .tsv). Samples must be rows, gene symbols must "
        "be columns. Expression must already be normalised (log-CPM, "
        "log-intensity, log2-TPM, etc.). Do NOT pass raw integer counts — "
        "decoupleR activity estimation requires continuous normalised input.",
    ],
    resource: Annotated[
        Literal["progeny", "collectri", "hallmark"],
        "Prior knowledge resource to score against: "
        "'progeny' — 14 cancer signalling pathway signatures (PROGENy), "
        "'collectri' — TF-target regulons from CollecTRI (~1 000 TFs), "
        "'hallmark' — 50 MSigDB Hallmark gene sets.",
    ] = "progeny",
    organism: Annotated[
        Literal["human", "mouse"],
        "Organism for the prior knowledge resource.",
    ] = "human",
    method: Annotated[
        Literal["ulm", "mlm", "zscore"],
        "Activity scoring method: "
        "'ulm' (Univariate Linear Model — recommended, fastest), "
        "'mlm' (Multivariate Linear Model — accounts for co-linearity), "
        "'wsum' (Weighted Sum — no distributional assumptions).",
    ] = "ulm",
    input_scale: Annotated[
        Literal["auto", "log", "linear"],
        "Scale of the input expression matrix. "
        "'auto' (recommended) detects the scale and log2(x+1)-transforms the "
        "matrix when it looks linear (e.g. raw TPM), since ULM assumes roughly "
        "symmetric log-scale input; 'log' trusts the matrix as-is; 'linear' "
        "always log2(x+1)-transforms. The transform actually applied is "
        "reported as 'applied_transform' in the return value.",
    ] = "auto",
    exclude_samples_column: Annotated[
        str | None,
        "Metadata column to match `exclude_samples_matching` against. Default "
        "None matches the expression matrix row index (sample IDs). Set this "
        "when the marker lives in a metadata column instead — e.g. in "
        "gse205154_sears the row index is GSM accessions and the fibroblast "
        "'-F' suffix is only in obs['sample_title'], so excluding them needs "
        "exclude_samples_column='sample_title'. Only available for .h5ad input.",
    ] = None,
    exclude_samples_matching: Annotated[
        str | None,
        "Optional regular expression matched against sample IDs (the expression "
        "matrix row index); matching samples are dropped BEFORE scoring. Use to "
        "segment out sample types that a dataset's metadata does not separate — "
        "e.g. '-F$' drops the 7 fibroblast samples in gse205154_sears, which GEO "
        "annotates as 'Primary' and which can inflate stromal pathway scores "
        "(TGFb, NFkB). Excluded sample IDs are reported in 'excluded_samples'.",
    ] = None,
    out_prefix: Annotated[
        str | None,
        "Output file prefix for saved CSVs. Defaults to '<resource>_<method>_<timestamp>'.",
    ] = None,
) -> dict:
    """
    Score bulk expression samples for pathway, TF, or hallmark activity.

    Call this when:
    - You have a pre-normalised bulk expression matrix (samples × genes) and
      want per-sample activity scores across a cohort.
    - You want PROGENy pathway scores, CollecTRI TF regulon scores, or
      Hallmark gene set scores — without first computing DE statistics.
    - You are preparing activity matrices for downstream analysis such as
      subtype comparison, survival correlation, or visualisation.

    This tool complements the DE-based enrichment tools
    (decoupler_tf_enrichment_collectri, decoupler_pathway_enrichment_progeny,
    decoupler_hallmark_enrichment) which score from DE statistics.  Use this
    tool when you want sample-level scores rather than contrast-level scores.

    Input expression must be pre-normalised.  Raw read counts will produce
    incorrect activity scores — run normalisation first.

    Returns
    -------
    Activity scores (samples × activities) and p-values are saved as CSV files
    in tmp/outputs/, together with a cohort-wide **landscape figure**
    (per-sample heatmap + mean±SD bar) at 'landscape_path'.  Use that figure
    when the user asks to "see the landscape" — do NOT hand-roll a plot; a
    figure written outside tmp/outputs/ is never surfaced in the UI.
    The return dict includes artifact paths, gene coverage statistics, the
    input-scale transform applied, the pinned PROGENy footprint, and warnings.

    Next step: use the returned activity_path with downstream tools, or call
    dataset_validate_contrast() to define groups before comparing scores
    between conditions.
    """
    # ── Load expression ───────────────────────────────────────────────────
    try:
        expr_df, symbol_info = _load_expression_df(expression_path)
    except FileNotFoundError as exc:
        return {
            "valid": False,
            "error": str(exc),
            "expression_path": expression_path,
        }
    except Exception as exc:
        return {
            "valid": False,
            "error": f"Failed to load expression from '{expression_path}': {exc}",
            "expression_path": expression_path,
        }

    # ── Optional sample exclusion ─────────────────────────────────────────
    # Some cohorts label sample types only in the sample ID (gse205154_sears
    # annotates its 7 '-F' fibroblast samples as 'Primary' in GEO), so there is
    # no metadata column to filter on. This hook makes the exclusion explicit
    # and auditable rather than silent.
    excluded_samples: list[str] = []
    exclusion_warnings: list[str] = []
    if exclude_samples_matching:
        obs = (symbol_info or {}).get("obs")
        if exclude_samples_column:
            if obs is None:
                return {
                    "valid": False,
                    "error": (
                        f"exclude_samples_column='{exclude_samples_column}' requires .h5ad "
                        "input (CSV/TSV carry no metadata table)."
                    ),
                }
            if exclude_samples_column not in obs.columns:
                return {
                    "valid": False,
                    "error": (
                        f"exclude_samples_column='{exclude_samples_column}' not found. "
                        f"Available metadata columns: {list(obs.columns)}"
                    ),
                }
            match_against = obs[exclude_samples_column].reindex(expr_df.index).astype(str)
        else:
            match_against = expr_df.index.to_series().astype(str)

        try:
            mask = match_against.str.contains(exclude_samples_matching, regex=True).fillna(False)
        except Exception as exc:
            return {
                "valid": False,
                "error": f"Invalid exclude_samples_matching regex '{exclude_samples_matching}': {exc}",
            }
        mask = mask.to_numpy()

        # A filter that matches nothing must never pass silently: the caller
        # believes those samples are gone. Observed live — the agent passed
        # '-F$' against a GSM-accession index while the '-F' suffix only exists
        # in obs['sample_title'], so zero fibroblasts were excluded and nothing
        # said so.
        if not mask.any():
            where = (
                f"metadata column '{exclude_samples_column}'"
                if exclude_samples_column
                else "the sample-ID index"
            )
            sample_vals = list(match_against[:3])
            exclusion_warnings.append(
                f"exclude_samples_matching='{exclude_samples_matching}' matched NO samples "
                f"against {where} — nothing was excluded. Values look like {sample_vals}. "
                "If the marker lives in a metadata column, pass exclude_samples_column."
            )
        excluded_samples = expr_df.index[mask].astype(str).tolist()
        expr_df = expr_df.loc[~mask]
        if expr_df.empty:
            return {
                "valid": False,
                "error": (
                    f"exclude_samples_matching='{exclude_samples_matching}' removed "
                    f"all {len(excluded_samples)} samples — nothing left to score."
                ),
                "excluded_samples": excluded_samples,
            }

    # ── Score ─────────────────────────────────────────────────────────────
    try:
        result = score_bulk_samples_with_decoupler(
            expression_df=expr_df,
            resource=resource,
            organism=organism,
            method=method,
            input_scale=input_scale,
        )
    except ValueError as exc:
        return {
            "valid": False,
            "error": str(exc),
            "expression_path": expression_path,
        }
    except Exception as exc:
        return {
            "valid": False,
            "error": f"Scoring failed: {exc}",
            "expression_path": expression_path,
        }

    # ── Save outputs ──────────────────────────────────────────────────────
    prefix = out_prefix or f"{resource}_{method}_{_timestamp}"
    activity_path = OUTPUT_DIR / f"{prefix}_activities.csv"
    pvalue_path = OUTPUT_DIR / f"{prefix}_pvalues.csv"

    result["activity_df"].to_csv(activity_path)
    result["pvalue_df"].to_csv(pvalue_path)

    # ── Landscape figure ──────────────────────────────────────────────────
    # Emitted by the tool (not left to agent-authored matplotlib) so it always
    # lands in OUTPUT_DIR and is picked up by the end-of-run inline-plot sweep.
    landscape_path = OUTPUT_DIR / f"{prefix}_landscape.png"
    warnings_out = list(result["warnings"])
    if symbol_info:
        warnings_out.extend(symbol_info["warnings"])
    landscape_info: dict | None = None
    try:
        landscape_info = plot_activity_landscape(
            result["activity_df"],
            landscape_path,
            title=f"{resource} activity landscape ({method}, n={result['n_samples']} samples)",
        )
        if landscape_info["truncated"]:
            warnings_out.append(
                f"Landscape figure shows the {landscape_info['n_features_plotted']} "
                f"most variable of {landscape_info['n_features_total']} activities; "
                "all activities are in the CSV."
            )
    except Exception as exc:  # plotting must never fail the analysis
        warnings_out.append(f"Landscape figure could not be rendered: {exc}")

    warnings_out.extend(exclusion_warnings)
    if excluded_samples:
        warnings_out.append(
            f"Excluded {len(excluded_samples)} sample(s) matching "
            f"'{exclude_samples_matching}' before scoring: {excluded_samples}."
        )

    return {
        "valid": True,
        "expression_path": expression_path,
        "n_samples": result["n_samples"],  # samples in expression matrix
        "n_obs": result["n_samples"],  # alias kept for backward compatibility
        "shape": (result["n_samples"], result["n_activities"]),
        "n_activities": result["n_activities"],
        "n_network_genes": result["n_network_genes"],
        "n_matched_genes": result["n_matched_genes"],
        "coverage_pct": result["coverage_pct"],
        "resource": result["resource"],
        "organism": result["organism"],
        "method": result["method"],
        "input_scale_requested": result["input_scale_requested"],
        "input_scale_detected": result["input_scale_detected"],
        "applied_transform": result["applied_transform"],
        "expression_max": result["expression_max"],
        "network_top": result["network_top"],
        "excluded_samples": excluded_samples,
        "n_excluded_samples": len(excluded_samples),
        "exclude_samples_column": exclude_samples_column,
        "csv_orientation": "rows = samples, columns = activity names",
        "gene_symbols_resolved": bool(symbol_info and symbol_info["renamed"]),
        "gene_symbol_column": (symbol_info or {}).get("symbol_column"),
        "method_caveat": DECOUPLER_DISCLAIMER_PER_SAMPLE,
        "warnings": warnings_out,
        "activity_path": str(activity_path.resolve()),
        "pvalue_path": str(pvalue_path.resolve()),
        "landscape_path": (str(landscape_path.resolve()) if landscape_info else None),
        "artifacts": [
            {
                "description": (
                    f"{resource} activity scores ({method}) — "
                    "samples × activities, save for downstream comparison"
                ),
                "path": str(activity_path.resolve()),
            },
            {
                "description": f"{resource} activity p-values ({method})",
                "path": str(pvalue_path.resolve()),
            },
        ]
        + (
            [
                {
                    "description": (
                        f"{resource} cohort-wide activity landscape ({method}) — "
                        "per-sample heatmap + mean±SD bar"
                    ),
                    "path": landscape_info["path"],
                }
            ]
            if landscape_info
            else []
        ),
        "next_step": (
            f"Activity scores for {result['n_activities']} {resource} features "
            f"saved to activity_path, and the cohort-wide landscape figure to "
            "landscape_path (show that figure rather than plotting your own). "
            "To compare scores between sample groups, call "
            "dataset_compare_activity_by_group(activity_path, metadata_path, ...)."
        ),
    }


# ---------------------------------------------------------------------------
# Custom-signature scoring (ADR-0006 Role 2)
# ---------------------------------------------------------------------------


@bulk_dataset_mcp.tool
def dataset_score_signature(
    expression_path: Annotated[
        str,
        "Path to a pre-normalised samples × genes expression file "
        "(.h5ad, .csv, or .tsv). Samples are rows, gene symbols are columns. "
        "Must be normalised (log-CPM, log-intensity, log2-TPM, etc.) — not raw "
        "integer counts.",
    ],
    signature_path: Annotated[
        str,
        "Path to a signature table (.csv or .tsv) with columns "
        "'source, target[, weight]' — one row per gene in a signature. "
        "'source' is the signature name (e.g. a Loveless-derived cell-state "
        "program), 'target' is an HGNC gene symbol, and 'weight' (optional) is "
        "the gene's signed membership/loading. This is a precomputed, "
        "offline-derived signature artifact — NOT the single-cell atlas itself.",
    ],
    method: Annotated[
        Literal["ulm", "mlm", "zscore"],
        "Activity scoring method (same engine as dataset_score_bulk_samples).",
    ] = "ulm",
    organism: Annotated[
        Literal["human", "mouse"],
        "Organism label, carried through to diagnostics.",
    ] = "human",
    label: Annotated[
        str,
        "Free-form name for this signature set, used in output filenames (e.g. 'loveless_basal').",
    ] = "signature",
    out_prefix: Annotated[
        str | None,
        "Output file prefix for saved CSVs. Defaults to '<label>_<method>_<timestamp>'.",
    ] = None,
) -> dict:
    """
    Score a bulk cohort against a precomputed single-cell-derived signature.

    This is the **bulk fast path** for the Loveless resource (ADR-0006 Role 2):
    the full single-cell atlas is NOT loaded at query time. Heavy single-cell
    work is precomputed offline at ingestion into a signature table, and this
    tool scores that signature against any pre-normalised bulk cohort using the
    same decoupleR engine as dataset_score_bulk_samples — so per-sample signature
    activity in a bulk dataset costs the same as a normal activity-scoring call.

    Call this for requests like "score the Loveless basal / CXCL10+ CAF signature
    in TCGA-PAAD". For built-in resources (PROGENy / CollecTRI / Hallmark) use
    dataset_score_bulk_samples instead; for analysing the single-cell *subset*
    itself (pseudobulk DE), use the single-cell tools.

    Gene-coverage diagnostics, the zero-overlap guard, and the small-cohort
    warning are identical to dataset_score_bulk_samples. Low coverage between the
    signature genes and the cohort is surfaced in 'warnings'.

    Returns
    -------
    Per-sample signature activity scores and p-values saved as CSVs in
    tmp/outputs/, with coverage statistics and the number of signatures scored.

    Next step: compare scores between groups with
    dataset_compare_activity_by_group(activity_path, metadata_path, ...).
    """
    # ── Load expression ───────────────────────────────────────────────────
    try:
        expr_df, symbol_info = _load_expression_df(expression_path)
    except FileNotFoundError as exc:
        return {"valid": False, "error": str(exc), "expression_path": expression_path}
    except Exception as exc:
        return {
            "valid": False,
            "error": f"Failed to load expression from '{expression_path}': {exc}",
            "expression_path": expression_path,
        }

    # ── Score against signature ────────────────────────────────────────────
    try:
        result = score_samples_against_signature(
            expression_df=expr_df,
            signature=signature_path,
            method=method,
            organism=organism,
            label=label,
        )
    except FileNotFoundError as exc:
        return {"valid": False, "error": str(exc), "signature_path": signature_path}
    except ValueError as exc:
        return {"valid": False, "error": str(exc), "signature_path": signature_path}
    except Exception as exc:
        return {"valid": False, "error": f"Signature scoring failed: {exc}"}

    # ── Save outputs ──────────────────────────────────────────────────────
    prefix = out_prefix or f"{label}_{method}_{_timestamp}"
    activity_path = OUTPUT_DIR / f"{prefix}_activities.csv"
    pvalue_path = OUTPUT_DIR / f"{prefix}_pvalues.csv"

    result["activity_df"].to_csv(activity_path)
    result["pvalue_df"].to_csv(pvalue_path)

    return {
        "valid": True,
        "expression_path": expression_path,
        "signature_path": signature_path,
        "n_samples": result["n_samples"],
        "n_signatures": result["n_signatures"],
        "n_activities": result["n_activities"],
        "n_network_genes": result["n_network_genes"],
        "n_matched_genes": result["n_matched_genes"],
        "coverage_pct": result["coverage_pct"],
        "label": label,
        "organism": result["organism"],
        "method": result["method"],
        "csv_orientation": "rows = samples, columns = signature names",
        "gene_symbols_resolved": bool(symbol_info and symbol_info["renamed"]),
        "gene_symbol_column": (symbol_info or {}).get("symbol_column"),
        "method_caveat": DECOUPLER_DISCLAIMER_PER_SAMPLE,
        "warnings": result["warnings"] + ((symbol_info or {}).get("warnings") or []),
        "activity_path": str(activity_path.resolve()),
        "pvalue_path": str(pvalue_path.resolve()),
        "artifacts": [
            {
                "description": (
                    f"'{label}' signature activity scores ({method}) — samples × signatures"
                ),
                "path": str(activity_path.resolve()),
            },
            {
                "description": f"'{label}' signature activity p-values ({method})",
                "path": str(pvalue_path.resolve()),
            },
        ],
        "next_step": (
            f"Per-sample scores for {result['n_signatures']} signature(s) saved to "
            "activity_path. To compare scores between sample groups, call "
            "dataset_compare_activity_by_group(activity_path, metadata_path, ...)."
        ),
    }


# ---------------------------------------------------------------------------
# Internal helper for metadata loading
# ---------------------------------------------------------------------------


def _load_metadata_from_file(metadata_path: str) -> pd.DataFrame:
    """
    Load sample metadata from .h5ad (returns adata.obs), .csv, or .tsv.

    Accepts a local path or an http(s)/ftp URL.  Returns a DataFrame with sample
    IDs as the index.
    """
    local_path, is_temp = _resolve_to_local_path(metadata_path)
    if not is_temp and not Path(local_path).exists():
        raise FileNotFoundError(f"Metadata file not found: {metadata_path}")

    suffix = Path(metadata_path.split("?", 1)[0]).suffix.lower()
    try:
        if suffix == ".h5ad":
            import scanpy as sc

            return sc.read_h5ad(local_path).obs.copy()
        elif suffix in (".tsv", ".txt"):
            return pd.read_csv(local_path, sep="\t", index_col=0)
        else:
            return pd.read_csv(local_path, index_col=0)
    finally:
        if is_temp:
            Path(local_path).unlink(missing_ok=True)


# ---------------------------------------------------------------------------
# Group comparison tool
# ---------------------------------------------------------------------------


@bulk_dataset_mcp.tool
def dataset_compare_activity_by_group(
    activity_path: Annotated[
        str,
        "Path to a samples × activities CSV file produced by "
        "dataset_score_bulk_samples (activity_path from its return value). "
        "Rows are samples, columns are activity names (pathway names, TF names, "
        "hallmark names).",
    ],
    metadata_path: Annotated[
        str,
        "Path to a sample metadata file (.h5ad, .csv, or .tsv). "
        "Index must contain sample IDs that match activity_path rows. "
        "For .h5ad files, adata.obs is used.",
    ],
    group_column: Annotated[
        str,
        "Column in the metadata file containing the group labels "
        "(e.g. 'tumor subtype'). Use dataset_list_valid_sample_groups() "
        "to discover available columns and their values.",
    ],
    test_group: Annotated[
        str,
        "Label of the foreground / test condition (e.g. 'Classical').",
    ],
    control_group: Annotated[
        str,
        "Label of the reference / background condition (e.g. 'Basal'). "
        "Must differ from test_group.",
    ],
    subset_query: Annotated[
        str | None,
        "Optional pandas query string to restrict samples before comparison, "
        "e.g. \"tissue == 'tumor'\". Applied before group membership is checked.",
    ] = None,
    method: Annotated[
        str,
        "Statistical method. Currently only 'welch_ttest' is supported.",
    ] = "welch_ttest",
    out_prefix: Annotated[
        str | None,
        "Output file prefix for the results CSV. Defaults to 'activity_comparison_<timestamp>'.",
    ] = None,
) -> dict:
    """
    Compare TF/pathway/hallmark activity scores between two metadata-defined
    sample groups.

    Call this after dataset_score_bulk_samples when the user asks whether
    a TF, pathway, or hallmark activity differs significantly between two
    conditions (e.g. Classical vs Basal PDAC, treated vs untreated).

    Runs Welch's t-test across all activities simultaneously and applies
    Benjamini-Hochberg FDR correction.  Effect size is Cohen's d — positive
    values indicate higher activity in the test group.

    Returns a table sorted by padj with columns:
    activity, mean_test, mean_control, effect_size (Cohen's d), statistic,
    pvalue, padj, n_test, n_control.

    Next step: examine significant activities (padj < 0.05) with the largest
    |effect_size|.  Use the comparison_path CSV for further visualisation.
    """
    # ── Load inputs ────────────────────────────────────────────────────────
    try:
        act_df = pd.read_csv(activity_path, index_col=0)
    except FileNotFoundError:
        return {"valid": False, "error": f"Activity file not found: {activity_path}"}
    except Exception as exc:
        return {"valid": False, "error": f"Failed to load activity file: {exc}"}

    try:
        meta_df = _load_metadata_from_file(metadata_path)
    except FileNotFoundError:
        return {"valid": False, "error": f"Metadata file not found: {metadata_path}"}
    except Exception as exc:
        return {"valid": False, "error": f"Failed to load metadata file: {exc}"}

    # ── Compare ─────────────────────────────────────────────────────────────
    try:
        result = compare_activity_by_group(
            activity_df=act_df,
            metadata_df=meta_df,
            group_column=group_column,
            test_group=test_group,
            control_group=control_group,
            subset_query=subset_query,
            method=method,
        )
    except ValueError as exc:
        return {"valid": False, "error": str(exc)}
    except Exception as exc:
        return {"valid": False, "error": f"Comparison failed: {exc}"}

    # ── Save results ─────────────────────────────────────────────────────────
    prefix = out_prefix or f"activity_comparison_{_timestamp}"
    comparison_path = OUTPUT_DIR / f"{prefix}_results.csv"
    result["dataframe"].to_csv(comparison_path)

    # Top hits: significant activities sorted by |effect_size|
    sig = result["dataframe"][result["dataframe"]["padj"] < 0.05]
    top_hits = (
        sig.reindex(sig["effect_size"].abs().sort_values(ascending=False).index)
        .head(10)
        .index.tolist()
    )

    # ── The reporting table, returned INLINE ──────────────────────────────────
    # This tool is the end of the analysis chain: whatever it returns is what the
    # agent writes its <solution> from. Returning only activity NAMES meant every
    # run had to re-open comparison_path to quote an effect size or a padj — a
    # 2-4 step reporting tail that three rounds of prompt tightening did not fix
    # (the rule said "re-read once"; runs did 2, then 3). Prompting is the wrong
    # lever: the numbers exist right here. So hand back the actual rows.
    #
    # When nothing clears padj < 0.05 we still return the strongest rows, flagged
    # via `top_table_is_significant: False` — otherwise "no significant hits" is
    # itself a reason to go re-read the CSV, which is the loop we are closing.
    _ranked = result["dataframe"].reindex(
        result["dataframe"]["effect_size"].abs().sort_values(ascending=False).index
    )
    _source = _ranked.loc[_ranked.index.isin(sig.index)] if len(sig) else _ranked

    def _num(value):
        """Round for prompt economy; keep None rather than emitting NaN."""
        try:
            f = float(value)
        except (TypeError, ValueError):
            return None
        if f != f:  # NaN
            return None
        return round(f, 4)

    _cols = [
        c for c in ("mean_test", "mean_control", "effect_size", "padj") if c in _ranked.columns
    ]
    top_table = [
        {"activity": str(name), **{c: _num(row[c]) for c in _cols}}
        for name, row in _source.head(10).iterrows()
    ]

    return {
        "valid": True,
        "n_activities": result["n_activities"],
        "n_test": result["n_test"],
        "n_control": result["n_control"],
        "n_significant_05": result["significant_05"],
        "n_significant_01": result["significant_01"],
        "top_significant_activities": top_hits,
        "top_table": top_table,
        "top_table_is_significant": bool(len(sig)),
        "top_table_columns": ["activity", *_cols],
        "group_column": group_column,
        "test_group": test_group,
        "control_group": control_group,
        "subset_query": subset_query,
        "method": method,
        "warnings": result["warnings"],
        "comparison_path": str(comparison_path.resolve()),
        "artifacts": [
            {
                "description": (
                    f"Activity comparison: {test_group} vs {control_group} — "
                    "activity, mean_test, mean_control, effect_size, padj"
                ),
                "path": str(comparison_path.resolve()),
            }
        ],
        "next_step": (
            f"Found {result['significant_05']} significant activities (padj < 0.05). "
            + (
                "`top_table` holds the top significant rows with their "
                "mean_test, mean_control, effect_size and padj. "
                if len(sig)
                else "Nothing reached padj < 0.05; `top_table` holds the strongest "
                "rows anyway (top_table_is_significant = false) so the null result "
                "can be reported with numbers. "
            )
            + "Those numbers are everything the solution needs — do NOT re-open "
            "comparison_path to write the report. Load that CSV only to plot, or "
            "if you genuinely need a row beyond the top 10."
        ),
    }