Fail the run when an R statistics step fails or times out; draw PAC mosaics

CHANGELOG.md

@@ -1,5 +1,29 @@
 # Changelog
 
+## Unreleased
+
+### Behaviour changes (read these before re-running a study)
+
+- **A failed or timed-out R statistics step now fails the run** (exit 1). A run over a
+  whole paradigm still finishes its other modules first, then lists the failures. It
+  used to log the failure and exit 0: on the FORGE treatment re-run, roi_directed (all
+  three source arms) and roi_cross_freq's AAC/PPC tier were killed by a one-hour limit,
+  and region tables from the previous code version went on looking current.
+- **R statistics steps have no default time limit.** They were hard-coded to 3600 s
+  (roi_psd, roi_aperiodic, roi_connectivity, roi_directed, roi_cross_freq,
+  vertex_cluster) or 600 s (electrode and evoked modules). Set `r_timeout_sec` in a
+  module's config block to impose one. The three vertex modules with their own limit
+  lose it too; unlike the rest they still only log a failed step, which is left for
+  the vertex split.
+
+### Fixed
+
+- **roi_cross_freq draws its PAC mosaics.** The mosaic call named legacy columns
+  (`hedges_g`, `region`, `contrast`, `freq_pair`) that the native hypothesis table
+  does not have, so none was ever drawn. It now reads `effect_size` / `spatial` /
+  `hypothesis` / `band`, and the figures step draws them; `summary()` only did so
+  when figures were requested in the same run.
+
 ## v0.7.0 — 2026-09-10 (audit remediation, per-atlas resolution, roi_signature)
 
 A repo audit (2026-09-04) compared the README/CLAUDE.md against the code and

README.md

@@ -395,6 +395,7 @@ paradigms:
 | `bands` | all spectral/connectivity | frequency bands analysed |
 | `pipeline.atlas`, `atlas_files`, `atlas_dir` | atlas I/O, R region tier, mosaics | which parcellation the ROI data use: resolved by name through source-localization's `registry.yaml` to that atlas's own labels / mapping / categories / anatomy; `atlas_files` names the files of an unregistered atlas. The 10× voxel convention is read from each NIfTI header, never inferred from its filename |
 | `roi_categories` | region tier (Python + R), mosaics | category → ROI map. The study's map (or a profile's narrowing) always wins over the atlas default, on the Python and R sides alike |
+| `<analysis>.r_timeout_sec` | R-backed analyses | wall-clock limit for that module's R statistics step; unset = no limit. A step that fails or times out fails the run (exit 1); a paradigm-wide run finishes its other modules first |
 | `epoch_sampling` | spectral/connectivity, all levels | random-epoch resampling (`n_bootstrap: 0` = full timeseries). Precedence: global → `vertex.epoch_sampling` → per-analysis block |
 | `jobs` | `run --jobs` default | worker count when `--jobs` is not given |
 | `<profile>.{include_analyses, include_hypotheses, bands, rois}` | `run --profile` | a narrowed study written to its own tree (see below) |

src/source_analytics/analyses/base.py

@@ -12,6 +12,14 @@
 
 logger = logging.getLogger(__name__)
 
+
+class RStepFailed(RuntimeError):
+    """An R statistics step failed or timed out, so its tables were not (re)written.
+
+    Raised rather than logged: a failed or timed-out R step used to leave the run
+    exiting 0, so a batch reported success over tables that were never rewritten.
+    """
+
 VALID_STEPS = {"setup", "process", "aggregate", "statistics", "figures", "summary"}
 DEFAULT_RUN_STEPS = VALID_STEPS - {"figures"}
 
@@ -509,6 +517,24 @@ def _call_r_figures_only(self, r_script_name: str, data_csv: str) -> bool:
             logger.error("R figures-only timed out after 300s")
             return False
 
+    @property
+    def _r_timeout(self) -> float | None:
+        """Wall-clock limit for this module's R statistics step, in seconds, or None.
+
+        Set per module with ``r_timeout_sec`` in its config block. There is no
+        default: these steps are long -- roi_directed's directed-edge tier runs past
+        an hour on 26 parcels -- and a limit that silently kills them is worse than
+        waiting for them.
+        """
+        value = self.config.raw.get(self.name, {}).get("r_timeout_sec")
+        return float(value) if value else None
+
+    def _r_step_failed(self, msg: str, *args) -> None:
+        """Log an R statistics-step failure and raise :class:`RStepFailed`."""
+        text = msg % args if args else msg
+        logger.error(text)
+        raise RStepFailed(f"{self.name}: {text}")
+
     def _r_roi_categories_flags(self) -> list[str]:
         """``['--roi-categories', path]`` for the resolved atlas's own category file.
 

src/source_analytics/analyses/electrode_analysis.py

@@ -337,7 +337,7 @@ def summary(self) -> None:
         logger.info("Calling R: %s", " ".join(cmd))
         try:
             result = subprocess.run(
-                cmd, capture_output=True, text=True, timeout=600,
+                cmd, capture_output=True, text=True, timeout=self._r_timeout,
             )
             if result.stdout:
                 for line in result.stdout.strip().split("\n"):
@@ -347,8 +347,8 @@ def summary(self) -> None:
                     if line.strip():
                         logger.info("[R] %s", line)
             if result.returncode != 0:
-                logger.error("R script failed with exit code %d", result.returncode)
+                self._r_step_failed("R script failed with exit code %d", result.returncode)
         except FileNotFoundError:
             logger.error("Rscript not found. Install R to enable statistics and visualization.")
         except subprocess.TimeoutExpired:
-            logger.error("R script timed out after 600 seconds")
+            self._r_step_failed("R script timed out after %s s", self._r_timeout)

src/source_analytics/analyses/electrode_aperiodic_analysis.py

@@ -280,7 +280,7 @@ def summary(self) -> None:
         logger.info("Calling R: %s", " ".join(cmd))
         try:
             result = subprocess.run(
-                cmd, capture_output=True, text=True, timeout=600,
+                cmd, capture_output=True, text=True, timeout=self._r_timeout,
             )
             if result.stdout:
                 for line in result.stdout.strip().split("\n"):
@@ -290,12 +290,10 @@ def summary(self) -> None:
                     if line.strip():
                         logger.info("[R] %s", line)
             if result.returncode != 0:
-                logger.error(
-                    "R script failed with exit code %d", result.returncode,
-                )
+                self._r_step_failed("R script failed with exit code %d", result.returncode)
         except FileNotFoundError:
             logger.error(
                 "Rscript not found. Install R to enable statistics.",
             )
         except subprocess.TimeoutExpired:
-            logger.error("R script timed out after 600 seconds")
+            self._r_step_failed("R script timed out after %s s", self._r_timeout)

src/source_analytics/analyses/electrode_comparison_analysis.py

@@ -675,7 +675,7 @@ def summary(self) -> None:
         logger.info("Calling R: %s", " ".join(cmd))
         try:
             result = subprocess.run(
-                cmd, capture_output=True, text=True, timeout=600,
+                cmd, capture_output=True, text=True, timeout=self._r_timeout,
             )
             if result.stdout:
                 for line in result.stdout.strip().split("\n"):
@@ -685,8 +685,8 @@ def summary(self) -> None:
                     if line.strip():
                         logger.info("[R] %s", line)
             if result.returncode != 0:
-                logger.error("R script failed with exit code %d", result.returncode)
+                self._r_step_failed("R script failed with exit code %d", result.returncode)
         except FileNotFoundError:
             logger.error("Rscript not found. Install R to enable comparison report.")
         except subprocess.TimeoutExpired:
-            logger.error("R script timed out after 600 seconds")
+            self._r_step_failed("R script timed out after %s s", self._r_timeout)

src/source_analytics/analyses/electrode_evoked_analysis.py

@@ -428,7 +428,7 @@ def summary(self) -> None:
                 cmd,
                 capture_output=True,
                 text=True,
-                timeout=600,
+                timeout=self._r_timeout,
             )
             if result.stdout:
                 for line in result.stdout.strip().split("\n"):
@@ -438,12 +438,10 @@ def summary(self) -> None:
                     if line.strip():
                         logger.info("[R] %s", line)
             if result.returncode != 0:
-                logger.error(
-                    "R script failed with exit code %d", result.returncode
-                )
+                self._r_step_failed("R script failed with exit code %d", result.returncode)
         except FileNotFoundError:
             logger.error(
                 "Rscript not found. Install R to enable statistics and visualization."
             )
         except subprocess.TimeoutExpired:
-            logger.error("R script timed out after 600 seconds")
+            self._r_step_failed("R script timed out after %s s", self._r_timeout)

src/source_analytics/analyses/roi_aperiodic_analysis.py

@@ -195,7 +195,7 @@ def summary(self) -> None:
                 cmd,
                 capture_output=True,
                 text=True,
-                timeout=3600,
+                timeout=self._r_timeout,
             )
             if result.stdout:
                 for line in result.stdout.strip().split("\n"):
@@ -205,11 +205,11 @@ def summary(self) -> None:
                     if line.strip():
                         logger.info("[R] %s", line)
             if result.returncode != 0:
-                logger.error("R script failed with exit code %d", result.returncode)
+                self._r_step_failed("R script failed with exit code %d", result.returncode)
         except FileNotFoundError:
             logger.error("Rscript not found. Install R to enable statistics and visualization.")
         except subprocess.TimeoutExpired:
-            logger.error("R script timed out after 3600 seconds")
+            self._r_step_failed("R script timed out after %s s", self._r_timeout)
 
         # Render brain mosaics from posthoc effect sizes
         if self._generate_figures:
@@ -244,5 +244,5 @@ def _render_brain_mosaics(self) -> None:
             correction_label="FDR",
             facet_cols=["hypothesis", "dv"],
             colorbar_label="Hedges' g",
-            auto_slices=True,atlas=self._atlas_dir
+            auto_slices=True, atlas=self._atlas_dir
         )

src/source_analytics/analyses/roi_connectivity_analysis.py

@@ -343,7 +343,7 @@ def summary(self) -> None:
                 cmd,
                 capture_output=True,
                 text=True,
-                timeout=3600,
+                timeout=self._r_timeout,
             )
             if result.stdout:
                 for line in result.stdout.strip().split("\n"):
@@ -353,10 +353,10 @@ def summary(self) -> None:
                     if line.strip():
                         logger.info("[R] %s", line)
             if result.returncode != 0:
-                logger.error("R script failed with exit code %d", result.returncode)
+                self._r_step_failed("R script failed with exit code %d", result.returncode)
         except FileNotFoundError:
             logger.error(
                 "Rscript not found. Install R to enable statistics and visualization."
             )
         except subprocess.TimeoutExpired:
-            logger.error("R script timed out after 3600 seconds")
+            self._r_step_failed("R script timed out after %s s", self._r_timeout)

src/source_analytics/analyses/roi_cross_freq_analysis.py

@@ -251,34 +251,46 @@ def figures(self) -> None:
         for m in self._edge_metrics_on_disk():
             self._call_r_figures_only("roi_cross_freq_edges_analysis.R", f"{m}_edges.csv")
             break  # one call handles every edge CSV present
+        if "pac" in self._metrics:
+            # From the tables on disk, as roi_psd does: summary() drew these only
+            # when figures were requested in the same run, so a separate figures
+            # pass (the usual way to redraw) never did.
+            self._render_brain_mosaics()
 
     # -------------------------------------------------------------- summary
     def summary(self) -> None:
         """Statistics + figures + report via R (PAC first, then AAC/PPC)."""
-        if "pac" in self._metrics:
-            self._run_pac_r()
+        failed = []
+        if "pac" in self._metrics and not self._run_pac_r():
+            failed.append("PAC")
         edge_metrics = self._edge_metrics_on_disk()
-        if edge_metrics:
-            self._run_edges_r(edge_metrics)
-
-    def _run_edges_r(self, metrics: list[str]) -> None:
+        if edge_metrics and not self._run_edges_r(edge_metrics):
+            failed.append("AAC/PPC")
+        if failed:
+            # Both tiers were attempted first, so one failing does not cost the
+            # other its tables.
+            self._r_step_failed("R step(s) failed: %s", ", ".join(failed))
+
+    def _run_edges_r(self, metrics: list[str]) -> bool:
         """AAC/PPC hypotheses (global / cells / region pairs) via R."""
-        self._run_r_script(
+        return self._run_r_script(
             "roi_cross_freq_edges_analysis.R",
             extra_args=["--metric", ",".join(metrics)],
             label="AAC/PPC",
         )
 
-    def _run_pac_r(self) -> None:
+    def _run_pac_r(self) -> bool:
         data_dir = self.output_dir / "data"
         if not (data_dir / "pac_values.csv").exists():
             logger.error("pac_values.csv not found -- skipping PAC R analysis")
-            return
-        if self._run_r_script("roi_pac_analysis.R", label="PAC") and self._generate_figures:
+            return True   # nothing to test is a skip, not a failure
+        ok = self._run_r_script("roi_pac_analysis.R", label="PAC")
+        if ok and self._generate_figures:
             self._render_brain_mosaics()
+        return ok
 
     def _run_r_script(self, script_name: str, *, extra_args: list[str] | None = None,
-                      label: str = "R", timeout: int = 3600) -> bool:
+                      label: str = "R", timeout: float | None = None) -> bool:
         """Run one of this module's R scripts with the standard argument set.
 
         Returns True when the script exited 0.
@@ -320,9 +332,10 @@ def _run_r_script(self, script_name: str, *, extra_args: list[str] | None = None
         if extra_args:
             cmd.extend(extra_args)
 
+        limit = timeout if timeout is not None else self._r_timeout
         logger.info("Calling R (%s): %s", label, " ".join(cmd))
         try:
-            result = subprocess.run(cmd, capture_output=True, text=True, timeout=timeout)
+            result = subprocess.run(cmd, capture_output=True, text=True, timeout=limit)
             for stream in (result.stdout, result.stderr):
                 if stream:
                     for line in stream.strip().split("\n"):
@@ -335,7 +348,7 @@ def _run_r_script(self, script_name: str, *, extra_args: list[str] | None = None
         except FileNotFoundError:
             logger.error("Rscript not found. Install R to enable %s statistics.", label)
         except subprocess.TimeoutExpired:
-            logger.error("%s R script timed out after %d seconds", label, timeout)
+            logger.error("%s R script timed out after %s s", label, limit)
         return False
 
     def _render_brain_mosaics(self) -> None:
@@ -353,7 +366,11 @@ def _render_brain_mosaics(self) -> None:
         render_posthoc_mosaics(
             posthoc_csv, roi_cats, self.fig_dir,
             analysis_name="roi_cross_freq",
-            effect_col="hedges_g", roi_col="region",
-            facet_cols=["contrast", "freq_pair"],
-            colorbar_label="Hedges' g",atlas=self._atlas_dir
+            # The native hypothesis schema: `spatial` holds the region and `band`
+            # the frequency pair. The legacy names (hedges_g / region / contrast /
+            # freq_pair) matched no column, so no PAC mosaic was ever drawn.
+            effect_col="effect_size", roi_col="spatial",
+            p_col="p_value", q_col="q_value", correction_label="FDR",
+            facet_cols=["hypothesis", "band"],
+            colorbar_label="Hedges' g", atlas=self._atlas_dir,
         )

src/source_analytics/analyses/roi_directed_analysis.py

@@ -234,7 +234,7 @@ def summary(self) -> None:
         if wanted_hyp:
             cmd.extend(["--hypothesis", ",".join(sorted(wanted_hyp))])
 
-        r_timeout = 3600
+        r_timeout = self._r_timeout
         logger.info("Calling R: %s", " ".join(cmd))
         try:
             result = subprocess.run(
@@ -251,10 +251,10 @@ def summary(self) -> None:
                     if line.strip():
                         logger.info("[R] %s", line)
             if result.returncode != 0:
-                logger.error("R script failed with exit code %d", result.returncode)
+                self._r_step_failed("R script failed with exit code %d", result.returncode)
         except FileNotFoundError:
             logger.error(
                 "Rscript not found. Install R to enable statistics and visualization."
             )
         except subprocess.TimeoutExpired:
-            logger.error("R script timed out after %d seconds", r_timeout)
+            self._r_step_failed("R script timed out after %s s", self._r_timeout)

src/source_analytics/analyses/roi_evoked_analysis.py

@@ -321,7 +321,7 @@ def summary(self) -> None:
                 cmd,
                 capture_output=True,
                 text=True,
-                timeout=600,
+                timeout=self._r_timeout,
             )
             if result.stdout:
                 for line in result.stdout.strip().split("\n"):
@@ -331,8 +331,8 @@ def summary(self) -> None:
                     if line.strip():
                         logger.info("[R] %s", line)
             if result.returncode != 0:
-                logger.error("R script failed with exit code %d", result.returncode)
+                self._r_step_failed("R script failed with exit code %d", result.returncode)
         except FileNotFoundError:
             logger.error("Rscript not found. Install R to enable statistics and visualization.")
         except subprocess.TimeoutExpired:
-            logger.error("R script timed out after 600 seconds")
+            self._r_step_failed("R script timed out after %s s", self._r_timeout)

src/source_analytics/analyses/roi_psd_analysis.py

@@ -231,7 +231,7 @@ def summary(self) -> None:
                 cmd,
                 capture_output=True,
                 text=True,
-                timeout=3600,
+                timeout=self._r_timeout,
             )
             if result.stdout:
                 for line in result.stdout.strip().split("\n"):
@@ -241,11 +241,11 @@ def summary(self) -> None:
                     if line.strip():
                         logger.info("[R] %s", line)
             if result.returncode != 0:
-                logger.error("R script failed with exit code %d", result.returncode)
+                self._r_step_failed("R script failed with exit code %d", result.returncode)
         except FileNotFoundError:
             logger.error("Rscript not found. Install R to enable statistics and visualization.")
         except subprocess.TimeoutExpired:
-            logger.error("R script timed out after 3600 seconds")
+            self._r_step_failed("R script timed out after %s s", self._r_timeout)
 
         # Render brain mosaics from posthoc effect sizes
         if self._generate_figures:
@@ -280,5 +280,5 @@ def _render_brain_mosaics(self) -> None:
             correction_label="FDR",
             facet_cols=["hypothesis", "band", "dv"],
             colorbar_label="Hedges' g",
-            auto_slices=True,atlas=self._atlas_dir
+            auto_slices=True, atlas=self._atlas_dir
         )