Upload folder using huggingface_hub

app/rnaseq/deseq2_analysis.R

@@ -54,6 +54,11 @@ if (nzchar(covariates_raw)) {
   if (any(!vapply(covariates, safe_name, logical(1)))) stop("covariate name has unsupported characters")
 }
 
+progress_path <- Sys.getenv("BIONEXUS_EXPRESSION_PROGRESS", "")
+progress <- function(step, status) {
+  if (nzchar(progress_path)) cat(toJSON(list(step = as.integer(step), status = status), auto_unbox = TRUE), "\n", file = progress_path, append = TRUE, sep = "")
+}
+progress(1, "running")
 txi <- NULL
 if (input_kind == "salmon") {
   suppressPackageStartupMessages(library(tximport))
@@ -85,6 +90,8 @@ if (input_kind == "salmon") {
 }
 if (anyDuplicated(colnames(counts)) > 0) stop("count matrix contains duplicate sample names")
 
+progress(1, "complete")
+progress(2, "running")
 meta <- read.delim(metadata_path, check.names = FALSE, stringsAsFactors = FALSE)
 if (!("sample" %in% names(meta))) stop("metadata must contain a 'sample' column")
 if (!(condition_col %in% names(meta))) stop(paste0("metadata is missing condition column '", condition_col, "'"))
@@ -241,17 +248,23 @@ design_audit <- list(
 if (length(design_warnings) == 0) design_audit$status <- "PASS"
 write(toJSON(design_audit, auto_unbox = TRUE, pretty = TRUE, null = "null", digits = 10), file.path(outdir, "design_audit.json"))
 
+progress(2, "complete")
+progress(3, "running")
 dds <- if (is.null(txi)) {
   DESeqDataSetFromMatrix(countData = counts, colData = meta, design = design_formula)
 } else {
   DESeqDataSetFromTximport(txi, colData = meta, design = design_formula)
 }
+progress(3, "complete")
+progress(4, "running")
 genes_input <- nrow(dds)
 keep <- rowSums(counts(dds) >= min_count) >= min_samples
 dds <- dds[keep, ]
 genes_kept <- nrow(dds)
 if (genes_kept < 2) stop("pre-filtering retained fewer than two genes")
 
+progress(4, "complete")
+progress(5, "running")
 dds <- estimateSizeFactors(dds)
 normalized <- counts(dds, normalized = TRUE)
 size_factors <- sizeFactors(dds)
@@ -275,6 +288,8 @@ size_factor_df <- data.frame(sample = names(size_factors), size_factor = as.nume
 write.table(size_factor_df, file.path(outdir, "size_factors.tsv"), sep = "\t", quote = FALSE, row.names = FALSE)
 write.table(data.frame(gene = rownames(normalized), normalized, check.names = FALSE), file.path(outdir, "normalized_counts.tsv"), sep = "\t", quote = FALSE, row.names = FALSE)
 
+progress(5, "complete")
+progress(6, "running")
 # vst() is fast for ordinary whole-transcriptome matrices but requires enough rows
 # for its trend subsampling. varianceStabilizingTransformation() is the exact
 # DESeq2 transform and is robust for small deterministic validation subsets.
@@ -325,6 +340,8 @@ if (stage == "qc") {
   quit(save = "no", status = 0)
 }
 
+progress(6, "complete")
+progress(7, "running")
 dds <- DESeq(dds)
 res <- results(dds, contrast = c(condition_col, test_level, reference_level), alpha = alpha)
 res_df <- as.data.frame(res)
@@ -358,6 +375,8 @@ write.table(res_df, file.path(outdir, "deseq2_all_results.tsv"), sep = "\t", quo
 write.table(deg_df, file.path(outdir, "deseq2_significant.tsv"), sep = "\t", quote = FALSE, row.names = FALSE, na = "")
 
 
+progress(7, "complete")
+progress(8, "running")
 pca$.condition <- pca[[condition_col]]
 pca_plot <- ggplot(pca, aes(x = PC1, y = PC2, label = name, shape = .condition)) +
   geom_point(size = 3.2) +
@@ -399,6 +418,8 @@ volcano_plot <- ggplot(volcano_df, aes(x = log2FoldChange, y = minus_log10_padj,
   theme_minimal(base_size = 11) + theme(legend.position = "bottom")
 plot_svg_pdf_png("volcano", 8, 6, function() print(volcano_plot))
 
+progress(8, "complete")
+progress(9, "running")
 heatmap_written <- FALSE
 heatmap_basis <- "none"
 heatmap_gene_count <- 0L
@@ -445,6 +466,8 @@ if (length(selected_genes) >= 2) {
   heatmap_written <- TRUE
 }
 
+progress(9, if (heatmap_written) "complete" else "unavailable")
+progress(10, "running")
 script_arg <- grep("^--file=", commandArgs(), value = TRUE)[[1]]
 source(file.path(dirname(sub("^--file=", "", script_arg)), "go_enrichment.R"))
 enrichment <- tryCatch(
@@ -458,6 +481,7 @@ enrichment <- tryCatch(
 )
 
 saveRDS(dds, file.path(outdir, "dds_fitted.rds"))
+progress(10, if (enrichment$status == "SUCCEEDED") "complete" else "unavailable")
 summary <- list(
   input_kind = input_kind,
   normalization_method = if (is.null(txi)) "median_of_ratios" else "tximport_length_offsets",

app/rnaseq/durable.py

@@ -131,9 +131,31 @@ def status(job_id: str, user: str) -> dict:
             result["checkpoint_sha256"] = digest
     return {"job_id": job_id, "state": row["state"], "phase": row["phase"],
             "attempts": row["attempts"], "error": row["error"], "result": result,
+            "progress": read_progress(root() / job_id / f"progress_{row['attempts']}.jsonl"),
             **_monitor(row)}
 
 
+def read_progress(path: Path) -> list[dict]:
+    """Only complete, bounded engine events from the current attempt are public."""
+    if not path.is_file():
+        return []
+    events = []
+    with path.open("rb") as stream:
+        body = stream.read(65536)
+    for line in body.splitlines(keepends=True):
+        if not line.endswith(b"\n"):
+            continue
+        try:
+            event = json.loads(line)
+            if (isinstance(event, dict) and type(event.get("step")) is int
+                    and 1 <= event["step"] <= 10
+                    and event.get("status") in {"running", "complete", "unavailable", "failed"}):
+                events.append({"step": event["step"], "status": event["status"]})
+        except (ValueError, UnicodeDecodeError):
+            continue
+    return events[-64:]
+
+
 def _monitor(row) -> dict:
     age = max(0.0, time.time() - row["heartbeat"]) if row["heartbeat"] else None
     return {"created_at": row["created"], "updated_at": row["updated"],
@@ -234,7 +256,8 @@ def run_job(job: dict) -> None:
             params=ExpressionParameters(**payload["params"]), source_label="reviewed-recovery",
             source_metadata={**payload["source"], "input_checksums": payload["checksums"], "qc_review": payload.get("qc_review")},
             stage=job["phase"], checkpoint_path=checkpoint, input_kind=payload["input_kind"],
-            tx2gene_path=folder / "tx2gene.tsv" if payload["input_kind"] == "salmon" else None)
+            tx2gene_path=folder / "tx2gene.tsv" if payload["input_kind"] == "salmon" else None,
+            progress_path=folder / f"progress_{job['attempts'] + 1}.jsonl")
         if job["phase"] == "qc":
             result["checkpoint_sha256"] = sha256_file(checkpoint)
             result["checkpoint_file"] = checkpoint.name

app/rnaseq/expression.py

@@ -236,7 +236,7 @@ def validate_artifacts(outdir: Path, summary: dict, stage: str) -> None:
         raise RnaSeqExpressionError("Required output artifacts missing: " + ", ".join(missing))
 
 
-def _run_r(counts_path: Path, metadata_path: Path, outdir: Path, params: ExpressionParameters, timeout_seconds: int = 600, *, stage: str = "full", checkpoint_path: Path | None = None, input_kind: str = "raw_counts", tx2gene_path: Path | None = None) -> dict[str, Any]:
+def _run_r(counts_path: Path, metadata_path: Path, outdir: Path, params: ExpressionParameters, timeout_seconds: int = 600, *, stage: str = "full", checkpoint_path: Path | None = None, input_kind: str = "raw_counts", tx2gene_path: Path | None = None, progress_path: Path | None = None) -> dict[str, Any]:
     rscript = shutil.which(os.environ.get("BIONEXUS_RSCRIPT", "Rscript"))
     if not rscript:
         raise RnaSeqExpressionError("Rscript is not installed in the backend runtime; DESeq2 execution is unavailable.")
@@ -261,7 +261,8 @@ def _run_r(counts_path: Path, metadata_path: Path, outdir: Path, params: Express
     ]
     try:
         environment = dict(os.environ, BIONEXUS_EXPRESSION_STAGE=stage, BIONEXUS_EXPRESSION_INPUT=input_kind,
-                           BIONEXUS_EXPRESSION_CHECKPOINT=str(checkpoint_path or ""), BIONEXUS_TX2GENE=str(tx2gene_path or ""))
+                           BIONEXUS_EXPRESSION_CHECKPOINT=str(checkpoint_path or ""), BIONEXUS_TX2GENE=str(tx2gene_path or ""),
+                           BIONEXUS_EXPRESSION_PROGRESS=str(progress_path or ""))
         completed = subprocess.run(argv, env=environment, capture_output=True, text=True, timeout=timeout_seconds, check=False)
     except subprocess.TimeoutExpired as exc:
         raise RnaSeqExpressionError(f"DESeq2 execution exceeded {timeout_seconds} seconds.") from exc
@@ -294,6 +295,7 @@ def execute_expression_analysis(
     checkpoint_path: Path | None = None,
     input_kind: str = "raw_counts",
     tx2gene_path: Path | None = None,
+    progress_path: Path | None = None,
 ) -> dict[str, Any]:
     params.validate()
     if counts_path.stat().st_size > MAX_COUNTS_BYTES:
@@ -314,7 +316,8 @@ def execute_expression_analysis(
         outdir = Path(tmp) / "results"
         outdir.mkdir(parents=True, exist_ok=True)
         summary = _run_r(counts_path, metadata_path, outdir, params, stage=stage, checkpoint_path=checkpoint_path,
-                         input_kind=input_kind, tx2gene_path=tx2gene_path)
+                         input_kind=input_kind, tx2gene_path=tx2gene_path,
+                         **({"progress_path": progress_path} if progress_path else {}))
         validate_artifacts(outdir, summary, stage)
 
         provenance = {

app/routers/geo_search.py

@@ -59,6 +59,7 @@ class AutomaticSelection(BaseModel):
     group_key: str | None = None
     source_sha256: str | None = None
     source_record_sha256: str | None = None
+    require_durable: bool = False
 
 
 async def _prepare_automatic(selection: AutomaticSelection):
@@ -96,6 +97,10 @@ async def prepare_geo_analysis(selection: AutomaticSelection):
 @router.post('/run')
 async def run_geo_analysis(selection: AutomaticSelection, user_id: str = Depends(require_user_id)):
     try:
+        if selection.require_durable:
+            queue_status = await run_in_threadpool(durable.readiness)
+            if not queue_status['available']:
+                raise RnaSeqExpressionError('Live queued analysis unavailable: ' + ', '.join(queue_status['missing']))
         prepared, matrix = await _prepare_automatic(selection)
         if prepared['status'] != 'ready':
             raise GeoCountsError('Load the study and choose the indicated comparison before running.')

app/routers/jobs.py

@@ -72,7 +72,7 @@ async def get_job(job_id: str, user_id: str = Depends(require_user_id)):
 
     # Hydrate from Storage if result was offloaded. Ownership has already been
     # established above; an unowned job never reaches artifact retrieval.
-    if job.get("storage_url") and not job.get("context_json", {}).get("blast"):
+    if job.get("storage_url") and job.get("status") in {"complete", "failed"}:
         from app.services.artifact_storage import download_json
         results = download_json(job["storage_url"])
         if results:

app/routers/pipeline_v2.py

@@ -82,6 +82,25 @@ def _is_real_uuid(job_id: str) -> bool:
         return False
 
 
+def _persist_blast_preview(job_id: str, context: dict) -> None:
+    """Publish completed search evidence without declaring downstream work done."""
+    if not _is_real_uuid(job_id):
+        return
+    try:
+        client = get_supabase()
+        rows = client.table("jobs").select("user_id,context_json").eq("id", job_id).execute().data or []
+        if not rows:
+            return
+        stored = rows[0].get("context_json") or {}
+        if isinstance(stored, str):
+            import json
+            stored = json.loads(stored)
+        preview = {**(stored if isinstance(stored, dict) else {}), **context}
+        client.table("jobs").update({"context_json": preview}).eq("id", job_id).eq("user_id", rows[0]["user_id"]).execute()
+    except Exception:
+        logger.warning("Could not persist early BLAST results for %s", job_id)
+
+
 def _persist_v2_final(job_id: str, status: str, context: dict, error: str | None = None):
     """Update the persisted wizard job row with the final status + full context."""
     if not _is_real_uuid(job_id):
@@ -515,6 +534,8 @@ async def _execute(job_id: str, sequence: str, steps: list[str], status_callback
                 _failed_step = "blast"
                 _failed_error = result.get("error", "BLAST completed without usable hits")
         context["blast"] = result
+        if not provider_failed:
+            await asyncio.to_thread(_persist_blast_preview, job_id, context)
 
     if denovo_mode:
         context["query"]["confidence"] = "de_novo"