Fix Filters.apply() task filter dropping anat rows

The --task filter applied pl.col("task") == value to all rows, but
anat/dwi files have task=null, so they were silently dropped.  This
caused rbc functional --task <label> to fail with FileNotFoundError
on desc-brain_T1w for any dataset.

Fix: use pl.col("task").is_null() | (pl.col("task") == value) so
rows without a task entity (anat, dwi, fmap) pass through.

Affects functional and all orchestration; metrics/qc pre-filter to
datatype=func so they were never hit.

Restore --task in integration fixtures now that the fix is in place.

Author: nx10

src/rbc/orchestration/__init__.py

@@ -63,7 +63,7 @@ def apply(self, df: pl.DataFrame, *base_exprs: pl.Expr) -> pl.DataFrame:
         if len(self.session_label) > 0:
             exprs.append(pl.col("ses").is_in(self.session_label))
         if self.task is not None:
-            exprs.append(pl.col("task") == self.task)
+            exprs.append(pl.col("task").is_null() | (pl.col("task") == self.task))
         if not exprs:
             return df
         return df.filter(pl.all_horizontal(exprs))

tests/integration/longitudinal/conftest.py

@@ -25,6 +25,7 @@
 )
 
 _SUB = "01"
+_TASK = "fingerfootlips"
 
 
 def _rbc_exe() -> str:
@@ -161,9 +162,6 @@ def ds000114_func_derivatives(
     Only ses-test is processed (one session is sufficient to exercise
     the longitudinal functional chain).
     """
-    # Note: do NOT pass --task here.  The Filters.apply() task filter
-    # applies to ALL rows including anat, and anat rows have task=null,
-    # so --task would drop all anat rows and break resolve_functional.
     _run_rbc(
         [
             "functional",
@@ -177,6 +175,8 @@ def ds000114_func_derivatives(
             _SUB,
             "--session-label",
             "test",
+            "--task",
+            _TASK,
         ],
     )
     return ds000114_anat_derivatives
@@ -209,6 +209,8 @@ def longitudinal_func_output(
             _SUB,
             "--session-label",
             "test",
+            "--task",
+            _TASK,
         ],
     )
     return ds000114_func_derivatives

tests/unit/orchestration/test_filters.py

@@ -70,10 +70,13 @@ def test_session_filter(self, bids_df: pl.DataFrame) -> None:
         result = Filters(session_label=["baseline"]).apply(bids_df)
         assert set(result["ses"].unique().to_list()) == {"baseline"}
 
-    def test_task_filter(self, bids_df: pl.DataFrame) -> None:
-        """Task filter keeps only matching tasks (nulls excluded)."""
+    def test_task_filter_preserves_anat(self, bids_df: pl.DataFrame) -> None:
+        """Task filter keeps matching tasks AND rows with null task (anat)."""
         result = Filters(task="rest").apply(bids_df)
-        assert all(t == "rest" for t in result["task"].to_list())
+        tasks = result["task"].to_list()
+        assert all(t in ("rest", None) for t in tasks)
+        # Anat rows (task=null) must survive
+        assert result.filter(pl.col("datatype") == "anat").height > 0
 
     def test_combined_filters(self, bids_df: pl.DataFrame) -> None:
         """Participant + session + task filters compose correctly."""
@@ -82,7 +85,8 @@ def test_combined_filters(self, bids_df: pl.DataFrame) -> None:
             session_label=["baseline"],
             task="rest",
         ).apply(bids_df)
-        assert len(result) == 3  # raw bold + MNI preproc + T1w preproc
+        # anat T1w (task=null) + raw bold + MNI preproc + T1w preproc = 4
+        assert len(result) == 4
         assert all(s == "01" for s in result["sub"].to_list())
         assert all(s == "baseline" for s in result["ses"].to_list())