diff --git a/AnaProd/MergeAnaTuples.py b/AnaProd/MergeAnaTuples.py
index 793d23c4..68bb5bcf 100644
--- a/AnaProd/MergeAnaTuples.py
+++ b/AnaProd/MergeAnaTuples.py
@@ -209,6 +209,12 @@ def mergeAnaTuples(
     else:
         tree_list = [(central, central, "Events")]
 
+    if not tree_list:
+        raise RuntimeError(
+            f"No trees to merge for dataset {dataset_name} "
+            f"(empty or missing report tree lists)."
+        )
+
     if len(root_outputs) > 1 and len(tree_list) > 1:
         raise NotImplementedError(
             "Cannot write multiple output files when there are multiple uncertainties."
diff --git a/AnaProd/tasks.py b/AnaProd/tasks.py
index c70982a2..9bd5a2bc 100644
--- a/AnaProd/tasks.py
+++ b/AnaProd/tasks.py
@@ -15,7 +15,12 @@
     check_root_file_integrity,
     get_tree_entries,
 )
-from FLAF.run_tools.law_customizations import Task, HTCondorWorkflow, copy_param
+from FLAF.run_tools.law_customizations import (
+    Task,
+    HTCondorWorkflow,
+    CrabWorkflow,
+    copy_param,
+)
 from FLAF.Common.Utilities import getCustomisationSplit, ServiceThread
 from .AnaTupleFileList import CreateMergePlan
 from .MergeAnaTuples import mergeAnaTuples
@@ -111,9 +116,10 @@ def WF_complete(ref_task):
         return InputFileTask.WF_complete_
 
 
-class AnaTupleFileTask(Task, HTCondorWorkflow, law.LocalWorkflow):
+class AnaTupleFileTask(Task, HTCondorWorkflow, CrabWorkflow, law.LocalWorkflow):
     max_runtime = copy_param(HTCondorWorkflow.max_runtime, 40.0)
-    n_cpus = copy_param(HTCondorWorkflow.n_cpus, 2)
+    # tautau CMSSW AnaTuple used ~7.5 GB RSS on CRAB; 2 cores cap at 5000 MB.
+    n_cpus = copy_param(HTCondorWorkflow.n_cpus, 4)
 
     @property
     def bundle_flavours(self):
@@ -325,7 +331,9 @@ def run(self):
                 shutil.rmtree(job_home)
 
 
-class AnaTupleFileListBuilderTask(Task, HTCondorWorkflow, law.LocalWorkflow):
+class AnaTupleFileListBuilderTask(
+    Task, HTCondorWorkflow, CrabWorkflow, law.LocalWorkflow
+):
     max_runtime = copy_param(HTCondorWorkflow.max_runtime, 24.0)
     n_cpus = copy_param(HTCondorWorkflow.n_cpus, 1)
     bundle_flavours = ["core", "inputFileList"]
@@ -536,7 +544,7 @@ def run(self):
             shutil.copy(input_local.abspath, self.output().abspath)
 
 
-class AnaTupleMergeTask(Task, HTCondorWorkflow, law.LocalWorkflow):
+class AnaTupleMergeTask(Task, HTCondorWorkflow, CrabWorkflow, law.LocalWorkflow):
     max_runtime = copy_param(HTCondorWorkflow.max_runtime, 48.0)
     n_cpus = copy_param(HTCondorWorkflow.n_cpus, 2)
     delete_inputs_after_merge = luigi.BoolParameter(default=False)
diff --git a/Analysis/AnalysisCacheProducer.py b/Analysis/AnalysisCacheProducer.py
index f035038f..ac14c4f0 100644
--- a/Analysis/AnalysisCacheProducer.py
+++ b/Analysis/AnalysisCacheProducer.py
@@ -269,7 +269,12 @@ def createAnalysisCache(
     parser.add_argument("--workingDir", required=True, type=str)
     parser.add_argument("--histTupleDef", type=str)
     parser.add_argument("--LAWrunVersion", required=True, type=str)
-    parser.add_argument("--user-custom", type=str, default=None)
+    parser.add_argument(
+        "--user-custom",
+        type=str,
+        default=None,
+        help="Optional user_custom yaml (staged on remote workers; resolved by Setup).",
+    )
     args = parser.parse_args()
 
     startTime = time.time()
diff --git a/Analysis/tasks.py b/Analysis/tasks.py
index cfd54e18..ce074738 100644
--- a/Analysis/tasks.py
+++ b/Analysis/tasks.py
@@ -10,6 +10,7 @@
 from FLAF.run_tools.law_customizations import (
     Task,
     HTCondorWorkflow,
+    CrabWorkflow,
     copy_param,
 )
 from FLAF.AnaProd.tasks import (
@@ -61,7 +62,7 @@ def _anaTuple_outputs(task):
     return cache
 
 
-class HistTupleProducerTask(Task, HTCondorWorkflow, law.LocalWorkflow):
+class HistTupleProducerTask(Task, HTCondorWorkflow, CrabWorkflow, law.LocalWorkflow):
     max_runtime = copy_param(HTCondorWorkflow.max_runtime, 5.0)
     n_cpus = copy_param(HTCondorWorkflow.n_cpus, 4)
     # many short per-file branches: group several per HTCondor job to bound nJobs.
@@ -201,6 +202,7 @@ def requires(self):
             "anaTuple": AnaTupleMergeTask.req(
                 self,
                 max_runtime=AnaTupleMergeTask.max_runtime._default,
+                n_cpus=AnaTupleMergeTask.n_cpus._default,
                 branch=prod_br,
                 branches=(prod_br,),
                 customisations=self.customisations,
@@ -515,7 +517,9 @@ def _split_merged_marker(split_target):
     return split_target.sibling(split_target.basename + ".merged", type="f")
 
 
-class HistFromNtupleProducerTask(Task, HTCondorWorkflow, law.LocalWorkflow):
+class HistFromNtupleProducerTask(
+    Task, HTCondorWorkflow, CrabWorkflow, law.LocalWorkflow
+):
     max_runtime = copy_param(HTCondorWorkflow.max_runtime, 10.0)
     n_cpus = copy_param(HTCondorWorkflow.n_cpus, 2)
     variables = luigi.Parameter(default="")
@@ -688,7 +692,7 @@ def run(self):
         HistFromNtupleProducer = os.path.join(
             self._flaf_root(), "Analysis", "HistProducerFromNTuple.py"
         )
-        nMT = self.n_cpus * 2 if self.effective_workflow == "htcondor" else 8
+        nMT = self.n_cpus * 2 if self.effective_workflow in ("htcondor", "crab") else 8
 
         # Determine which variables still need to be produced for this (dataset, chunk).
         outputs = self.output()
@@ -766,7 +770,7 @@ def _localize(inp):
             shutil.rmtree(job_home)
 
 
-class HistMergerTask(Task, HTCondorWorkflow, law.LocalWorkflow):
+class HistMergerTask(Task, HTCondorWorkflow, CrabWorkflow, law.LocalWorkflow):
     max_runtime = copy_param(HTCondorWorkflow.max_runtime, 5.0)
     n_cpus = copy_param(HTCondorWorkflow.n_cpus, 2)
     variables = luigi.Parameter(default="")
@@ -986,7 +990,7 @@ def _localize(var_file):
                     )
 
 
-class AnalysisCacheTask(Task, HTCondorWorkflow, law.LocalWorkflow):
+class AnalysisCacheTask(Task, HTCondorWorkflow, CrabWorkflow, law.LocalWorkflow):
     max_runtime = copy_param(HTCondorWorkflow.max_runtime, 2.0)
     n_cpus = copy_param(HTCondorWorkflow.n_cpus, 1)
     producer_to_run = luigi.Parameter()
@@ -1003,10 +1007,14 @@ def bundle_flavours(self):
             flavours.append("cmssw")
         return flavours
 
-    # Need to override this from HTCondorWorkflow to have separate data pathways for different cache tasks
+    # Need to override this from HTCondorWorkflow/CrabWorkflow to have separate data
+    # pathways for different cache tasks
     def htcondor_output_directory(self):
         return law.LocalDirectoryTarget(self.local_path(self.producer_to_run))
 
+    def crab_output_directory(self):
+        return law.LocalDirectoryTarget(self.local_path(self.producer_to_run))
+
     def __init__(self, *args, **kwargs):
         ana_v = kwargs.get("ana_version") or kwargs.get("anaCache_version")
         if ana_v:
@@ -1240,6 +1248,10 @@ def run(self):
                         analysisCacheProducer_cmd.extend(
                             ["--cacheFiles", local_anacaches_str]
                         )
+                    if self.user_custom:
+                        analysisCacheProducer_cmd.extend(
+                            ["--user-custom", self.user_custom]
+                        )
                     # Check if cmssw env is required
                     prod_env = (
                         self.cmssw_env
@@ -1269,7 +1281,7 @@ def run(self):
                 shutil.rmtree(job_home)
 
 
-class HistPlotTask(Task, HTCondorWorkflow, law.LocalWorkflow):
+class HistPlotTask(Task, HTCondorWorkflow, CrabWorkflow, law.LocalWorkflow):
     max_runtime = copy_param(HTCondorWorkflow.max_runtime, 2.0)
     n_cpus = copy_param(HTCondorWorkflow.n_cpus, 1)
     variables = luigi.Parameter(default="")
@@ -1518,7 +1530,9 @@ def bool_flag(key, default):
                 ps_call(cmd, verbose=1)
 
 
-class AnalysisCacheAggregationTask(Task, HTCondorWorkflow, law.LocalWorkflow):
+class AnalysisCacheAggregationTask(
+    Task, HTCondorWorkflow, CrabWorkflow, law.LocalWorkflow
+):
     max_runtime = copy_param(HTCondorWorkflow.max_runtime, 2.0)
     n_cpus = copy_param(HTCondorWorkflow.n_cpus, 1)
     producer_to_aggregate = luigi.Parameter()
diff --git a/Common/Setup.py b/Common/Setup.py
index 640b93ee..4b231a3f 100644
--- a/Common/Setup.py
+++ b/Common/Setup.py
@@ -10,6 +10,57 @@
 from FLAF.Common.Utilities import create_processor_instances
 
 
+def resolve_user_custom_path(user_custom):
+    """Resolve --user-custom on the submit host or a remote worker.
+
+    Absolute submit-host paths are missing on CRAB/HTCondor bundle workers.
+    Jobs stage the file as a job input; law renames it with a content-hash
+    suffix (``name_<hash>.yaml``). Search common job dirs for the basename or
+    a hashed variant. Returns an absolute path when a file is found.
+    """
+
+    def _abs_if_file(p):
+        if p and os.path.isfile(p):
+            return os.path.abspath(p)
+        return None
+
+    if not user_custom:
+        return user_custom
+    path = user_custom
+    if not os.path.isabs(path):
+        ana = os.getenv("ANALYSIS_PATH") or ""
+        path = os.path.join(ana, path) if ana else path
+    found = _abs_if_file(path)
+    if found:
+        return found
+    base = os.path.basename(user_custom)
+    stem, ext = os.path.splitext(base)
+    search_dirs = [
+        os.environ.get("LAW_JOB_INIT_DIR", ""),
+        os.environ.get("LAW_JOB_HOME", ""),
+        "/srv",
+        os.getcwd(),
+        os.getenv("ANALYSIS_PATH") or "",
+    ]
+    for d in search_dirs:
+        if not d:
+            continue
+        found = _abs_if_file(os.path.join(d, base))
+        if found:
+            return found
+        if not os.path.isdir(d):
+            continue
+        try:
+            for name in os.listdir(d):
+                if name.startswith(stem + "_") and name.endswith(ext):
+                    found = _abs_if_file(os.path.join(d, name))
+                    if found:
+                        return found
+        except OSError:
+            pass
+    return path
+
+
 def select_items(all_items, filters):
     def name_match(name, pattern):
         if pattern[0] == "^":
@@ -265,12 +316,12 @@ def __init__(
         self.period = period
         self.law_run_version = law_run_version
 
-        # Resolve a relative user_custom file against the analysis path. The path is passed
-        # through to subprocesses (e.g. HistTupleProducer.py) verbatim, where the working
-        # directory is not the analysis directory (on HTCondor it is the job scratch dir), so
-        # it must be made absolute here for every caller, not only in Task.__init__.
-        if user_custom_file is not None and not os.path.isabs(user_custom_file):
-            user_custom_file = os.path.join(ana_path, user_custom_file)
+        # Resolve user_custom for every caller (Task and standalone scripts). Relative
+        # paths are rooted under ANALYSIS_PATH; absolute submit-host paths that are
+        # missing on remote workers are remapped to staged job inputs (see
+        # resolve_user_custom_path).
+        if user_custom_file:
+            user_custom_file = resolve_user_custom_path(user_custom_file)
 
         self.config_path_order = [
             os.path.join(ana_path, "FLAF", "config"),
@@ -375,12 +426,37 @@ def __init__(
 
         self.histTuple_flavor = self.global_params["histTuple_flavor"]
         print(f"Using histTuple flavor {self.histTuple_flavor}")
-        self.histTuple_plot_vars = self.global_params["histTuple_flavors"][
-            self.histTuple_flavor
-        ]["variables"]
-        self.histTuple_fullres_vars = self.global_params["histTuple_flavors"][
-            self.histTuple_flavor
-        ]["fullResolution_variables"]
+        self.histTuple_plot_vars = list(
+            self.global_params["histTuple_flavors"][self.histTuple_flavor]["variables"]
+        )
+        self.histTuple_fullres_vars = list(
+            self.global_params["histTuple_flavors"][self.histTuple_flavor][
+                "fullResolution_variables"
+            ]
+        )
+
+        # Optional top-level `variables:` from user_custom / global.yaml.
+        # - When the active flavor already lists variables: treat as a restriction
+        #   (keep only names in the list) — used by CI user_custom flavors.
+        # - When the flavor list is empty (e.g. H_mumu default): use the list as the
+        #   active plot/full-res set so user_custom alone can drive a short CI chain
+        #   without requiring histTuple_flavor: CI.
+        user_vars = self.global_params.get("variables")
+        if user_vars:
+
+            def _var_name(v):
+                return v["name"] if isinstance(v, dict) else v
+
+            selected = {_var_name(v) for v in user_vars}
+            if self.histTuple_plot_vars:
+                self.histTuple_plot_vars = [
+                    v for v in self.histTuple_plot_vars if _var_name(v) in selected
+                ]
+                self.histTuple_fullres_vars = [
+                    v for v in self.histTuple_fullres_vars if _var_name(v) in selected
+                ]
+            else:
+                self.histTuple_plot_vars = list(user_vars)
 
         # Whether up/down-variation histograms are produced. The histTuple flavor usually
         # dictates this (uncertainties are only needed for the limit-setting shape variable),
@@ -581,14 +657,35 @@ def _create_fs_instance(self, path_or_paths):
             cache_validity = cfg.get("localPathCacheValidity", 600)
             host = cfg.get("remotePathCacheHost", None)
             port = cfg.get("remotePathCachePort", None)
+            # cms-flaf.cern.ch is behind the CERN firewall; CRAB workers at other
+            # sites cannot reach it. Use the in-process PathCache, seeded from the
+            # snapshot shipped at submit, with a longer TTL so many jobs do not
+            # re-stat the same remote paths.
+            on_crab_worker = bool(
+                os.environ.get("LAW_CRAB_JOB_NUMBER") or os.environ.get("CRAB_Id")
+            )
+            if on_crab_worker:
+                host = None
+                port = None
+                cache_validity = int(
+                    cfg.get(
+                        "crabLocalPathCacheValidity",
+                        max(cache_validity * 24, 86400),
+                    )
+                )
             verbose = cfg.get("verbose", 0)
-            return WLCGFileSystem(
+            fs = WLCGFileSystem(
                 path_or_paths,
                 local_path_cache_validity_period=cache_validity,
                 path_cache_host=host,
                 path_cache_port=port,
                 verbose=verbose,
             )
+            if on_crab_worker:
+                from FLAF.RunKit.law_gfal import apply_shipped_path_cache
+
+                apply_shipped_path_cache(fs)
+            return fs
 
     def get_fs(self, fs_name, custom_paths=None):
         fs_instance = None
@@ -640,23 +737,49 @@ def fs_rucio(self):
     @property
     def cmssw_env(self):
         if self.cmssw_env_ is None:
-            self.cmssw_env_ = get_cmsenv(cmssw_path=os.getenv("FLAF_CMSSW_BASE"))
+            flaf_cmssw = os.getenv("FLAF_CMSSW_BASE")
+            self.cmssw_env_ = get_cmsenv(cmssw_path=flaf_cmssw)
             for var in [
                 "HOME",
                 "FLAF_PATH",
+                "CORRECTIONS_PATH",
                 "ANALYSIS_PATH",
                 "ANALYSIS_DATA_PATH",
                 "X509_USER_PROXY",
                 "FLAF_CMSSW_BASE",
                 "FLAF_CMSSW_ARCH",
+                "PYTHONSAFEPATH",
+                "LAW_CRAB_JOB_NUMBER",
+                "CRAB_Id",
+                "LAW_JOB_INIT_DIR",
+                "LAW_JOB_HOME",
+                "FLAF_SHIPPED_PATH_CACHE",
             ]:
                 if var in os.environ:
                     self.cmssw_env_[var] = os.environ[var]
+            # scram runtime (inside get_cmsenv) may still emit the submit-host AFS
+            # CMSSW_BASE when ProjectRename failed on a CRAB worker. Force the
+            # relocated release path so CMSSW-dependent models / includeLibTool resolve
+            # inside the bundle, not /afs/...
+            if flaf_cmssw and os.path.isdir(flaf_cmssw):
+                self.cmssw_env_["CMSSW_BASE"] = flaf_cmssw
+                self.cmssw_env_["FLAF_CMSSW_BASE"] = flaf_cmssw
+            # Prepend overlay parents (if set) then ANALYSIS_PATH so `import FLAF`
+            # / `import Corrections` match env.sh. Without this, CMSSW python
+            # resolves the submodule copy and misses overlay-only modules.
+            py_prefix = [self.ana_path]
+            for env_key in ("FLAF_PATH", "CORRECTIONS_PATH"):
+                overlay = os.environ.get(env_key)
+                if overlay and os.path.isdir(overlay):
+                    parent = os.path.dirname(os.path.abspath(overlay))
+                    if parent and parent not in py_prefix:
+                        py_prefix.insert(0, parent)
+            py_prefix_str = ":".join(py_prefix)
             if "PYTHONPATH" not in self.cmssw_env_:
-                self.cmssw_env_["PYTHONPATH"] = self.ana_path
+                self.cmssw_env_["PYTHONPATH"] = py_prefix_str
             else:
                 self.cmssw_env_["PYTHONPATH"] = (
-                    f'{self.ana_path}:{self.cmssw_env["PYTHONPATH"]}'
+                    f"{py_prefix_str}:{self.cmssw_env_['PYTHONPATH']}"
                 )
         return self.cmssw_env_
 
diff --git a/RunKit/includeCMSSWlibs.py b/RunKit/includeCMSSWlibs.py
index 96eb754d..4b97d68e 100644
--- a/RunKit/includeCMSSWlibs.py
+++ b/RunKit/includeCMSSWlibs.py
@@ -8,7 +8,16 @@
 
 def includeLibTool(tool="", wantLib=False):
     command = ["scram", "tool", "info", tool]
-    directory = os.environ["CMSSW_BASE"]
+    # Prefer FLAF_CMSSW_BASE (always the analysis soft/CMSSW release, correctly
+    # relocated on CRAB/HTCondor bundle workers). CMSSW_BASE may still point at the
+    # submit-host AFS path if scram ProjectRename did not fully re-export the env.
+    directory = os.environ.get("FLAF_CMSSW_BASE") or os.environ["CMSSW_BASE"]
+    if not os.path.isdir(directory):
+        raise FileNotFoundError(
+            f"CMSSW release directory not found: {directory} "
+            f"(FLAF_CMSSW_BASE={os.environ.get('FLAF_CMSSW_BASE')!r}, "
+            f"CMSSW_BASE={os.environ.get('CMSSW_BASE')!r})"
+        )
     returncode, output, err = ps_call(
         command, catch_stdout=True, cwd=directory, verbose=0
     )
diff --git a/RunKit/law_gfal.py b/RunKit/law_gfal.py
index 0343a6b0..06d9c3d6 100644
--- a/RunKit/law_gfal.py
+++ b/RunKit/law_gfal.py
@@ -1,3 +1,4 @@
+import json
 import time
 import os
 import sys
@@ -94,6 +95,28 @@ def get(self, path):
     def get_many(self, paths):
         return {path: self.get(path)[0] for path in paths}
 
+    def iter_valid(self):
+        for path, entry in list(self.cache.items()):
+            if entry.is_valid():
+                yield path, entry.exists
+
+    def load_entries(self, entries):
+        """Refresh entries with this cache's validity period (snapshot has no timestamps)."""
+        negatives = []
+        positives = []
+        for item in entries:
+            path = item.get("path")
+            if not path:
+                continue
+            if item.get("exists"):
+                positives.append(path)
+            else:
+                negatives.append(path)
+        for path in negatives:
+            self.set(path, False)
+        for path in positives:
+            self.set(path, True)
+
     def invalidate(self, path):
         to_remove = []
         for p in self.cache:
@@ -185,6 +208,88 @@ def invalidate(self, path):
         self.local_cache.invalidate(path)
 
 
+SHIPPED_PATH_CACHE_BASENAME = "path_cache.json"
+SHIPPED_PATH_CACHE_ENV = "FLAF_SHIPPED_PATH_CACHE"
+
+_shipped_path_cache_entries = None
+
+
+def local_path_cache(fs):
+    """Return the in-process PathCache for a WLCG/GFAL filesystem, if any."""
+    fi = getattr(fs, "file_interface", None)
+    pc = getattr(fi, "path_cache", None)
+    if pc is None:
+        return None
+    return getattr(pc, "local_cache", pc)
+
+
+def collect_setup_path_cache_entries(setup):
+    """Union of valid path-cache entries from every FS the Setup has already created."""
+    entries = {}
+    for fs in getattr(setup, "fs_dict", {}).values():
+        pc = local_path_cache(fs)
+        if pc is None:
+            continue
+        for path, exists in pc.iter_valid():
+            entries[path] = exists
+    return [{"path": path, "exists": exists} for path, exists in entries.items()]
+
+
+def write_path_cache_file(path, entries):
+    with open(path, "w") as f:
+        json.dump({"entries": entries}, f)
+
+
+def _resolve_shipped_path_cache_file():
+    env_path = os.environ.get(SHIPPED_PATH_CACHE_ENV, "")
+    if env_path and os.path.isfile(env_path):
+        return env_path
+    stem, ext = os.path.splitext(SHIPPED_PATH_CACHE_BASENAME)
+    search_dirs = [
+        os.environ.get("LAW_JOB_INIT_DIR", ""),
+        os.environ.get("LAW_JOB_HOME", ""),
+        "/srv",
+        os.getcwd(),
+    ]
+    for d in search_dirs:
+        if not d:
+            continue
+        direct = os.path.join(d, SHIPPED_PATH_CACHE_BASENAME)
+        if os.path.isfile(direct):
+            return direct
+        if not os.path.isdir(d):
+            continue
+        try:
+            for name in os.listdir(d):
+                if name.startswith(stem + "_") and name.endswith(ext):
+                    cand = os.path.join(d, name)
+                    if os.path.isfile(cand):
+                        return cand
+        except OSError:
+            pass
+    return None
+
+
+def apply_shipped_path_cache(fs):
+    """Load a submit-time path-cache snapshot into ``fs`` (once per process)."""
+    global _shipped_path_cache_entries
+    if _shipped_path_cache_entries is None:
+        _shipped_path_cache_entries = []
+        path = _resolve_shipped_path_cache_file()
+        if path:
+            try:
+                with open(path) as f:
+                    data = json.load(f)
+                _shipped_path_cache_entries = data.get("entries") or []
+                os.environ[SHIPPED_PATH_CACHE_ENV] = path
+            except (OSError, ValueError, TypeError):
+                _shipped_path_cache_entries = []
+    pc = local_path_cache(fs)
+    if pc is None or not _shipped_path_cache_entries:
+        return
+    pc.load_entries(_shipped_path_cache_entries)
+
+
 class GFALFileInterface(RemoteFileInterface):
     local_prefix = "file://"
 
diff --git a/bootstrap.sh b/bootstrap.sh
index 9e7b0519..fe11673c 100644
--- a/bootstrap.sh
+++ b/bootstrap.sh
@@ -29,6 +29,24 @@ action() {
         export X509_USER_PROXY="${LAW_JOB_INIT_DIR}/voms.proxy"
     fi
 
+    # Rucio: pilot accounts (e.g. CRAB cmsplt01) are not valid RUCIO_ACCOUNT values.
+    # Prefer submit-time account (rendered below), then VOMS DN username, never pilot USER.
+    local rucio_account_submit="{{rucio_account}}"
+    if [ -z "${RUCIO_ACCOUNT:-}" ]; then
+        if [ -n "${rucio_account_submit}" ]; then
+            export RUCIO_ACCOUNT="${rucio_account_submit}"
+        elif [ -n "${X509_USER_PROXY:-}" ] && command -v voms-proxy-info >/dev/null 2>&1; then
+            # Standard CERN user cert: .../OU=Users/CN=<username>/CN=<digits>/...
+            export RUCIO_ACCOUNT="$(voms-proxy-info -identity 2>/dev/null \
+                | sed -n 's|.*/OU=Users/CN=\([^/]*\).*|\1|p' | head -1)"
+        fi
+        if [ -n "${RUCIO_ACCOUNT:-}" ]; then
+            echo "bootstrap: RUCIO_ACCOUNT=${RUCIO_ACCOUNT}"
+        else
+            echo "bootstrap: WARNING: RUCIO_ACCOUNT unset (Rucio client may fail on this worker)"
+        fi
+    fi
+
     if [ -n "${bundle_list}" ]; then
         local lcg_setup="/cvmfs/sft.cern.ch/lcg/views/LCG_108a/x86_64-el9-gcc15-opt/setup.sh"
         if [ -f "${lcg_setup}" ]; then
@@ -86,20 +104,70 @@ action() {
             if [ -n "${flaf_cmssw_version}" ]; then
                 local cmssw_dir="${bundle_dir}/soft/${flaf_cmssw_version}"
                 if [ -d "${cmssw_dir}/src" ]; then
-                    echo "bootstrap: relocating CMSSW ${flaf_cmssw_version}"
+                    # Relocate a tarball-shipped CMSSW release to this worker path.
+                    # Do NOT `eval scram runtime` before ProjectRename: that bakes the
+                    # submit-host (AFS) LOCALTOP into the shell as CMSSW_BASE, and
+                    # anaTupleProducer (loadTF / scram tool info) then fails with
+                    # FileNotFoundError on the AFS path (no AFS on CRAB workers).
+                    echo "bootstrap: relocating CMSSW ${flaf_cmssw_version} -> ${cmssw_dir}"
                     source /cvmfs/cms.cern.ch/cmsset_default.sh 2>/dev/null || true
                     local prev_dir="${PWD}"
+                    # Discover submit-host LOCALTOP from the shipped Self file (if any).
+                    local old_localtop=""
+                    if [ -f "${cmssw_dir}/config/Self" ]; then
+                        old_localtop=$(sed -n 's/.*LOCALTOP="\([^"]*\)".*/\1/p' \
+                            "${cmssw_dir}/config/Self" 2>/dev/null | head -1)
+                        # XML form: <Project path="..."/>
+                        if [ -z "${old_localtop}" ]; then
+                            old_localtop=$(sed -n 's/.*path="\([^"]*\)".*/\1/p' \
+                                "${cmssw_dir}/config/Self" 2>/dev/null | head -1)
+                        fi
+                    fi
                     cd "${cmssw_dir}/src"
-                    eval "$(scramv1 runtime -sh 2>/dev/null)" || true
-                    scramv1 b ProjectRename "${cmssw_dir}" 2>/dev/null || true
+                    # ProjectRename updates scram's Self/LOCALTOP to cmssw_dir.
+                    if ! scramv1 b ProjectRename "${cmssw_dir}"; then
+                        echo "bootstrap: WARNING: scram ProjectRename failed; applying path sed fallback"
+                        if [ -n "${old_localtop}" ] && [ "${old_localtop}" != "${cmssw_dir}" ]; then
+                            # Escape for sed s|old|new|g: BRE metacharacters on the
+                            # pattern side; \, &, | on both sides (delimiter specials).
+                            local old_esc new_esc
+                            old_esc=$(printf '%s' "${old_localtop}" | sed 's/[.[\*^$\\|\&]/\\&/g')
+                            new_esc=$(printf '%s' "${cmssw_dir}" | sed 's/[\\|\&]/\\&/g')
+                            # Patch Self + .SCRAM metadata that embed the submit-host path.
+                            if [ -f "${cmssw_dir}/config/Self" ]; then
+                                sed -i "s|${old_esc}|${new_esc}|g" "${cmssw_dir}/config/Self"
+                            fi
+                            find "${cmssw_dir}/.SCRAM" -type f 2>/dev/null \
+                                | xargs -r sed -i "s|${old_esc}|${new_esc}|g"
+                        fi
+                    fi
+                    # Drop any AFS-tainted CMSSW_* vars that a prior runtime might have set.
+                    unset CMSSW_BASE CMSSW_RELEASE_BASE CMSSW_SEARCH_PATH RELEASETOP LOCALTOP 2>/dev/null || true
                     cd "${prev_dir}"
+                    # Verify scram now resolves to the bundle path (best-effort).
+                    if [ -n "${old_localtop}" ]; then
+                        local new_self
+                        new_self=$(sed -n 's/.*path="\([^"]*\)".*/\1/p' \
+                            "${cmssw_dir}/config/Self" 2>/dev/null | head -1)
+                        [ -z "${new_self}" ] && new_self=$(sed -n 's/.*LOCALTOP="\([^"]*\)".*/\1/p' \
+                            "${cmssw_dir}/config/Self" 2>/dev/null | head -1)
+                        echo "bootstrap: CMSSW Self path now: ${new_self:-unknown}"
+                    fi
                 fi
             fi
         fi
 
         echo "bootstrap: sourcing env.sh from bundle"
+        # Ensure no leftover AFS CMSSW_BASE from the pilot / prior steps.
+        unset CMSSW_BASE CMSSW_RELEASE_BASE RELEASETOP LOCALTOP 2>/dev/null || true
         export FLAF_NO_INSTALL=1
         source "${bundle_dir}/env.sh"
+        # After env.sh, force CMSSW_BASE to the bundled release when present so
+        # subprocesses (includeLibTool, HHbtag models) never see an AFS path.
+        if [ -n "${FLAF_CMSSW_BASE:-}" ] && [ -d "${FLAF_CMSSW_BASE}" ]; then
+            export CMSSW_BASE="${FLAF_CMSSW_BASE}"
+            echo "bootstrap: CMSSW_BASE=${CMSSW_BASE}"
+        fi
     else
         if [ "${analysis_path}" = "NONE" ]; then
             echo "ERROR: analysis_path is NONE but no bundle_list was provided"
@@ -117,5 +185,10 @@ action() {
         [ -n "${corrections_path}" ] && export CORRECTIONS_PATH="${corrections_path}"
         source "${analysis_path}/env.sh"
     fi
+
+    # Re-assert after env.sh (Rucio cvmfs init may re-read USER on some pilots).
+    if [ -n "${rucio_account_submit}" ]; then
+        export RUCIO_ACCOUNT="${rucio_account_submit}"
+    fi
 }
 action
diff --git a/docs/configuration/user-custom.md b/docs/configuration/user-custom.md
index 8ec854ab..598e8aff 100644
--- a/docs/configuration/user-custom.md
+++ b/docs/configuration/user-custom.md
@@ -36,7 +36,8 @@ Replace `<initial>`/`<user>` with yours (e.g. `k` / `kandroso`). With just this,
 | `compute_unc_histograms` | bool | Whether to also fill histograms for those variations. Prefer setting this **per histTuple flavor** in `global.yaml` (`histTuple_flavors.<flavor>.compute_unc_histograms`) — uncertainties are usually only needed for the limit-setting shape flavor, so the flavor should dictate it. The value here (or in `global.yaml`) is used as the fallback when the active flavor does not set it. |
 | `store_noncentral` | bool | Whether to keep the non-central (systematic-shift) outputs, not just the central one. |
 | `remove_merged_inputs` | bool | If `true`, `HistMergerTask` deletes each variable's per-chunk split histograms (`HistFromNtupleProducerTask` outputs) after merging, to save space, leaving a tiny per-chunk `.merged` marker in place of each. Safe: the producer stays "complete" for exactly the chunks that were merged (it finds the split *or* its marker), so the task graph stays consistent (no re-run); a chunk that was never produced has no marker and is still produced. Default `false` — intermediates are kept. |
-| `variables` | list | Restrict which variables are produced/plotted. Omit for the full set. |
+| `variables` | list | Restrict which variables are produced/plotted (applied to the active `histTuple_flavor` list). If that flavor's variable list is empty (e.g. H_mumu `default`), this list is used as the active set. Omit for the full flavor set. |
+| `histTuple_flavor` | string | Optional. Selects which `histTuple_flavors` entry drives the variable lists (e.g. `CI` for the short H_mumu CI set). |
 | `hist_from_ntuple_max_hists` | int | Max histograms `HistFromNtupleProducerTask` books in one RDataFrame pass. The count is variables × selections × (Central + every Up/Down). Default `4000`; `0` disables batching. Lower this (do not raise CI memory) if a job OOMs. |
 
 !!! tip "`TestModel` is the fast path"
diff --git a/docs/workflow/arguments.md b/docs/workflow/arguments.md
index d0972c46..3b63f56d 100644
--- a/docs/workflow/arguments.md
+++ b/docs/workflow/arguments.md
@@ -14,7 +14,7 @@ also provides built-in options for status and cleanup.
 |---|---|---|
 | `--version` | *(required)* | Label that namespaces this run's outputs. Different versions never collide. |
 | `--period` | *(required)* | The [era](../concepts/eras.md), e.g. `Run3_2022`. |
-| `--workflow` | `local` | `local` (this machine) or `htcondor` (batch). See [HTCondor](htcondor.md). |
+| `--workflow` | `local` | `local` (this machine), `htcondor` (CERN batch), or `crab` (WLCG). See [HTCondor](htcondor.md) and [CRAB](crab.md). |
 | `--branches` | *(all)* | Which branches to run, e.g. `0`, `0,2`, `5-7`. Restricts only the launched task, not its dependencies. |
 | `--test` | `-1` | Process only N events per input file (`-1` = all). Great for smoke tests. |
 | `--process` | `""` | Restrict to one process (e.g. `custom_CI_Signal`). |
@@ -28,13 +28,27 @@ also provides built-in options for status and cleanup.
 | Option | Default | Meaning |
 |---|---|---|
 | `--transfer-logs` | off | Bring job logs back to `data/`. Recommended. |
-| `--parallel-jobs` | *(unbounded)* | Cap concurrent branches, e.g. `--parallel-jobs 100`. |
+| `--parallel-jobs` | unbounded (HTCondor) / **5000** (CRAB) | Cap concurrent jobs. On CRAB this is also the max size of each CRAB task. |
 | `--max-runtime` | *(task default)* | Per-job wall-clock limit. |
 | `--n-cpus` | `1` | CPUs requested per job. |
-| `--priority` | `0` | Job priority. |
-| `--bundle` | off | Ship a code/environment tarball to the worker. See [HTCondor → bundles](htcondor.md#bundles-shipping-the-code-to-workers). |
+| `--priority` | `0` | Job priority (HTCondor). |
+| `--bundle` | off | Ship a code/environment tarball to the worker. See [HTCondor → bundles](htcondor.md#bundles-shipping-the-code-to-workers). Always on for `--workflow crab`. |
 | `--htcondor-spool` | off | Spool job files to the schedd. |
 
+## CRAB options (on every workflow task)
+
+| Option | Default | Meaning |
+|---|---|---|
+| `--workflow crab` | — | Submit branches via CMS CRAB (WLCG). See [CRAB](crab.md). |
+
+Optional site white/black lists go in `global.yaml` under `crab:` (not CLI flags).
+Unset whitelist ⇒ all T1/T2/T3 sites. Default `--parallel-jobs` on CRAB is 5000
+(`crab.parallel_jobs`); a new CRAB task is submitted only when at least
+`crab.refill_fraction` (default 0.2) of those slots are free. `Site.storageSite`
+/ `Data.outLFNDirBase` are derived from `fs_default`. Memory is
+`2000 MB * n_cpus` (`crab.memory_mb_per_cpu`; CRAB / site-guaranteed default),
+capped at the CRAB client limit (5000 MB for 1 core, `2500 MB * n_cpus` otherwise).
+
 ## Status & cleanup (LAW built-ins)
 
 | Option | Meaning |
diff --git a/docs/workflow/crab.md b/docs/workflow/crab.md
new file mode 100644
index 00000000..c416ecac
--- /dev/null
+++ b/docs/workflow/crab.md
@@ -0,0 +1,163 @@
+# Running on CRAB (WLCG)
+
+HTCondor covers the CERN local batch farm. For jobs that should run anywhere on the
+**WLCG** (CMS CRAB), FLAF tasks can be submitted with `--workflow crab`. The implementation
+uses [law's CMS CRAB workflow](https://github.com/riga/law) (`law.contrib.cms.CrabWorkflow`).
+
+Analysis **outputs and job logs** use FLAF remote I/O only (`fs_default` via gfal/`davs://`,
+plus `stageout_logs.sh`). CRAB `transferOutputs` / `transferLogs` are forced **off** so
+nothing is duplicated onto CRAB's stageout area. The CRAB client still needs
+`Site.storageSite` / `Data.outLFNDirBase` for a submit-time write check — those
+fields are derived from `fs_default`, not configured separately.
+
+## Prerequisites
+
+1. A valid **VOMS proxy** for the CMS VO (`voms-proxy-init --voms cms -valid 192:00`).
+2. A **MyProxy** credential valid for **at least 5 days**, registered so **CRAB task
+   workers can retrieve it**. A plain `myproxy-init -d -n` is not enough: the credential
+   must use the SHA1 username law/CRAB expect and include CRAB retriever DNs. From an
+   existing VOMS proxy (no grid-cert passphrase):
+
+   ```sh
+   export X509_USER_PROXY=/tmp/x509up_u$(id -u)   # or your proxy path
+   # identity DN → SHA1 username (same as law)
+   SHA1=$(python3 - <<'PY'
+   import hashlib, subprocess
+   out = subprocess.check_output(["voms-proxy-info", "-identity"], text=True).strip()
+   print(hashlib.sha1(out.encode()).hexdigest())
+   PY
+   )
+   RETR='/DC=ch/DC=cern/OU=computers/CN=crab-(preprod|prod|dev)-tw(01|02|03).cern.ch|/DC=ch/DC=cern/OU=computers/CN=stefanov(m|m2).cern.ch|/DC=ch/DC=cern/OU=computers/CN=dciangot-tw.cern.ch|/DC=ch/DC=cern/OU=computers/CN=crab-(preprod|prod)-tw(01|02).cern.ch|/DC=ch/DC=cern/OU=computers/CN=crab-dev-tw(01|02|03|04).cern.ch|/DC=ch/DC=cern/OU=Organic Units/OU=Users/CN=cmscrab/CN=(817881|373708)/CN=Robot: cms crab|/DC=ch/DC=cern/OU=Organic Units/OU=Users/CN=crabint1/CN=373708/CN=Robot: CMS CRAB Integration 1'
+   GT_PROXY_MODE=rfc myproxy-init -n -s myproxy.cern.ch \
+     -C "$X509_USER_PROXY" -y "$X509_USER_PROXY" \
+     -l "$SHA1" -t 168 -c 168 \
+     -x -R "$RETR" -x -Z "$RETR" -m cms
+   myproxy-info -s myproxy.cern.ch -l "$SHA1"   # expect timeleft >= 5 days + retrieval policy
+   ```
+
+   FLAF fails early if the VOMS proxy or a suitable MyProxy credential is missing
+   (instead of waiting for server-side `SUBMITFAILED`). There is no password-file
+   fallback.
+3. CRAB client available (via CMSSW / the law CMSSW sandbox; default sandbox name is set in
+   law's config as `job.crab_sandbox_name`).
+4. **Bundles**: CRAB workers do not mount AFS. FLAF always ships code via `BundleTask`
+   when `--workflow crab` is used (same tarballs as HTCondor `--bundle`). Tasks already
+   declare `bundle_flavours`.
+5. Remote `fs_default` (e.g. `davs://eoshome-...`) so bundles and analysis outputs are on a
+   grid-accessible filesystem.
+
+## Config
+
+CRAB's write-check site is taken from `fs_default`:
+
+| `fs_default` | CRAB `storageSite` + `outLFNDirBase` |
+|---|---|
+| `T3_CH_CERNBOX:/store/user/<you>/...` | as written |
+| `davs://eoshome-<initial>.cern.ch:.../eos/user/<initial>/<you>/...` | `T3_CH_CERNBOX` + `/store/user/<you>/...` |
+
+The CRAB client requires `Site.whitelist` because law uses dummy `userInputFiles`
+(no input dataset). FLAF defaults that list to `T1_*`, `T2_*`, `T3_*` so jobs
+can run at every CMS processing site. Restrict or exclude sites only if you need
+to, in `global.yaml` / `user_custom.yaml`:
+
+```yaml
+crab:
+  # whitelist: [T2_CH_CERN]   # omit to use all T1/T2/T3 sites
+  # blacklist: [T2_US_MIT]
+  # parallel_jobs: 5000       # default --parallel-jobs; CLI wins if set
+  # refill_fraction: 0.2      # new CRAB task only when this fraction of slots is free
+  # memory_mb_per_cpu: 2000   # CRAB maxMemoryMB / n_cpus
+```
+
+Memory is `2000 MB * n_cpus` (override with `crab.memory_mb_per_cpu`), matching
+the CRAB default that all sites guarantee per core. Then capped at the CRAB
+client limit: 5000 MB for 1 core, `2500 MB * n_cpus` otherwise. There is no
+separate CRAB memory CLI flag. AnaTuple production defaults to 4 cores (8 GB)
+so tautau CMSSW jobs fit; 2 cores only allow 5 GB.
+
+Verify write access before the first campaign:
+
+```sh
+crab checkwrite --site=T3_CH_CERNBOX --lfn=/store/user/$USER
+```
+
+| Key | Meaning |
+|---|---|
+| `whitelist` | Optional. Restricts `Site.whitelist`. Default: `T1_*`, `T2_*`, `T3_*`. |
+| `blacklist` | Optional. CRAB `Site.blacklist` (applied on top of the whitelist). |
+| `parallel_jobs` | Optional. Default for `--parallel-jobs` on CRAB (CLI wins). Default: `5000`. Caps how many CRAB jobs are in flight and thus the size of each CRAB task. CRAB itself refuses more than 10 000 jobs in one task. |
+| `refill_fraction` | Optional. Submit a new CRAB task only when `parallel_jobs - n_active >= refill_fraction * parallel_jobs`. Default: `0.2`. Prevents a 1-job task every time a single job finishes. |
+| `memory_mb_per_cpu` | Optional. CRAB `JobType.maxMemoryMB` is this times `--n-cpus`, capped at 5000 MB (1 core) or `2500 MB * n_cpus`. Default: `2000` (CRAB / site-guaranteed per-core default). |
+
+## Submit
+
+```sh
+law run FLAF.Analysis.tasks.HistTupleProducerTask \
+  --period Run3_2022EE --version my_crab \
+  --workflow crab \
+  --branches 0 \
+  --test 1000 \
+  --user-custom /path/to/user_custom_with_crab.yaml
+```
+
+| Option | Why |
+|---|---|
+| `--workflow crab` | Submit via CRAB instead of local/HTCondor. |
+| `--parallel-jobs` | Jobs in flight (default **5000** on CRAB, unlimited on HTCondor). Each refill is one CRAB task. Also `crab.parallel_jobs` in `global.yaml`. |
+| `--max-runtime` / `--n-cpus` | Same as HTCondor; mapped to CRAB `maxJobRuntimeMin` / `numCores` / memory (`2000 MB * n_cpus`, CRAB-capped). |
+| `--transfer-logs` | On by default; enables remote log stageout when `fs_default` is WLCG. |
+
+You do **not** need `--bundle` for CRAB — bundles are forced whenever the workflow is `crab`.
+
+## How it fits with HTCondor + bundles
+
+| Mode | Code on worker | Typical use |
+|---|---|---|
+| `--workflow local` | Submit machine | Development, small tests |
+| `--workflow htcondor` | AFS (or `--bundle` tarball) | CERN farm production |
+| `--workflow htcondor --bundle` | Tarball from `fs_default` | HTCondor without AFS dependency |
+| `--workflow crab` | Tarball from `fs_default` (always) | Full WLCG via CRAB |
+
+## Monitor
+
+```sh
+law run FLAF.Analysis.tasks.HistTupleProducerTask \
+  --period Run3_2022EE --version my_crab --print-status 1,1
+```
+
+CRAB project directories live under `data/jobs/` (see `job.job_file_dir` in `law.cfg`). You can
+also use `crab status -d <project_dir>` from a CMSSW environment.
+
+## Caveats
+
+!!! warning "MyProxy must stay valid"
+    CRAB polls through MyProxy. Delegate a long-lived proxy before large campaigns
+    (`myproxy-init` as in Prerequisites).
+
+!!! note "Path-existence cache is shipped with the job"
+    `WLCGFileSystem.remotePathCacheHost` (`cms-flaf.cern.ch`) is behind the CERN
+    firewall, so CRAB workers do not use it. At submit time FLAF dumps the
+    in-process path cache and ships it with the job; the worker loads that
+    snapshot and uses a longer local TTL (`24 × localPathCacheValidity`, at
+    least 24 h) so concurrent jobs do not re-stat the same remote paths.
+
+!!! warning "First-time CRAB / grid mapfile"
+    New users may need a CRAB username mapping and write access to the chosen storage site
+    LFN. At CERN, prefer `T3_CH_CERNBOX` for `/store/user/...` (maps to personal EOS and
+    usually passes `crab checkwrite`); `T2_CH_CERN /store/user` often does not exist.
+
+!!! note "Distant sites still read `fs_default`"
+    The default whitelist lets jobs run anywhere, but the bundle and outputs stay
+    on `fs_default`. Personal EOS (`davs://eoshome-*.cern.ch`) can fail or stall
+    from far-away sites (gfal 112, HTTP 404, hung DNN). Law retries usually
+    recover; set `crab.whitelist` closer to CERN if that I/O is a problem.
+
+!!! warning "Do not replace a live bundle mid-campaign"
+    `BundleTask` can stay DONE after `core.tar.bz2` is deleted because of the
+    path-existence cache. Workers then get HTTP 404. Rebuild into a sibling file
+    and `mv` it over the live path; do not `cp` onto a file jobs may be
+    downloading (a mid-copy can stage out 0 bytes).
+
+!!! note "Test small first"
+    Validate with `--workflow local --branches 0 --test 1000`, then a single CRAB branch,
+    before large submissions.
diff --git a/docs/workflow/htcondor.md b/docs/workflow/htcondor.md
index 50be0576..c575b13d 100644
--- a/docs/workflow/htcondor.md
+++ b/docs/workflow/htcondor.md
@@ -53,6 +53,10 @@ A batch worker needs your code and environment. FLAF supports two modes:
 For most work the defaults are correct; you only think about bundles when a stage explicitly needs
 one (e.g. it declares a CMSSW bundle flavour) or when AFS is not available on the target pool.
 
+For jobs that should run on the full CMS WLCG (not only CERN HTCondor), use
+[`--workflow crab`](crab.md) — that path always uses bundles.
+
+
 !!! tip "Your edits to FLAF *do* reach the workers"
     Thanks to the dev overlay, non-bundle jobs run your edited `FLAF`/`Corrections`, and bundle
     jobs include them in the tarball — so testing framework changes on HTCondor works without
diff --git a/mkdocs.yml b/mkdocs.yml
index f3cd8003..bcc840f9 100644
--- a/mkdocs.yml
+++ b/mkdocs.yml
@@ -88,6 +88,7 @@ nav:
   - Full workflow:
       - Walkthrough: workflow/walkthrough.md
       - Running on HTCondor: workflow/htcondor.md
+      - Running on CRAB: workflow/crab.md
       - Command arguments: workflow/arguments.md
   - Configuration guide:
       - user_custom.yaml: configuration/user-custom.md
diff --git a/run_tools/law_customizations.py b/run_tools/law_customizations.py
index bc8affd8..5cdcfcdd 100644
--- a/run_tools/law_customizations.py
+++ b/run_tools/law_customizations.py
@@ -9,6 +9,8 @@
 import subprocess
 import tempfile
 
+from collections import OrderedDict
+
 from law.parser import global_cmdline_values
 
 from FLAF.RunKit.run_tools import natural_sort
@@ -17,6 +19,7 @@
 from FLAF.Common.Setup import Setup
 
 law.contrib.load("htcondor")
+law.contrib.load("cms")
 
 
 def copy_param(ref_param, new_default):
@@ -149,11 +152,7 @@ def __init__(self, *args, **kwargs):
         super(Task, self).__init__(*args, **kwargs)
         user_custom_file = None
         if self.user_custom:
-            user_custom_file = self.user_custom
-            if not os.path.isabs(user_custom_file):
-                user_custom_file = os.path.join(
-                    os.getenv("ANALYSIS_PATH"), user_custom_file
-                )
+            user_custom_file = self._resolve_user_custom_path(self.user_custom)
         self.setup = Setup.getGlobal(
             os.getenv("ANALYSIS_PATH"),
             self.period,
@@ -168,6 +167,46 @@ def __init__(self, *args, **kwargs):
         self._dataset_id_name_dict = None
         self._dataset_name_id_dict = None
 
+    @staticmethod
+    def _resolve_user_custom_path(user_custom):
+        from FLAF.Common.Setup import resolve_user_custom_path
+
+        return resolve_user_custom_path(user_custom)
+
+    def _stage_user_custom_input(self, config):
+        """Ship user_custom yaml as a job input for remote workers (bundle/CRAB)."""
+        if not self.user_custom:
+            return
+        path = self.user_custom
+        if not os.path.isabs(path):
+            path = os.path.join(os.getenv("ANALYSIS_PATH") or "", path)
+        if not path or not os.path.isfile(path):
+            return
+        from law.job.base import JobInputFile
+
+        # share=True, increment=False keeps a stable basename when possible; resolve
+        # still accepts law's hashed names if increment is forced elsewhere.
+        config.input_files["user_custom"] = JobInputFile(
+            path=path, copy=True, share=True, render=False, increment=False
+        )
+
+    def _stage_path_cache_input(self, config):
+        """Dump the submit-process path cache and ship it with the CRAB job."""
+        from law.job.base import JobInputFile
+        from FLAF.RunKit.law_gfal import (
+            SHIPPED_PATH_CACHE_BASENAME,
+            collect_setup_path_cache_entries,
+            write_path_cache_file,
+        )
+
+        out_dir = self.local_path()
+        os.makedirs(out_dir, exist_ok=True)
+        path = os.path.join(out_dir, SHIPPED_PATH_CACHE_BASENAME)
+        write_path_cache_file(path, collect_setup_path_cache_entries(self.setup))
+        config.input_files["path_cache"] = JobInputFile(
+            path=path, copy=True, share=True, render=False, increment=False
+        )
+
     # Process-local memoization of create_branch_map results, shared across task
     # instances. The same branch map is otherwise rebuilt many times during task
     # initialization because every `X.req(...).create_branch_map()` constructs a fresh
@@ -472,6 +511,17 @@ def _get_bundle_source(pat: str) -> str:
                         f"No files found for bundle flavour '{self.flavour}'"
                     )
 
+                # CMSSW analysis customisations (HHbtag, ClassicSVfit, …) are installed as
+                # absolute AFS symlinks under soft/CMSSW_*/src. On CRAB those targets do not
+                # exist. Materialize any absolute symlink that points outside the staging
+                # tree so the tarball is self-contained. Relative / internal links stay.
+                if self.flavour == "cmssw":
+                    n_mat = self._materialize_external_symlinks(staging)
+                    if n_mat:
+                        print(
+                            f"bundle[cmssw]: materialized {n_mat} external symlink(s)"
+                        )
+
                 subprocess.run(
                     [
                         "tar",
@@ -488,6 +538,51 @@ def _get_bundle_source(pat: str) -> str:
                 )
         print(f"bundle[{self.flavour}]: done")
 
+    @staticmethod
+    def _materialize_external_symlinks(root: str) -> int:
+        """Replace absolute external symlinks under *root* with real file/dir copies.
+
+        Returns the number of symlinks replaced. Relative symlinks and absolute ones
+        that already resolve inside *root* are left unchanged.
+        """
+        root_real = os.path.realpath(root)
+        n = 0
+        # Collect first so we do not walk into trees we just replaced.
+        external = []
+        for dirpath, dirnames, filenames in os.walk(root, followlinks=False):
+            for name in dirnames + filenames:
+                path = os.path.join(dirpath, name)
+                if not os.path.islink(path):
+                    continue
+                target = os.readlink(path)
+                if not os.path.isabs(target):
+                    continue
+                # Resolve once; skip broken links with a warning.
+                try:
+                    resolved = os.path.realpath(path)
+                except OSError:
+                    print(f"bundle[cmssw]: warning: broken symlink {path} -> {target}")
+                    continue
+                if not os.path.exists(resolved):
+                    print(
+                        f"bundle[cmssw]: warning: dangling symlink {path} -> {target}"
+                    )
+                    continue
+                # Already points inside the staging tree → fine to keep.
+                if resolved == root_real or resolved.startswith(root_real + os.sep):
+                    continue
+                external.append((path, resolved))
+
+        for path, resolved in external:
+            os.unlink(path)
+            if os.path.isdir(resolved):
+                shutil.copytree(resolved, path, symlinks=True)
+            else:
+                shutil.copy2(resolved, path)
+            n += 1
+            print(f"bundle[cmssw]: materialized {path} <- {resolved}")
+        return n
+
 
 class CERNHTCondorJobFileFactory(law.htcondor.HTCondorJobFileFactory):
     """HTCondor job file factory that stages transfer_input_files to EOS and uses protocol URLs.
@@ -589,7 +684,7 @@ class HTCondorWorkflow(law.htcondor.HTCondorWorkflow):
         default=False,
         significant=False,
         description="download pre-built bundle archives on workers instead of accessing AFS; "
-        "tasks declare which flavours they need via bundle_flavours",
+        "tasks declare which flavours they need via bundle_flavours. Always on for --workflow crab.",
     )
     htcondor_spool = luigi.BoolParameter(
         default=True,
@@ -606,35 +701,6 @@ class HTCondorWorkflow(law.htcondor.HTCondorWorkflow):
     ]
     bundle_flavours = []
 
-    def workflow_requires(self):
-        if self.bundle and self.bundle_flavours:
-            bundles = []
-            for item in self.bundle_flavours:
-                if isinstance(item, (list, tuple)) and len(item) == 2:
-                    flavour, bversion = item
-                    bundles.append(
-                        BundleTask.req(self, flavour=flavour, version=bversion)
-                    )
-                else:
-                    flavour = item
-                    bundles.append(BundleTask.req(self, flavour=flavour))
-            return {"bundles": bundles}
-        return {}
-
-    def htcondor_check_job_completeness(self):
-        return False
-
-    def htcondor_poll_callback(self, poll_data):
-        update_kinit(verbose=0)
-        return True
-
-    def htcondor_output_directory(self):
-        # the directory where submission meta data should be stored
-        return law.LocalDirectoryTarget(self.local_path())
-
-    def htcondor_log_directory(self):
-        return None
-
     def _flaf_root(self):
         # FLAF source root, respecting the dev overlay: flaf_dev.sh sets FLAF_PATH to
         # the top-level FLAF_all/FLAF, while the analysis env.sh sets it to the pinned
@@ -646,32 +712,62 @@ def _flaf_root(self):
             os.getenv("ANALYSIS_PATH"), "FLAF"
         )
 
-    def htcondor_stageout_file(self):
-        return os.path.join(self._flaf_root(), "run_tools", "stageout_logs.sh")
-
-    def htcondor_bootstrap_file(self):
-        # each job can define a bootstrap file that is executed prior to the actual job
-        # in order to setup software and environment variables
-        return os.path.join(self._flaf_root(), "bootstrap.sh")
-
-    def htcondor_job_file_factory_cls(self):
-        return CERNHTCondorJobFileFactory
+    def _uses_bundles(self):
+        """Whether this submission should ship and unpack code bundles on the worker.
 
-    def htcondor_job_config(self, config, job_num, branches):
+        Bundles are optional for HTCondor (shared AFS is available) but required for CRAB
+        (WLCG workers have no AFS mount).
+        """
+        if not self.bundle_flavours:
+            return False
+        if getattr(self, "effective_workflow", None) == "crab":
+            return True
+        return bool(self.bundle)
+
+    def _bundle_requirements(self):
+        """Return BundleTask requirements for configured flavours (empty if unused)."""
+        if not self._uses_bundles():
+            return {}
+        bundles = []
+        for item in self.bundle_flavours:
+            if isinstance(item, (list, tuple)) and len(item) == 2:
+                flavour, bversion = item
+                bundles.append(BundleTask.req(self, flavour=flavour, version=bversion))
+            else:
+                bundles.append(BundleTask.req(self, flavour=item))
+        return {"bundles": bundles}
+
+    def _apply_bundle_render_variables(self, config):
+        """Set bootstrap render variables for bundle download (or clear them)."""
+        if not self._uses_bundles():
+            config.render_variables["bundle_list"] = ""
+            return
+        if not isinstance(self.fs_default, WLCGFileSystem):
+            raise RuntimeError(
+                "bundle / crab workflows require fs_default to be a remote filesystem "
+                "(davs://, root://, ...)"
+            )
+        bundle_parts = []
+        for item in self.bundle_flavours:
+            if isinstance(item, (list, tuple)) and len(item) == 2:
+                flavour, bversion = item
+            else:
+                flavour = item
+                bversion = self.version
+            bundle_url = self.remote_target(
+                bversion, "bundles", self.period, f"{flavour}.tar.bz2"
+            ).uri()
+            bundle_parts.append(f"{flavour}:{bundle_url}")
+        config.render_variables["bundle_list"] = " ".join(bundle_parts)
+
+    def _apply_bootstrap_path_render_variables(self, config):
+        """Set analysis_path / FLAF_PATH / CORRECTIONS_PATH / token-server for bootstrap.sh."""
         ana_path = os.getenv("ANALYSIS_PATH")
-        # NON-bundle jobs run on the shared AFS workspace and source the analysis env.sh
-        # there.  Forward FLAF_PATH / CORRECTIONS_PATH so the worker uses the same FLAF /
-        # Corrections as the submit side (the submodule copies in production, or the edited
-        # top-level copies when flaf_dev.sh is active) — bootstrap.sh exports them before
-        # sourcing env.sh.  In production these equal $ANALYSIS_PATH/FLAF(/Corrections), so
-        # forwarding them is transparent.
-        #
-        # BUNDLE jobs instead set analysis_path=NONE and ship FLAF / Corrections inside the
-        # tarball; they must NOT receive FLAF_PATH / CORRECTIONS_PATH, otherwise the in-bundle
-        # env.sh would point them back at the AFS workspace and the worker would access AFS.
+        # Bundle (and always-on CRAB) jobs unpack code on the worker and must not point back
+        # at AFS. Non-bundle HTCondor jobs source the shared workspace and forward overlay paths.
         flaf_path = ""
         corrections_path = ""
-        if self.bundle and self.bundle_flavours:
+        if self._uses_bundles():
             config.render_variables["analysis_path"] = "NONE"
         else:
             config.render_variables["analysis_path"] = ana_path
@@ -679,23 +775,64 @@ def htcondor_job_config(self, config, job_num, branches):
             corrections_path = os.getenv("CORRECTIONS_PATH", "") or ""
         config.render_variables["flaf_path"] = flaf_path
         config.render_variables["corrections_path"] = corrections_path
+        # Rucio account for workers: CRAB pilots have USER=cmsplt01, which is not a Rucio
+        # account. Bake the submitter account so bootstrap can export RUCIO_ACCOUNT.
+        config.render_variables["rucio_account"] = (
+            os.environ.get("RUCIO_ACCOUNT") or os.environ.get("USER") or ""
+        )
 
-        # token server for rate-limiting job starts to avoid AFS overload.
-        # Not needed in bundle mode: workers never touch AFS, so there is no load concern.
         runTokenServer = self.global_params.get("runTokenServer", None)
-        if runTokenServer and not (self.bundle and self.bundle_flavours):
+        if runTokenServer and not self._uses_bundles():
             config.render_variables["run_token_server_host"] = runTokenServer["host"]
             config.render_variables["run_token_server_port"] = str(
                 runTokenServer["port"]
             )
-            # ship get_token.py with the job so it is available before AFS is accessed
             config.input_files["get_token_script"] = os.path.join(
-                ana_path, "FLAF", "run_tools", "get_run_token.py"
+                self._flaf_root(), "run_tools", "get_run_token.py"
             )
         else:
             config.render_variables["run_token_server_host"] = ""
             config.render_variables["run_token_server_port"] = ""
 
+    def _log_remote_base_url(self):
+        # Must match remote_log_dir_target() (used by --print-status and the
+        # HTCondor submit proxy) so producer sub-paths stay consistent.
+        if isinstance(self.fs_default, WLCGFileSystem):
+            return self.remote_log_dir_target().uri()
+        return ""
+
+    def workflow_requires(self):
+        return self._bundle_requirements()
+
+    def htcondor_check_job_completeness(self):
+        return False
+
+    def htcondor_poll_callback(self, poll_data):
+        update_kinit(verbose=0)
+        return True
+
+    def htcondor_output_directory(self):
+        # the directory where submission meta data should be stored
+        return law.LocalDirectoryTarget(self.local_path())
+
+    def htcondor_log_directory(self):
+        return None
+
+    def htcondor_stageout_file(self):
+        return os.path.join(self._flaf_root(), "run_tools", "stageout_logs.sh")
+
+    def htcondor_bootstrap_file(self):
+        # each job can define a bootstrap file that is executed prior to the actual job
+        # in order to setup software and environment variables
+        return os.path.join(self._flaf_root(), "bootstrap.sh")
+
+    def htcondor_job_file_factory_cls(self):
+        return CERNHTCondorJobFileFactory
+
+    def htcondor_job_config(self, config, job_num, branches):
+        self._apply_bootstrap_path_render_variables(config)
+        self._stage_user_custom_input(config)
+
         # force to run on AlmaLinux9, https://batchdocs.web.cern.ch/local/submit.html
         config.custom_content.append(
             ("requirements", 'TARGET.OpSysAndVer =?= "AlmaLinux9"')
@@ -718,10 +855,7 @@ def htcondor_job_config(self, config, job_num, branches):
             ("environment", '"LAW_HTCONDOR_JOB_POSTFIX=$(law_job_postfix)"')
         )
 
-        # Compute the remote destination directory for the stageout script.
-        log_remote_base_url = ""
-        if isinstance(self.fs_default, WLCGFileSystem):
-            log_remote_base_url = self.remote_log_dir_target().uri()
+        log_remote_base_url = self._log_remote_base_url()
         config.render_variables["log_remote_base_url"] = log_remote_base_url
 
         # Redirect the sandbox log copy to /dev/null only when stageout will
@@ -730,31 +864,11 @@ def htcondor_job_config(self, config, job_num, branches):
         if log_remote_base_url:
             config.output_files["stdall.txt"] = "/dev/null"
 
-        # bundle: build a space-separated list of "flavour:url" pairs for bootstrap.sh.
-        if self.bundle and self.bundle_flavours:
-            if not isinstance(self.fs_default, WLCGFileSystem):
-                raise RuntimeError(
-                    "--bundle requires fs_default to be a remote filesystem (davs://, root://, ...)"
-                )
-            bundle_parts = []
-            for item in self.bundle_flavours:
-                if isinstance(item, (list, tuple)) and len(item) == 2:
-                    flavour, bversion = item
-                else:
-                    flavour = item
-                    bversion = self.version
-                bundle_url = self.remote_target(
-                    bversion, "bundles", self.period, f"{flavour}.tar.bz2"
-                ).uri()
-                bundle_parts.append(f"{flavour}:{bundle_url}")
-            config.render_variables["bundle_list"] = " ".join(bundle_parts)
-
-            if not self.htcondor_spool:
-                config._worker_files_remote_dir = self.remote_dir_target(
-                    self.version, "worker_files", self.period
-                )
-        else:
-            config.render_variables["bundle_list"] = ""
+        self._apply_bundle_render_variables(config)
+        if self._uses_bundles() and not self.htcondor_spool:
+            config._worker_files_remote_dir = self.remote_dir_target(
+                self.version, "worker_files", self.period
+            )
 
         return config
 
@@ -839,3 +953,483 @@ def _submit_group(self, *args, **kwargs):
 # our `_submit_group` override (remote log path rewrite) would never run.  Flip the
 # flag so this class is recognised as the "htcondor" workflow provider.
 HTCondorWorkflow._defined_workflow_proxy = True
+
+
+class FLAFCrabJobFileFactory(law.cms.CrabJobFileFactory):
+    """CrabJobFileFactory for FLAF: no CRAB-side product/log stageout.
+
+    Analysis products and job logs are written by FLAF itself (remote targets via
+    gfal + ``stageout_logs.sh``). CRAB is used only as a batch backend, so we force:
+
+    - ``General.transferOutputs = False``
+    - ``General.transferLogs = False``
+    - no ``JobType.outputFiles``
+    - ``JobType.disableAutomaticOutputCollection = True`` (law default)
+
+    ``Site.storageSite`` / ``Data.outLFNDirBase`` remain required by the CRAB client
+    for a valid config and the submit-time write check, but FLAF never places analysis
+    outputs there.
+
+    Also strips deprecated ``JobType.sendPythonFolder`` (rejected by modern CRAB).
+    """
+
+    def create(self, **kwargs):
+        # Prevent law from promoting custom_log_file into CRAB JobType.outputFiles
+        # (which would set transferOutputs=True and duplicate FLAF log stageout).
+        kwargs = dict(kwargs)
+        kwargs["output_files"] = []
+        # Keep a local log file name for the law job script if transfer_logs requested,
+        # but do not register it as a CRAB output.
+        custom_log = kwargs.get("custom_log_file")
+
+        job_file, c = super().create(**kwargs)
+
+        if hasattr(c, "crab"):
+            c.crab.General.transferOutputs = False
+            c.crab.General.transferLogs = False
+            if getattr(c.crab, "JobType", None) is not None:
+                c.crab.JobType.sendPythonFolder = None
+                c.crab.JobType.outputFiles = None
+                c.crab.JobType.disableAutomaticOutputCollection = True
+        c.output_files = []
+        if custom_log:
+            c.custom_log_file = custom_log
+
+        try:
+            self._rewrite_crab_job_file(job_file)
+        except Exception as exc:
+            print(f"WARNING: could not post-process crab job file {job_file}: {exc}")
+        return job_file, c
+
+    @staticmethod
+    def _rewrite_crab_job_file(job_file):
+        """Rewrite the generated CRAB cfg to drop output transfer and deprecated keys."""
+        with open(job_file) as f:
+            lines = f.readlines()
+
+        new_lines = []
+        skip_list = False
+        for ln in lines:
+            stripped = ln.strip()
+
+            # Skip deprecated option entirely.
+            if "sendPythonFolder" in ln:
+                continue
+
+            # Force no CRAB-side transfers (FLAF owns remote I/O).
+            if "General.transferOutputs" in ln:
+                new_lines.append("cfg.General.transferOutputs = False\n")
+                continue
+            if "General.transferLogs" in ln:
+                new_lines.append("cfg.General.transferLogs = False\n")
+                continue
+
+            # Drop JobType.outputFiles (single line or multi-line list).
+            if "JobType.outputFiles" in ln:
+                if stripped.endswith("[") or ("[" in stripped and "]" not in stripped):
+                    skip_list = True
+                continue
+            if skip_list:
+                if "]" in stripped:
+                    skip_list = False
+                continue
+
+            if "JobType.disableAutomaticOutputCollection" in ln:
+                new_lines.append(
+                    "cfg.JobType.disableAutomaticOutputCollection = True\n"
+                )
+                continue
+
+            new_lines.append(ln)
+
+        with open(job_file, "w") as f:
+            f.writelines(new_lines)
+
+
+# Require VOMS + MyProxy before submit. The CRAB server retrieves the user proxy
+# from myproxy.cern.ch (>= ~5 days remaining). A local VOMS proxy alone is not
+# enough: the client may accept the task, then the server returns SUBMITFAILED.
+# Do not fall back to interactive delegation or a law.cfg password file.
+_FLAFCrabWorkflowProxyBase = law.cms.CrabWorkflow.workflow_proxy_cls
+
+
+_CRAB_DEFAULT_PARALLEL_JOBS = 5000
+_CRAB_DEFAULT_REFILL_FRACTION = 0.2
+
+
+def _cli_has_parallel_jobs():
+    """True when the user passed ``--parallel-jobs`` (or a task-prefixed form)."""
+    parser = luigi.cmdline_parser.CmdlineParser.get_instance()
+    tokens = list(getattr(parser, "cmdline_args", None) or [])
+    for tok in tokens:
+        if tok in ("--parallel-jobs", "--parallel_jobs"):
+            return True
+        if tok.startswith("--parallel-jobs=") or tok.startswith("--parallel_jobs="):
+            return True
+        if tok.endswith("-parallel-jobs") or tok.endswith("-parallel_jobs"):
+            return True
+        if "-parallel-jobs=" in tok or "-parallel_jobs=" in tok:
+            return True
+    return False
+
+
+class _FLAFCrabWorkflowProxy(_FLAFCrabWorkflowProxyBase):
+    def __init__(self, *args, **kwargs):
+        super(_FLAFCrabWorkflowProxy, self).__init__(*args, **kwargs)
+        self._apply_crab_parallel_jobs()
+
+    def _crab_refill_fraction(self):
+        raw = self.task._crab_cfg().get(
+            "refill_fraction", _CRAB_DEFAULT_REFILL_FRACTION
+        )
+        try:
+            frac = float(raw)
+        except (TypeError, ValueError):
+            frac = _CRAB_DEFAULT_REFILL_FRACTION
+        return min(max(frac, 0.0), 1.0)
+
+    def _apply_crab_parallel_jobs(self):
+        """CRAB default is 5000 jobs in flight; yaml then CLI override.
+
+        Multi-workflow tasks inherit HTCondor's unlimited ``parallel_jobs``, so
+        the CrabWorkflow class default never wins. Apply the CRAB default here.
+        """
+        if _cli_has_parallel_jobs():
+            return
+        yaml_n = self.task._crab_cfg().get("parallel_jobs")
+        if yaml_n is not None:
+            self._set_parallel_jobs(int(yaml_n))
+            return
+        if self.poll_data.n_parallel == self.n_parallel_max:
+            self._set_parallel_jobs(_CRAB_DEFAULT_PARALLEL_JOBS)
+
+    def _should_submit_crab_group(self):
+        """Refill only when enough slots are free (default 20% of parallel_jobs).
+
+        The first wave always submits. Unlimited ``parallel_jobs`` keeps law's
+        original behaviour (one group with every remaining job).
+        """
+        n_parallel = self.poll_data.n_parallel
+        if n_parallel >= self.n_parallel_max:
+            return True
+        is_first_wave = (not self.job_data.jobs) and (not self._submitted)
+        if is_first_wave:
+            return True
+        free = n_parallel - self.poll_data.n_active
+        return free >= self._crab_refill_fraction() * n_parallel
+
+    def submit(self, retry_jobs=None):
+        if self._should_submit_crab_group():
+            return super(_FLAFCrabWorkflowProxy, self).submit(retry_jobs)
+
+        # Park retries as unsubmitted so the next eligible refill picks them up
+        # as one larger CRAB task instead of a 1-job task now.
+        if retry_jobs:
+            for job_num, branches in retry_jobs.items():
+                if self._can_skip_job(job_num, branches):
+                    continue
+                self.job_data.jobs.pop(job_num, None)
+                self.job_data.unsubmitted_jobs[job_num] = branches
+            self.dump_job_data()
+        return OrderedDict()
+
+    def setup_job_manager(self):
+        """Require a valid VOMS proxy and a MyProxy credential (>= 5 days)."""
+        proxy = os.environ.get("X509_USER_PROXY", "")
+        if not proxy or not os.path.isfile(proxy):
+            raise RuntimeError(
+                "CRAB submission requires a valid VOMS proxy (X509_USER_PROXY). "
+                "Run: voms-proxy-init --voms cms -valid 192:00"
+            )
+        if not law.wlcg.check_vomsproxy_validity(proxy_file=proxy):
+            raise RuntimeError(
+                f"VOMS proxy at {proxy} is missing or expired; run "
+                "`voms-proxy-init --voms cms -valid 192:00`"
+            )
+        kwargs = {"proxy": proxy}
+
+        min_myproxy_seconds = 5 * 24 * 3600
+
+        # MyProxy usernames may be either the DN (`myproxy-init -d`) or a SHA1 of
+        # the DN (law encode_username=True / some crab helpers). Accept either form.
+        for encode in (False, True):
+            try:
+                info = (
+                    law.wlcg.get_myproxy_info(encode_username=encode, silent=True) or {}
+                )
+            except Exception:
+                info = {}
+            if info.get("username") and info.get("timeleft", 0) >= min_myproxy_seconds:
+                kwargs["myproxy_username"] = info["username"]
+                return kwargs
+
+        raise RuntimeError(
+            "CRAB requires a MyProxy credential valid for at least 5 days "
+            "(CRAB server retrieves it from myproxy.cern.ch). "
+            "Run once interactively:\n"
+            "  myproxy-init -d -n -s myproxy.cern.ch\n"
+            "  # verify: myproxy-info -d -s myproxy.cern.ch  (timeleft >= 5 days)\n"
+            "See docs/workflow/crab.md for the CRAB-retriever form."
+        )
+
+
+_EOSHOME_FS_RE = re.compile(
+    r"^davs://eoshome-[a-z0-9]+\.cern\.ch(?::\d+)?/eos/user/[a-z0-9]/([^/]+)(/.*)?$",
+    re.IGNORECASE,
+)
+
+
+def _crab_stageout_from_fs_spec(fs_spec):
+    """Map ``fs_default`` to CRAB ``(storageSite, outLFNDirBase)``.
+
+    Accepted forms (same as storage docs):
+
+    - ``T3_CH_CERNBOX:/store/user/<user>/...``
+    - ``davs://eoshome-<initial>.cern.ch:.../eos/user/<initial>/<user>/...``
+      → ``T3_CH_CERNBOX`` + ``/store/user/<user>/...``
+    """
+    if isinstance(fs_spec, (list, tuple)):
+        if not fs_spec:
+            raise RuntimeError("fs_default is empty; CRAB needs a remote filesystem")
+        fs_spec = fs_spec[0]
+    if not isinstance(fs_spec, str) or not fs_spec.strip():
+        raise RuntimeError("fs_default must be a string (or list of strings)")
+    spec = fs_spec.strip().rstrip("/")
+
+    if "://" not in spec and ":" in spec:
+        site, lfn = spec.split(":", 1)
+        site, lfn = site.strip(), lfn.strip()
+        if site and lfn.startswith("/"):
+            return site, lfn
+
+    m = _EOSHOME_FS_RE.match(spec)
+    if m:
+        user, rest = m.group(1), m.group(2) or ""
+        return "T3_CH_CERNBOX", f"/store/user/{user}{rest}"
+
+    raise RuntimeError(
+        "CRAB derives Site.storageSite and Data.outLFNDirBase from fs_default. "
+        "Use a WLCG site path (T3_CH_CERNBOX:/store/user/<you>/...) or a CERN "
+        f"EOS davs://eoshome-... URL. Got: {fs_spec}"
+    )
+
+
+class CrabWorkflow(law.cms.CrabWorkflow):
+    """CRAB (WLCG) remote workflow, built on law.contrib.cms.CrabWorkflow.
+
+    CRAB is only the batch backend. **All analysis products and logs use FLAF remote
+    I/O** (``fs_default`` / gfal via task targets and ``stageout_logs.sh``). CRAB
+    ``transferOutputs`` / ``transferLogs`` / ``JobType.outputFiles`` are forced off so
+    nothing is duplicated onto CRAB's stageout area.
+
+    ``Site.storageSite`` / ``Data.outLFNDirBase`` are derived from ``fs_default``
+    (submit-time write check only). Memory is ``2000 MB * n_cpus`` (override
+    with ``crab.memory_mb_per_cpu``), matching the CRAB / site-guaranteed default.
+
+    Law injects dummy ``userInputFiles`` when ``Data.inputDataset`` is empty,
+    and the CRAB client then requires ``Site.whitelist``. If ``crab.whitelist``
+    is unset, FLAF defaults to ``T1_*`` / ``T2_*`` / ``T3_*`` so jobs can run
+    at every CMS processing site. Optional ``crab.blacklist`` still excludes
+    sites.
+
+    CRAB workers have no AFS, so code is always shipped via the existing BundleTask
+    mechanism (same as ``--bundle`` on HTCondor). Tasks must declare ``bundle_flavours``.
+
+    Config (``global.yaml`` / user_custom YAML), all optional::
+
+        crab:
+          # whitelist: [T2_CH_CERN]   # omit to use all T1/T2/T3 sites
+          # blacklist: [T2_US_MIT]
+          # parallel_jobs: 5000       # --parallel-jobs default; CLI wins
+          # refill_fraction: 0.2      # refill when free slots >= this * parallel_jobs
+          # memory_mb_per_cpu: 2000   # CRAB JobType.maxMemoryMB / n_cpus
+    """
+
+    # Re-declare in the class body so law's metaclass sets _defined_workflow_proxy=True
+    # and find_workflow_cls('crab') resolves to *this* class (not law.cms.CrabWorkflow).
+    workflow_proxy_cls = _FLAFCrabWorkflowProxy
+
+    poll_interval = copy_param(law.cms.CrabWorkflow.poll_interval, 5)
+    # When True, law names the worker log ``stdall.txt`` and FLAF stageout_logs.sh
+    # uploads it to fs_default. CRAB itself never transfers this file.
+    transfer_logs = luigi.BoolParameter(
+        default=True,
+        significant=False,
+        description="enable FLAF remote log stageout (stdall.txt via stageout_logs.sh); "
+        "CRAB transferLogs stays off",
+    )
+
+    def _crab_cfg(self):
+        return self.global_params.get("crab") or {}
+
+    def _ensure_crab_pset(self, n_threads):
+        """Write a minimal CRAB PSet with numberOfThreads matching JobType.numCores."""
+        n_threads = max(1, int(n_threads))
+        out_dir = self.local_path()
+        os.makedirs(out_dir, exist_ok=True)
+        path = os.path.join(out_dir, f"crab_PSet_threads{n_threads}.py")
+        content = f"""# Auto-generated by FLAF for CRAB (threads must match JobType.numCores).
+import FWCore.ParameterSet.Config as cms
+
+process = cms.Process("LAW")
+process.source = cms.Source("PoolSource", fileNames=cms.untracked.vstring([""]))
+process.output = cms.OutputModule(
+    "PoolOutputModule", fileName=cms.untracked.string("out.root")
+)
+process.maxEvents = cms.untracked.PSet(input=cms.untracked.int32(1))
+process.options = cms.untracked.PSet(
+    allowUnscheduled=cms.untracked.bool(True),
+    wantSummary=cms.untracked.bool(False),
+    numberOfThreads=cms.untracked.uint32({n_threads}),
+    numberOfStreams=cms.untracked.uint32(0),
+)
+process.out = cms.EndPath(process.output)
+"""
+        if (not os.path.exists(path)) or open(path).read() != content:
+            with open(path, "w") as f:
+                f.write(content)
+        return path
+
+    def crab_stageout_location(self):
+        """Return (storageSite, outLFNDirBase) derived from ``fs_default``.
+
+        FLAF does **not** store analysis outputs here (CRAB transferOutputs is forced
+        off; products go to ``fs_default``). CRAB still requires these fields and runs
+        a submit-time write check against them.
+        """
+        return _crab_stageout_from_fs_spec(self.global_params.get("fs_default"))
+
+    def crab_output_directory(self):
+        return law.LocalDirectoryTarget(self.local_path())
+
+    def crab_request_name(self, submit_jobs):
+        # CRAB: no dots, max 100 characters.
+        import uuid
+
+        parts = [
+            self.task_family.replace(".", "_"),
+            str(self.version).replace(".", "_"),
+            str(self.period).replace(".", "_"),
+            uuid.uuid4().hex[:8],
+        ]
+        name = "_".join(parts)
+        return re.sub(r"[^A-Za-z0-9_\-]", "_", name)[:100]
+
+    def crab_bootstrap_file(self):
+        from law.job.base import JobInputFile
+
+        return JobInputFile(
+            path=os.path.join(self._flaf_root(), "bootstrap.sh"),
+            copy=True,
+            share=True,
+            render_job=True,
+        )
+
+    def crab_stageout_file(self):
+        from law.job.base import JobInputFile
+
+        return JobInputFile(
+            path=os.path.join(self._flaf_root(), "run_tools", "stageout_logs.sh"),
+            copy=True,
+            share=True,
+            render_job=True,
+        )
+
+    def crab_workflow_requires(self):
+        # Always require bundles for CRAB (no AFS on WLCG workers).
+        if not self.bundle_flavours:
+            raise RuntimeError(
+                f"{self.__class__.__name__}: --workflow crab requires bundle_flavours "
+                "on the task (code/environment shipped via BundleTask)"
+            )
+        return self._bundle_requirements()
+
+    def crab_check_job_completeness(self):
+        return False
+
+    def crab_poll_callback(self, poll_data):
+        update_kinit(verbose=0)
+        return True
+
+    def crab_job_file_factory_cls(self):
+        return FLAFCrabJobFileFactory
+
+    def crab_job_file(self):
+        # Same deps_depth=0 patch as HTCondor: avoid huge print_deps on the worker.
+        from law.job.base import JobInputFile
+
+        original = law.util.law_src_path("job", "law_job.sh")
+        custom = os.path.join(
+            os.getenv("ANALYSIS_DATA_PATH"), "law_job_no_print_deps.sh"
+        )
+        if not os.path.exists(custom) or os.path.getmtime(original) > os.path.getmtime(
+            custom
+        ):
+            with open(original) as f:
+                content = f.read()
+            content = re.sub(r'\bdeps_depth="[0-9]+"', 'deps_depth="0"', content)
+            with open(custom, "w") as f:
+                f.write(content)
+            os.chmod(custom, 0o755)
+        return JobInputFile(path=custom, copy=True, share=True, render_job=True)
+
+    def crab_job_config(self, config, job_nums, branches=None):
+        # law 0.1.20 calls crab_job_config(config, list(keys), list(values)); the base
+        # signature documents a single submit_jobs arg, but the call site passes two lists.
+        if not self.bundle_flavours:
+            raise RuntimeError(
+                f"{self.__class__.__name__}: --workflow crab requires bundle_flavours"
+            )
+
+        self._apply_bootstrap_path_render_variables(config)
+        self._apply_bundle_render_variables(config)
+        self._stage_user_custom_input(config)
+        self._stage_path_cache_input(config)
+
+        log_remote_base_url = self._log_remote_base_url()
+        config.render_variables["log_remote_base_url"] = log_remote_base_url
+
+        # Cores + memory. CRAB requires JobType.numCores == PSet numberOfThreads.
+        # Default 2000 MB/CPU (CRAB default; all sites guarantee this per core),
+        # then clamp to the CRAB client max (5000 MB for 1 core, 2500 MB * n_cpus
+        # otherwise).
+        n_cpus = max(1, int(getattr(self, "n_cpus", 1) or 1))
+        try:
+            mb_per_cpu = int(self._crab_cfg().get("memory_mb_per_cpu", 2000))
+        except (TypeError, ValueError):
+            mb_per_cpu = 2000
+        # CRAB client cap: 5000 MB (1 core) or 2500 MB * n_cpus (multi-core).
+        crab_max = 5000 if n_cpus == 1 else 2500 * n_cpus
+        mem = min(n_cpus * max(mb_per_cpu, 1), crab_max)
+        pset_path = self._ensure_crab_pset(n_cpus)
+        config.crab.JobType.psetName = pset_path
+        config.crab.JobType.numCores = n_cpus
+        config.crab.JobType.maxMemoryMB = mem
+
+        # Runtime limit (hours → minutes). CRAB jobs must download/unpack bundles before
+        # the payload starts, so enforce a floor (default 60 min) even when the task's
+        # max_runtime is tiny (e.g. HelloWorld 0.1 h would otherwise be 6 min).
+        max_runtime = getattr(self, "max_runtime", None)
+        if max_runtime is not None and float(max_runtime) > 0:
+            try:
+                cfg_floor = int(self._crab_cfg().get("min_runtime_min", 60))
+                minutes = max(int(math.floor(float(max_runtime) * 60)), cfg_floor)
+                config.crab.JobType.maxJobRuntimeMin = minutes
+            except Exception:
+                # Older CRAB clients may not support maxJobRuntimeMin; ignore if rejected later.
+                pass
+
+        # Law always sets dummy userInputFiles (no inputDataset). The CRAB client
+        # then requires Site.whitelist. Default to every CMS processing site so
+        # analyses need not pin T2_CH_CERN. An explicit crab.whitelist still
+        # restricts; crab.blacklist excludes sites on top of the list used.
+        whitelist = list(self._crab_cfg().get("whitelist") or [])
+        blacklist = list(self._crab_cfg().get("blacklist") or [])
+        if not whitelist:
+            whitelist = ["T1_*", "T2_*", "T3_*"]
+        config.crab.Site.whitelist = [str(s) for s in whitelist]
+        config.crab.Data.ignoreLocality = True
+        if blacklist:
+            config.crab.Site.blacklist = [str(s) for s in blacklist]
+
+        return config
diff --git a/run_tools/stageout_logs.sh b/run_tools/stageout_logs.sh
index a892d1c5..8c3fe8aa 100644
--- a/run_tools/stageout_logs.sh
+++ b/run_tools/stageout_logs.sh
@@ -7,23 +7,36 @@ if [ -z "${log_remote_base_url}" ]; then
     exit 0
 fi
 
-postfix="${LAW_HTCONDOR_JOB_POSTFIX}"
-if [ -n "${postfix}" ]; then
-    log_file="stdall${postfix}.txt"
+# Resolve local log path and remote basename.
+# HTCondor: law may use a postfix / cluster_process name.
+# CRAB: the sandbox log is always stdall.txt in the job dir; include the CRAB job
+# number in the remote name so concurrent jobs do not overwrite each other.
+local_log_file=""
+remote_log_file=""
+
+if [ -n "${LAW_CRAB_JOB_NUMBER:-}" ]; then
+    local_log_file="stdall.txt"
+    remote_log_file="stdall_crab${LAW_CRAB_JOB_NUMBER}.txt"
 else
-    cluster="${LAW_HTCONDOR_JOB_CLUSTER}"
-    process="${LAW_HTCONDOR_JOB_PROCESS}"
-    if [ -n "${cluster}" ] && [ -n "${process}" ]; then
-        log_file="stdall_${cluster}_${process}.txt"
+    postfix="${LAW_HTCONDOR_JOB_POSTFIX}"
+    if [ -n "${postfix}" ]; then
+        local_log_file="stdall${postfix}.txt"
     else
-        log_file="stdall.txt"
+        cluster="${LAW_HTCONDOR_JOB_CLUSTER}"
+        process="${LAW_HTCONDOR_JOB_PROCESS}"
+        if [ -n "${cluster}" ] && [ -n "${process}" ]; then
+            local_log_file="stdall_${cluster}_${process}.txt"
+        else
+            local_log_file="stdall.txt"
+        fi
     fi
+    remote_log_file="${local_log_file}"
 fi
 
 if [ -n "${LAW_JOB_INIT_DIR}" ]; then
-    log_path="${LAW_JOB_INIT_DIR}/${log_file}"
+    log_path="${LAW_JOB_INIT_DIR}/${local_log_file}"
 else
-    log_path="${log_file}"
+    log_path="${local_log_file}"
 fi
 
 if [ ! -f "${log_path}" ]; then
@@ -31,7 +44,7 @@ if [ ! -f "${log_path}" ]; then
     exit 0
 fi
 
-log_remote_url="${log_remote_base_url%/}/${log_file}"
+log_remote_url="${log_remote_base_url%/}/${remote_log_file}"
 
 GFAL_COPY=$(which gfal-copy 2>/dev/null)
 if [ -z "${GFAL_COPY}" ]; then
diff --git a/test/hello_world_task.py b/test/hello_world_task.py
index e1f6045c..06bca493 100644
--- a/test/hello_world_task.py
+++ b/test/hello_world_task.py
@@ -1,11 +1,21 @@
+import json
+import os
+import tempfile
+import traceback
+
 import law
 import luigi
 
-from FLAF.run_tools.law_customizations import Task, HTCondorWorkflow, copy_param
+from FLAF.run_tools.law_customizations import (
+    Task,
+    HTCondorWorkflow,
+    CrabWorkflow,
+    copy_param,
+)
 
 
-class HelloWorldTask(Task, HTCondorWorkflow, law.LocalWorkflow):
-    max_runtime = copy_param(HTCondorWorkflow.max_runtime, 0.1)
+class HelloWorldTask(Task, HTCondorWorkflow, CrabWorkflow, law.LocalWorkflow):
+    max_runtime = copy_param(HTCondorWorkflow.max_runtime, 0.5)
     n_cpus = copy_param(HTCondorWorkflow.n_cpus, 1)
     poll_interval = copy_param(HTCondorWorkflow.poll_interval, 1)
     bundle_flavours = ["core"]
@@ -14,6 +24,22 @@ class HelloWorldTask(Task, HTCondorWorkflow, law.LocalWorkflow):
         significant=False,
         description="raise an exception in run() to test log transfer on crash",
     )
+    # Remote X509 / grid access probe (significant so different probes do not collide).
+    download_url = luigi.Parameter(
+        default="",
+        significant=True,
+        description="if set, probe VOMS/X509 access: gfal_stat/sum and download when small",
+    )
+    max_download_bytes = luigi.IntParameter(
+        default=20 * 1024 * 1024,
+        significant=False,
+        description="full download only if remote size is at most this many bytes",
+    )
+    test_rucio = luigi.BoolParameter(
+        default=False,
+        significant=True,
+        description="if true, attempt Rucio Client() authentication on the worker",
+    )
 
     def create_branch_map(self):
         return {0: "hello"}
@@ -23,12 +49,145 @@ def output(self):
             self.version, self.__class__.__name__, self.period, "hello_world_done.txt"
         )
 
+    def _probe_grid_access(self):
+        """Return a JSON-serializable report of X509/gfal/Rucio status on this host."""
+        from FLAF.RunKit.grid_tools import (
+            get_voms_proxy_info,
+            gfal_stat,
+            gfal_sum,
+            gfal_copy,
+            get_rucio_client,
+            copy_remote_file,
+        )
+
+        report = {
+            "host": os.uname().nodename if hasattr(os, "uname") else "unknown",
+            "cwd": os.getcwd(),
+            "X509_USER_PROXY": os.environ.get("X509_USER_PROXY"),
+            "proxy_file_exists": False,
+            "voms_proxy_info": None,
+            "voms_proxy_error": None,
+            "download_url": self.download_url,
+            "gfal_stat": None,
+            "gfal_stat_error": None,
+            "gfal_sum": None,
+            "gfal_sum_error": None,
+            "download": None,
+            "download_error": None,
+            "rucio": None,
+            "rucio_error": None,
+        }
+
+        proxy = os.environ.get("X509_USER_PROXY", "")
+        report["proxy_file_exists"] = bool(proxy) and os.path.isfile(proxy)
+
+        try:
+            info = get_voms_proxy_info()
+            # keep only plain types
+            report["voms_proxy_info"] = {
+                k: (float(v) if k == "timeleft" else str(v))
+                for k, v in info.items()
+                if k
+                in (
+                    "path",
+                    "timeleft",
+                    "identity",
+                    "issuer",
+                    "type",
+                    "strength",
+                    "VO",
+                )
+            }
+        except Exception as e:
+            report["voms_proxy_error"] = f"{type(e).__name__}: {e}"
+
+        if self.download_url:
+            try:
+                st = gfal_stat(self.download_url)
+                report["gfal_stat"] = {
+                    k: (int(v) if k == "size" else str(v)) for k, v in st.items()
+                }
+            except Exception as e:
+                report["gfal_stat_error"] = (
+                    f"{type(e).__name__}: {e}\n{traceback.format_exc()}"
+                )
+
+            try:
+                asum = gfal_sum(self.download_url, sum_type="adler32")
+                report["gfal_sum"] = {"adler32": str(asum)}
+            except Exception as e:
+                report["gfal_sum_error"] = (
+                    f"{type(e).__name__}: {e}\n{traceback.format_exc()}"
+                )
+
+            size = None
+            if report.get("gfal_stat") and "size" in report["gfal_stat"]:
+                size = int(report["gfal_stat"]["size"])
+            do_full = size is not None and size <= int(self.max_download_bytes)
+            report["download"] = {
+                "attempted": do_full,
+                "reason": (
+                    "size_ok"
+                    if do_full
+                    else (
+                        "size_unknown"
+                        if size is None
+                        else f"size {size} > max_download_bytes {self.max_download_bytes}"
+                    )
+                ),
+                "local_size": None,
+            }
+            if do_full:
+                try:
+                    tmpdir = tempfile.mkdtemp(prefix="hello_dl_")
+                    local = os.path.join(tmpdir, "download.bin")
+                    # Protocol URLs (davs/root/srm): gfal_copy. Pure /store/ LFNs: Rucio path.
+                    if self.download_url.startswith("/store/"):
+                        copy_remote_file(self.download_url, local, verbose=1)
+                    else:
+                        gfal_copy(self.download_url, local, verbose=1)
+                    report["download"]["local_size"] = os.path.getsize(local)
+                    report["download"]["ok"] = True
+                except Exception as e:
+                    report["download"]["ok"] = False
+                    report["download_error"] = (
+                        f"{type(e).__name__}: {e}\n{traceback.format_exc()}"
+                    )
+
+        if self.test_rucio:
+            try:
+                client = get_rucio_client()
+                # Lightweight authenticated call: whoami if available, else ping list.
+                who = None
+                if hasattr(client, "whoami"):
+                    who = client.whoami()
+                report["rucio"] = {
+                    "ok": True,
+                    "whoami": str(who) if who is not None else None,
+                    "account": os.environ.get("RUCIO_ACCOUNT"),
+                }
+            except Exception as e:
+                report["rucio"] = {"ok": False}
+                report["rucio_error"] = (
+                    f"{type(e).__name__}: {e}\n{traceback.format_exc()}"
+                )
+
+        return report
+
     def run(self):
         if self.force_fail:
             raise RuntimeError(
                 f"Forced failure for testing log transfer on crash. version = {self.version}"
             )
         print(f"hello world from {self.version}")
+
+        body = "done\n"
+        if self.download_url or self.test_rucio:
+            report = self._probe_grid_access()
+            print("=== grid access probe report ===")
+            print(json.dumps(report, indent=2, default=str))
+            body = json.dumps(report, indent=2, default=str) + "\n"
+
         with self.output().localize("w") as tmp:
             with open(tmp.path, "w") as f:
-                f.write("done\n")
+                f.write(body)