Run a gene set analysis end to end: upload validation and the job

Second slice of spec 012. Two layers, neither importing Chainlit, because
the awkward parts -- a job that outlasts a chat turn, a submission that
succeeds and then fails, a result that must be three sizes for three
audiences -- are the parts worth testing, and a browser is not needed to
test them.

**`upload.py`** caps a file at 20 MB against Chainlit's shipped 500 MB. The
host has 4.7 GB free of 88 GB and `~/update-beta-chat.sh` refuses to deploy
under 6 GB, so a handful of default-sized uploads would take the chat down
*and* block the fix. Shape is checked here rather than by the service,
which answers a malformed matrix minutes later through a status field, in
R's words.

**`job.py`** owns one analysis. `Finished` splits by audience: `for_model`
bounded and allow-listed, `links` the capability URL the model must never
see, `table_path` the full table for the user.

Found while writing it: a one-sample file was told it was not an expression
matrix -- it is one, it just cannot be compared against anything -- and
`gene_count` returned `-1` for a file too long to finish counting, a
sentinel of the same type as a real count, leaving a function whose result
gets shown to someone. It is `None` now, and the type says so.

**Adversarial review of the finished branch found four more, and three are
the same mistake.**

*The upload leaked on the likeliest paths.* The group checks sat above the
`try`, so a wrong group name or a miscounted label list -- the two mistakes
a user actually makes -- returned an error and left their matrix on the
disk. The leak was on the paths people take most often.

*The loading loop had only the wall-clock bound.* That is the bug I had
fixed in `await_result` earlier in the same review, in the same file,
twelve lines away. Fixing a loop is not fixing the loops. Both now carry
`MAX_POLLS` as well.

*`finished` and `failed` were defined on the status objects and used by
nobody* -- the loops compared the strings themselves, so "done" had three
definitions. One `TERMINAL_STATUSES` now, and `failed` is derived as
"terminal and not complete", so a status added later is a failure by
default rather than an analysis that silently reports no results.

*Result tables were written and never removed.* The upload is deleted the
moment it is submitted and then the output was kept forever, at ~2 MB each,
on a host with 4.7 GB free. Deleting the input and hoarding the output is
not a disk policy. `prune_results` runs before each write, by age then by
total size, and only ever touches files this module named.

**And the sabotage found a fifth.** Removing the `prune_results` call from
`await_result` broke nothing: every pruning test called the function
directly, so the directory could have grown forever with the suite green.
That it works is not the same as it being called.

Sabotage, each against the test written for it: validation back above the
`try` fails both deletion tests; removing the prune call fails the new
one; unbinding the loading loop hangs, which is the failure mode the bound
converts into an error.

67 tests in tests/gsa, ./checks.sh clean.

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

Author: adamjohnwright

src/gsa/client.py

@@ -105,6 +105,11 @@ def _checked(kind: str, value: str) -> str:
     return value
 
 
+#: Statuses that mean the service has stopped working on it. Listed once so
+#: a loop cannot disagree with the dataclass about what "done" means.
+TERMINAL_STATUSES = frozenset({"complete", "failed"})
+
+
 @dataclass(frozen=True)
 class LoadingStatus:
     """Progress of `POST /data/load`, which is not instant."""
@@ -116,7 +121,11 @@ class LoadingStatus:
 
     @property
     def finished(self) -> bool:
-        return self.status in {"complete", "failed"}
+        return self.status in TERMINAL_STATUSES
+
+    @property
+    def failed(self) -> bool:
+        return self.finished and self.status != "complete"
 
 
 @dataclass(frozen=True)
@@ -146,11 +155,15 @@ class AnalysisStatus:
 
     @property
     def finished(self) -> bool:
-        return self.status in {"complete", "failed"}
+        return self.status in TERMINAL_STATUSES
 
     @property
     def failed(self) -> bool:
-        return self.status == "failed"
+        # Anything terminal that is not success. Derived rather than
+        # `== "failed"`, so a status added to TERMINAL_STATUSES later is
+        # treated as a failure by default instead of being silently
+        # reported as a completed analysis with no results.
+        return self.finished and self.status != "complete"
 
 
 class GsaClient:

src/gsa/job.py

@@ -0,0 +1,327 @@
+"""Run one analysis from start to finished file.
+
+Sits between the client, which knows the service, and the chat, which knows
+the user. Nothing here imports Chainlit: the awkward parts -- polling a job
+that outlasts a chat turn, a submission that succeeds and then fails, a
+result that has to be two different sizes for two different audiences --
+are the parts worth testing, and a browser is not needed to test them.
+
+**The submission is a receipt.** Measured: `POST /analysis` returned 200 and
+the analysis then failed with `CONNECTION_FORCED - broker forced connection
+closure`, visible only through `/status`. So `submit_*` returns an ID and
+promises nothing, and `await_result` is where success or failure is decided.
+"""
+
+import contextlib
+import time
+from collections.abc import Awaitable, Callable
+from dataclasses import dataclass
+from pathlib import Path
+
+from gsa import results as gsa_results
+from gsa.client import AnalysisStatus, GsaClient, GsaError
+from gsa.upload import Matrix, discard
+from util.logging import logging
+
+logger = logging.getLogger(__name__)
+
+#: A PADOG run with 1,000 permutations took minutes in the measured run.
+#: This is a ceiling on waiting, not an expectation.
+DEFAULT_DEADLINE_SECONDS = 30 * 60
+POLL_INTERVAL_SECONDS = 10.0
+
+#: A hard ceiling on how many times the service is asked, independent of the
+#: interval.
+#:
+#: Two bounds are needed, and the second one took two attempts. The
+#: wall-clock deadline assumes each turn of the loop waits; the moment
+#: sleeping does not sleep -- a patched `_sleep`, a zero interval from a
+#: caller -- elapsed time never advances and the loop hammers the service.
+#: My first fix derived the limit as `deadline / interval`, which computes
+#: the bound from the quantity that is degenerate: `poll_interval=0` gave a
+#: limit of 1.8 million, so the guard against a runaway loop was itself
+#: unbounded in exactly the case it existed for. It left three pytest
+#: processes spinning at 100% CPU on a shared host.
+#:
+#: At the real 10s interval a 30-minute deadline is 180 polls, so this is
+#: generous.
+MAX_POLLS = 2_000
+
+#: The service's own default, and the method the round trip was measured
+#: with. Camera is faster; choosing between them wants evidence rather than
+#: preference, so it is recorded as an open question in spec 012.
+DEFAULT_METHOD = "PADOG"
+
+ProgressCallback = Callable[[AnalysisStatus], Awaitable[None]]
+
+
+class AnalysisFailedError(GsaError):
+    """The service accepted the analysis and then could not finish it."""
+
+
+@dataclass(frozen=True)
+class Finished:
+    """A completed analysis, split by audience.
+
+    Three fields, three destinations, kept apart deliberately:
+
+        for_model   bounded, allow-listed, safe to put in a prompt
+        links       the user's Pathway Browser view -- a capability URL
+                    that must not reach the model
+        table_path  the full table, for the user to download
+    """
+
+    analysis_id: str
+    for_model: dict[str, object]
+    links: list[tuple[str, str]]
+    table_path: Path
+
+
+async def submit_public_dataset(
+    client: GsaClient,
+    *,
+    resource_id: str,
+    dataset_id: str,
+    factor: str,
+    group1: str,
+    group2: str,
+    method: str = DEFAULT_METHOD,
+    deadline_seconds: float = DEFAULT_DEADLINE_SECONDS,
+) -> str:
+    """Load a public dataset and submit it. Returns an analysis ID.
+
+    The matrix is downloaded here and handed straight to `submit`. It is
+    1.2 MB for a small dataset and never leaves this function.
+    """
+    loading_id = await client.load_public_dataset(resource_id, dataset_id)
+    # Both bounds, same as `await_result`. This loop had only the
+    # wall-clock one -- the bug I had just fixed in its sibling, in the same
+    # file, during the same review. Fixing a loop is not fixing the loops.
+    deadline = time.monotonic() + deadline_seconds
+    polls = 0
+    while True:
+        status = await client.loading_status(loading_id)
+        polls += 1
+        if status.failed:
+            raise AnalysisFailedError(
+                f"loading {dataset_id} failed: {status.description}"
+            )
+        if status.finished:
+            break
+        if polls >= MAX_POLLS or time.monotonic() > deadline:
+            raise AnalysisFailedError(f"loading {dataset_id} did not finish in time")
+        await _sleep(POLL_INTERVAL_SECONDS)
+
+    summary = await client.dataset_summary(dataset_id)
+    groups = summary.factors.get(factor)
+    if not groups:
+        raise AnalysisFailedError(
+            f"{dataset_id} has no factor called {factor!r}. "
+            f"It has: {', '.join(sorted(summary.factors)) or 'none'}."
+        )
+    _check_comparable(groups, factor, group1, group2)
+
+    matrix = await client.download_matrix(dataset_id)
+    return await client.submit(
+        method=method,
+        dataset_name=dataset_id,
+        dataset_type=summary.type,
+        matrix=matrix,
+        samples=summary.samples,
+        analysis_group=groups,
+        group1=group1,
+        group2=group2,
+    )
+
+
+async def submit_uploaded_matrix(
+    client: GsaClient,
+    *,
+    matrix: Matrix,
+    dataset_type: str,
+    analysis_group: list[str],
+    group1: str,
+    group2: str,
+    method: str = DEFAULT_METHOD,
+) -> str:
+    """Submit a user's own matrix, then delete their file.
+
+    The file goes whether or not the analysis then succeeds: the service has
+    its own copy by then, and this host does not have room to keep ours.
+    `dataset_name` is deliberately not the user's filename -- that is one of
+    the strings the disclosure rules exist to keep out of a prompt.
+    """
+    # Everything, including the validation, inside the `try`.
+    #
+    # The checks used to sit above it, so a wrong group name or a
+    # miscounted label list -- the two mistakes a user is most likely to
+    # make -- returned an error and left their file on disk. The disk leak
+    # happened on exactly the paths people take most often.
+    try:
+        _check_comparable(analysis_group, "the grouping you gave", group1, group2)
+        if len(analysis_group) != len(matrix.samples):
+            raise AnalysisFailedError(
+                f"You gave {len(analysis_group)} group labels for "
+                f"{len(matrix.samples)} samples. There must be one label per "
+                f"sample, in the order the columns appear."
+            )
+        return await client.submit(
+            method=method,
+            dataset_name="uploaded",
+            dataset_type=dataset_type,
+            matrix=matrix.text,
+            samples=matrix.samples,
+            analysis_group=analysis_group,
+            group1=group1,
+            group2=group2,
+        )
+    finally:
+        discard(matrix.path)
+
+
+def _check_comparable(groups: list[str], label: str, group1: str, group2: str) -> None:
+    distinct = set(groups)
+    missing = [g for g in (group1, group2) if g not in distinct]
+    if missing:
+        raise AnalysisFailedError(
+            f"{', '.join(missing)} is not a value of {label}. "
+            f"It has: {', '.join(sorted(distinct))}."
+        )
+    if group1 == group2:
+        raise AnalysisFailedError("The two groups to compare must be different.")
+
+
+#: How long a written result table is kept, and how much of them in total.
+#:
+#: The upload is deleted the moment it is submitted, and then the *output*
+#: was kept forever -- each table is up to ~2 MB, on a host with 4.7 GB
+#: free. Deleting the input and hoarding the output is not a disk policy.
+#:
+#: The window only has to outlast a user downloading their own results.
+RESULT_MAX_AGE_SECONDS = 24 * 60 * 60
+RESULT_DIR_MAX_BYTES = 200 * 1024 * 1024
+
+
+def prune_results(
+    out_dir: Path,
+    *,
+    max_age_seconds: float = RESULT_MAX_AGE_SECONDS,
+    max_total_bytes: int = RESULT_DIR_MAX_BYTES,
+    now: float | None = None,
+) -> int:
+    """Delete old result tables. Returns how many were removed.
+
+    Called before each write, so the directory bounds itself and there is
+    no cron job to forget. Age first, then oldest-first until the total
+    fits: age alone leaves a burst of results unbounded, and size alone
+    keeps one stale file forever on a quiet week.
+
+    Only ever touches files this module wrote, matched by name. A cleanup
+    that globs a directory it does not own is one misconfiguration away
+    from deleting something else.
+    """
+    moment = time.time() if now is None else now
+    ours = sorted(
+        (path for path in out_dir.glob("reactome-gsa-*.tsv") if path.is_file()),
+        key=lambda path: path.stat().st_mtime,
+    )
+
+    removed = 0
+    surviving: list[Path] = []
+    for path in ours:
+        if moment - path.stat().st_mtime > max_age_seconds:
+            with contextlib.suppress(OSError):
+                path.unlink()
+                removed += 1
+        else:
+            surviving.append(path)
+
+    total = sum(path.stat().st_size for path in surviving if path.exists())
+    for path in surviving:
+        if total <= max_total_bytes:
+            break
+        try:
+            size = path.stat().st_size
+            path.unlink()
+        except OSError:
+            continue
+        total -= size
+        removed += 1
+
+    if removed:
+        logger.info("pruned gsa result tables", extra={"removed": removed})
+    return removed
+
+
+async def await_result(
+    client: GsaClient,
+    analysis_id: str,
+    *,
+    out_dir: Path,
+    on_progress: ProgressCallback | None = None,
+    deadline_seconds: float = DEFAULT_DEADLINE_SECONDS,
+    poll_interval: float = POLL_INTERVAL_SECONDS,
+) -> Finished:
+    """Poll until the analysis finishes, then write the table and return.
+
+    Raises `AnalysisFailedError` when the service says so -- which is the only
+    place it ever says so.
+    """
+    deadline = time.monotonic() + deadline_seconds
+    max_polls = min(
+        MAX_POLLS, max(2, int(deadline_seconds / max(poll_interval, 1.0)) + 2)
+    )
+    polls = 0
+
+    while True:
+        status = await client.analysis_status(analysis_id)
+        polls += 1
+        if on_progress is not None:
+            await on_progress(status)
+        if status.failed:
+            raise AnalysisFailedError(status.description or "the analysis failed")
+        if status.finished:
+            # Terminal and not failed, so: complete. Asking the status
+            # object rather than comparing the string again keeps one
+            # definition of "done" across both loops and the dataclass.
+            break
+        if polls >= max_polls or time.monotonic() > deadline:
+            raise AnalysisFailedError(
+                f"the analysis did not finish within "
+                f"{deadline_seconds / 60:.0f} minutes; it may still be "
+                f"running as {analysis_id}"
+            )
+        await _sleep(poll_interval)
+
+    parsed = gsa_results.parse(await client.result(analysis_id))
+    if not parsed.pathways:
+        # Complete, and yet nothing to report. Better to say so than to
+        # hand back an empty file and an exact-sounding zero.
+        raise AnalysisFailedError("the analysis finished but returned no pathway table")
+
+    out_dir.mkdir(parents=True, exist_ok=True)
+    prune_results(out_dir)
+    table_path = out_dir / f"reactome-gsa-{analysis_id}.tsv"
+    table_path.write_text(gsa_results.as_tsv(parsed))
+    logger.info(
+        "gsa result written",
+        extra={
+            "analysis": analysis_id,
+            "pathways": len(parsed.pathways),
+            "bytes": table_path.stat().st_size,
+        },
+    )
+
+    return Finished(
+        analysis_id=analysis_id,
+        for_model=gsa_results.for_model(parsed),
+        links=gsa_results.for_user(parsed),
+        table_path=table_path,
+    )
+
+
+async def _sleep(seconds: float) -> None:
+    # Indirected so a test can run the poll loop without waiting for it.
+    import asyncio
+
+    await asyncio.sleep(seconds)