src/core/top_table.py

"""
Inline reporting tables for the tools that terminate an analysis chain.

Why this exists — measured on the dev Space 2026-08-06, not assumed. The
"reporting tail" (2 extra steps between the last computation and <solution>) is
NOT a context-window problem: on an 8-step run only 11 messages exist when the
tail starts, and `memory_window` is 15, so nothing has been truncated and the
results observation is still in full view. The agent re-reads anyway because the
tool result contains no numbers to quote — `decoupler_differential_expression`
returns counts and paths, and the enrichment tools return
`top_result_names` (names only) plus TWO separate CSVs (activities and padj)
that have to be joined before anything can be said about significance.

That also explains why the tail is consistently 2 steps rather than 1: the first
re-read produces an unusable view (a `.head()` of an alphabetically-ordered padj
file, or a sort by a padj column that is 0.0 for thousands of genes), so a
second step is needed to join, filter and sort. Both steps do real work. The fix
is to hand back the joined, sorted, significance-filtered rows directly.

This mirrors what `dataset_compare_activity_by_group` already does with its own
`top_table`; these helpers exist so the three enrichment tools and the DE tool
share one implementation rather than four hand-rolled copies.
"""

from __future__ import annotations

# Rows per direction. Sized against the executor's MAX_OUTPUT_CHARS (4000): at
# 15 per direction the printed tool result overran the cap and was truncated,
# silently dropping the tail of the dict. 10 per direction keeps every measured
# result inside the cap, and 20 rows is also what these queries ask for
# ("report the top 20 genes").
TOP_N = 10


def _num(value, dp: int = 4):
    """Round for prompt economy; return None rather than a NaN (not JSON-safe)."""
    try:
        f = float(value)
    except (TypeError, ValueError):
        return None
    if f != f:  # NaN
        return None
    return round(f, dp)


def _both_directions(frame, sort_col: str, n: int) -> list:
    """Top n rows by `sort_col` from each end, highest first, no duplicates."""
    ordered = frame.sort_values(sort_col, ascending=False)
    if len(ordered) <= 2 * n:
        return list(ordered.index)
    return list(ordered.index[:n]) + list(ordered.index[-n:])


def de_top_table(results_df, n: int = TOP_N) -> tuple[list[dict], bool]:
    """Top up- and down-regulated genes with their statistics.

    Ranked by `stat` rather than `padj`: on a strong contrast padj is 0.0 for
    thousands of genes, so sorting by it returns an arbitrary slice of a tie —
    which is exactly what sent one measured run back for a second re-read.

    Returns (rows, is_significant). When nothing clears padj < 0.05 the strongest
    rows are returned anyway with is_significant False, so a null result can
    still be reported with numbers instead of triggering a trip to the CSV.
    """
    # pvalue is deliberately omitted: padj is what gets reported, and on a strong
    # contrast both are 0.0 — it cost characters against the output cap for nothing.
    cols = [c for c in ("log2FoldChange", "stat", "padj") if c in results_df.columns]
    if "stat" not in results_df.columns or results_df.empty:
        return [], False

    sig = results_df[results_df["padj"] < 0.05] if "padj" in results_df.columns else results_df
    source, is_sig = (sig, True) if len(sig) else (results_df, False)

    rows = [
        {"gene": str(g), **{c: _num(source.loc[g, c]) for c in cols}}
        for g in _both_directions(source, "stat", n)
    ]
    return rows, is_sig


def activity_top_table(acts, padj, significant_features, n: int = TOP_N) -> tuple[list[dict], bool]:
    """Join an activity frame with its padj frame into rows the agent can quote.

    `acts` and `padj` are decoupleR's (contrasts x features) frames; the first
    row is the contrast. They are written to two separate CSVs, so without this
    join the agent has to read both files back and merge them itself.
    """
    if acts is None or len(acts) == 0 or acts.shape[1] == 0:
        return [], False

    sig = [f for f in significant_features if f in acts.columns]
    source, is_sig = (sig, True) if sig else (list(acts.columns), False)

    # Select the contrast row POSITIONALLY. Label-based .loc returns a DataFrame
    # rather than a Series when the contrast label is duplicated, and the
    # sort_values() below then raises — which fails the whole tool and sends the
    # agent into a retry loop. iloc cannot degrade that way.
    acts_row = acts.iloc[0]
    padj_row = padj.iloc[0] if padj is not None and len(padj) else None

    scores = acts_row[source]
    ordered = scores.sort_values(ascending=False)
    picked = (
        list(ordered.index)
        if len(ordered) <= 2 * n
        else list(ordered.index[:n]) + list(ordered.index[-n:])
    )

    rows = []
    for feature in picked:
        row = {"feature": str(feature), "score": _num(scores[feature])}
        try:
            row["padj"] = _num(padj_row[feature]) if padj_row is not None else None
        except (KeyError, TypeError, IndexError):
            row["padj"] = None
        rows.append(row)
    return rows, is_sig


def top_table_next_step(kind: str, n_significant: int, is_significant: bool, path_hint: str) -> str:
    """The `next_step` string that tells the agent it already has the numbers."""
    head = f"Found {n_significant} significant {kind} (padj < 0.05). "
    if is_significant:
        body = (
            "`top_table` holds the strongest significant rows in BOTH directions, "
            "already joined with padj and sorted — no merge needed. "
        )
    else:
        body = (
            "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. "
        )
    return (
        head + body + "Those numbers are everything the solution needs — do NOT re-open "
        f"{path_hint} to write the report. Load that CSV only to plot, or if you "
        "genuinely need a row beyond the table."
    )