Fail closed in DAG postprocessing

.travis/test-core-units.sh

@@ -131,6 +131,10 @@ FILES=(
   # -- ILE consolidation precision.  Both DAG builders and BOTH cleaner passes,
   # in the legacy and hyperpipeline formats; 7 tests, 9 s, subprocesses only.
   "$C/test/test_cleanile_intrinsic_precision.py"
+  # -- DAG postprocessing must resolve its sibling helpers without relying on the
+  # submit host's PATH, and must fail closed on helper or empty-output failures.
+  # 6 tests, subprocesses only.
+  "$C/test/test_dag_postprocess_fail_closed.py"
   # -- EOS: --sampler-method portfolio in util_ConstructEOSPosterior.py, which failed on EVERY
   # invocation -- sampler.setup() was never called, so portfolio_breakpoints stayed None and the
   # first draw() raised; without --internal-use-lnL it stopped even earlier, in integrate().
@@ -318,16 +322,18 @@ done
 # runner's closure the count falls back to 347 and still passes.  Pinning 350 would turn an
 # unrelated dependency change into a red gate.
 # Two XML/grid template-finalization regressions, with no added skips.
-EXPECTED_TESTS=590
+# test_dag_postprocess_fail_closed.py adds 6 passing tests and no skips.
+EXPECTED_TESTS=596
 # Outcomes, not just exit status: a collection floor cannot see a test that collects, runs and
 # asserts nothing, and a pytest.skip can quietly absorb a lost gate.  The 13 skips are
 # environment legs -- cupy in test_seeding_reproducibility, device legs in
 # test_dslice_device_native, the nflows leg of
 # test_eos_posterior_tempering_kwarg::test_integrators_read_tempering_exp_from_kwargs
 # (mcsamplerNFlow is an optional dependency and is absent from the IGWN environment), and
 # the xfail in test_uv_symmetry.  test_eos_portfolio_sampler.py adds 12 tests and
-# test_cip_portfolio_members.py 4, none of them skips.
-EXPECTED_PASSED=577
+# test_cip_portfolio_members.py 4, and test_dag_postprocess_fail_closed.py 6;
+# none of them skips.
+EXPECTED_PASSED=583
 MAX_SKIPPED=13
 
 # The floors must be INTEGERS, and this is checked rather than assumed.  `[ 347 -lt FOO ]` does

MonteCarloMarginalizeCode/Code/bin/util_ILEdagPostprocess.sh

@@ -5,6 +5,23 @@
 #   For NR-based DAGs, (a) consolidates the output, (b) runs ILE simplification, then (c) creates an NR-indexed version.
 #   The second format uses a *portable* name, which is stable to me changing the underlying relationship between spins and label.
 
+set -o pipefail
+
+SCRIPT_DIR="$(cd -- "$(dirname -- "${BASH_SOURCE[0]}")" && pwd -P)"
+
+resolve_helper() {
+    local helper=$1
+    if [ -x "${SCRIPT_DIR}/${helper}" ]; then
+        printf '%s\n' "${SCRIPT_DIR}/${helper}"
+    elif command -v "${helper}" >/dev/null 2>&1; then
+        command -v "${helper}"
+    else
+        echo "ERROR: unable to locate required helper ${helper}" >&2
+        return 127
+    fi
+}
+
+CLEAN_ILE="$(resolve_helper util_CleanILE.py)" || exit $?
 
 DIR_PROCESS=$1
 BASE_OUT=$2
@@ -40,10 +57,12 @@ case "$(echo "${RIFT_HYPERPIPELINE_FORMAT:-}" | tr '[:upper:]' '[:lower:]')" in
     # an alias for --digits and ignores the advanced-physics flags it has no use
     # for (its columns are self-describing), but it now REFUSES a leftover value
     # rather than reading it as a shard filename.
-    util_CleanILE_hyperpipeline.py \
+    HYPER_CLEAN="$(resolve_helper util_CleanILE_hyperpipeline.py)" || exit $?
+    "${HYPER_CLEAN}" \
         --output "${BASE_OUT}.composite" \
         "${CLEAN_FLAGS[@]}" \
         ${DIR_PROCESS}/CME*.dat
+    clean_status=$?
     ;;
   *)
     # join together the .dat files
@@ -57,7 +76,14 @@ case "$(echo "${RIFT_HYPERPIPELINE_FORMAT:-}" | tr '[:upper:]' '[:lower:]')" in
 
     # clean them (=join duplicate lines)
     echo " Consolidating multiple instances of the monte carlo  .... "
-    util_CleanILE.py ${RND}_tmp.dat "${CLEAN_FLAGS[@]}" > ${RND}_clean.dat
+    "${CLEAN_ILE}" ${RND}_tmp.dat "${CLEAN_FLAGS[@]}" > ${RND}_clean.dat
+    clean_status=$?
+    if [ ${clean_status} -ne 0 ]; then
+        rm -f ${RND}_clean.dat
+        echo "ERROR: ILE consolidation failed with status ${clean_status}" >&2
+        rm -f "$BASE_OUT.composite"
+        exit ${clean_status}
+    fi
 
     # Sort on lnL.  The composite row is
     #   (event_id, intrinsic..., lnL, sigma_lnL, ntotal, neff)
@@ -72,9 +98,25 @@ case "$(echo "${RIFT_HYPERPIPELINE_FORMAT:-}" | tr '[:upper:]' '[:lower:]')" in
     else
         sort -rg -k$((NCOL-3)) ${RND}_clean.dat > $BASE_OUT.composite
     fi
+    output_status=$?
     rm -f ${RND}_clean.dat
+    if [ ${output_status} -ne 0 ]; then
+        echo "ERROR: failed to write consolidated ILE output with status ${output_status}" >&2
+        rm -f "$BASE_OUT.composite"
+        exit ${output_status}
+    fi
     ;;
 esac
+if [ ${clean_status} -ne 0 ]; then
+    echo "ERROR: ILE consolidation failed with status ${clean_status}" >&2
+    rm -f "$BASE_OUT.composite"
+    exit ${clean_status}
+fi
+if [ ! -s "$BASE_OUT.composite" ]; then
+    echo "ERROR: ILE consolidation produced an empty composite: $BASE_OUT.composite" >&2
+    rm -f "$BASE_OUT.composite"
+    exit 1
+fi
 
 # Manifest
 rm -f ${BASE_OUT}.manifest
@@ -87,6 +129,9 @@ cat ${DIR_PROCESS}/command-single.sh >>  ${BASE_OUT}.manifest
 env >> ${BASE_OUT}.environment  
 
 # tar file
-tar cvzf ${BASE_OUT}.tgz ${BASE_OUT}.composite  ${BASE_OUT}.manifest ${BASE_OUT}.environment
+if ! tar cvzf ${BASE_OUT}.tgz ${BASE_OUT}.composite  ${BASE_OUT}.manifest ${BASE_OUT}.environment; then
+    echo "ERROR: failed to create ${BASE_OUT}.tgz" >&2
+    exit 1
+fi
 
-exit 0 ;  # force end on success, for DAG
+exit 0

MonteCarloMarginalizeCode/Code/bin/util_NRdagPostprocess.sh

@@ -5,6 +5,24 @@
 #   For NR-based DAGs, (a) consolidates the output, (b) runs ILE simplification, then (c) creates an NR-indexed version.
 #   The second format uses a *portable* name, which is stable to me changing the underlying relationship between spins and label.
 
+set -o pipefail
+
+SCRIPT_DIR="$(cd -- "$(dirname -- "${BASH_SOURCE[0]}")" && pwd -P)"
+
+resolve_helper() {
+    local helper=$1
+    if [ -x "${SCRIPT_DIR}/${helper}" ]; then
+        printf '%s\n' "${SCRIPT_DIR}/${helper}"
+    elif command -v "${helper}" >/dev/null 2>&1; then
+        command -v "${helper}"
+    else
+        echo "ERROR: unable to locate required helper ${helper}" >&2
+        return 127
+    fi
+}
+
+CLEAN_ILE="$(resolve_helper util_CleanILE.py)" || exit $?
+RELABEL_ILE="$(resolve_helper util_NRRelabelILE.py)" || exit $?
 
 DIR_PROCESS=$1
 BASE_OUT=$2
@@ -21,9 +39,20 @@ find ${DIR_PROCESS} -name 'CME*.dat' -exec cat {} \; > ${DIR_PROCESS}_tmp.dat
 echo " Consolidating multiple instances of the monte carlo  .... "
 if [ "$4" == '--eccentricity' ]
 then
-    util_CleanILE.py ${DIR_PROCESS}_tmp.dat $4 | sort -rg -k11 > $BASE_OUT.composite
+    "${CLEAN_ILE}" ${DIR_PROCESS}_tmp.dat $4 | sort -rg -k11 > $BASE_OUT.composite
 else
-    util_CleanILE.py ${DIR_PROCESS}_tmp.dat $4 | sort -rg -k10 > $BASE_OUT.composite
+    "${CLEAN_ILE}" ${DIR_PROCESS}_tmp.dat $4 | sort -rg -k10 > $BASE_OUT.composite
+fi
+clean_status=$?
+if [ ${clean_status} -ne 0 ]; then
+    echo "ERROR: NR consolidation failed with status ${clean_status}" >&2
+    rm -f "$BASE_OUT.composite"
+    exit ${clean_status}
+fi
+if [ ! -s "$BASE_OUT.composite" ]; then
+    echo "ERROR: NR consolidation produced an empty composite: $BASE_OUT.composite" >&2
+    rm -f "$BASE_OUT.composite"
+    exit 1
 fi
 
 # index them
@@ -32,9 +61,20 @@ echo " Reindexing the data to   .... "
 if [ "$4" == '--eccentricity' ]
 then
     #    util_NRRelabelILE.py --group ${GROUP} --fname ${BASE_OUT}.composite --eccentricity | grep '^-1*' > ${BASE_OUT}.indexed
-    util_NRRelabelILE.py --group Sequence-RIT-All --fname ${BASE_OUT}.composite --eccentricity | sed -n '/ -----  BEST MATCHES ------ /,$p' > ${BASE_OUT}.indexed
+    "${RELABEL_ILE}" --group Sequence-RIT-All --fname ${BASE_OUT}.composite --eccentricity | sed -n '/ -----  BEST MATCHES ------ /,$p' > ${BASE_OUT}.indexed
 else
-    util_NRRelabelILE.py --group ${GROUP} --fname ${BASE_OUT}.composite | grep '^-1*' > ${BASE_OUT}.indexed
+    "${RELABEL_ILE}" --group ${GROUP} --fname ${BASE_OUT}.composite | grep '^-1*' > ${BASE_OUT}.indexed
+fi
+relabel_status=$?
+if [ ${relabel_status} -ne 0 ]; then
+    echo "ERROR: NR relabeling failed with status ${relabel_status}" >&2
+    rm -f "$BASE_OUT.indexed"
+    exit ${relabel_status}
+fi
+if [ ! -s "$BASE_OUT.indexed" ]; then
+    echo "ERROR: NR relabeling produced an empty index: $BASE_OUT.indexed" >&2
+    rm -f "$BASE_OUT.indexed"
+    exit 1
 fi
 
 # Manifest
@@ -50,4 +90,7 @@ cat ${DIR_PROCESS}/integrate.sub >>  ${BASE_OUT}.submit
 env >> ${BASE_OUT}.environment  
 
 # tar file
-tar cvzf ${BASE_OUT}.tgz ${BASE_OUT}.composite ${BASE_OUT}.indexed ${BASE_OUT}.manifest ${BASE_OUT}.environment ${BASE_OUT}.submit
+if ! tar cvzf ${BASE_OUT}.tgz ${BASE_OUT}.composite ${BASE_OUT}.indexed ${BASE_OUT}.manifest ${BASE_OUT}.environment ${BASE_OUT}.submit; then
+    echo "ERROR: failed to create ${BASE_OUT}.tgz" >&2
+    exit 1
+fi

MonteCarloMarginalizeCode/Code/test/test_dag_postprocess_fail_closed.py

@@ -0,0 +1,115 @@
+import os
+from pathlib import Path
+import shutil
+import subprocess
+
+import pytest
+
+
+BIN_DIR = Path(__file__).resolve().parents[1] / "bin"
+SANITIZED_PATH = "/usr/bin:/bin"
+
+
+def _copy_wrapper(tmp_path, name):
+    wrapper = tmp_path / name
+    shutil.copy2(BIN_DIR / name, wrapper)
+    return wrapper
+
+
+def _write_helper(directory, name, body):
+    helper = directory / name
+    helper.write_text("#!/bin/sh\n" + body)
+    helper.chmod(0o755)
+    return helper
+
+
+def _run_ile(wrapper, tmp_path):
+    input_dir = tmp_path / "ile"
+    input_dir.mkdir()
+    (input_dir / "CME_test.dat").write_text(
+        "0 1 2 3 4 5 6 7 8 9 10 11 12\n"
+    )
+    (input_dir / "test.psd.xml.gz").write_bytes(b"psd")
+    (input_dir / "command-single.sh").write_text("command\n")
+    base_out = tmp_path / "consolidated"
+    env = os.environ.copy()
+    env["PATH"] = SANITIZED_PATH
+    result = subprocess.run(
+        [str(wrapper), str(input_dir), str(base_out)],
+        cwd=tmp_path,
+        env=env,
+        text=True,
+        capture_output=True,
+    )
+    return result, base_out
+
+
+def _run_nr(wrapper, tmp_path):
+    input_dir = tmp_path / "nr"
+    input_dir.mkdir()
+    (input_dir / "CME_test.dat").write_text(
+        "0 1 2 3 4 5 6 7 8 9 10 11 12\n"
+    )
+    (input_dir / "test.psd.xml.gz").write_bytes(b"psd")
+    (input_dir / "command-single.sh").write_text("command\n")
+    (input_dir / "integrate.sub").write_text("queue 1\n")
+    base_out = tmp_path / "consolidated_nr"
+    env = os.environ.copy()
+    env["PATH"] = SANITIZED_PATH
+    result = subprocess.run(
+        [str(wrapper), str(input_dir), str(base_out), "Sequence-RIT-All"],
+        cwd=tmp_path,
+        env=env,
+        text=True,
+        capture_output=True,
+    )
+    return result, base_out
+
+
+@pytest.mark.parametrize("name", ["util_ILEdagPostprocess.sh", "util_NRdagPostprocess.sh"])
+def test_postprocess_fails_when_cleaner_is_unavailable(tmp_path, name):
+    assert shutil.which("util_CleanILE.py", path=SANITIZED_PATH) is None
+    wrapper = _copy_wrapper(tmp_path, name)
+    env = os.environ.copy()
+    env["PATH"] = SANITIZED_PATH
+    args = [str(wrapper), "missing-input", str(tmp_path / "out")]
+    if name == "util_NRdagPostprocess.sh":
+        args.append("Sequence-RIT-All")
+    result = subprocess.run(args, cwd=tmp_path, env=env, text=True, capture_output=True)
+    assert result.returncode != 0
+    assert "unable to locate required helper util_CleanILE.py" in result.stderr
+
+
+def test_ile_postprocess_resolves_sibling_cleaner_with_sanitized_path(tmp_path):
+    wrapper = _copy_wrapper(tmp_path, "util_ILEdagPostprocess.sh")
+    _write_helper(tmp_path, "util_CleanILE.py", 'cat "$1"\n')
+    result, base_out = _run_ile(wrapper, tmp_path)
+    assert result.returncode == 0, result.stderr
+    assert base_out.with_suffix(".composite").stat().st_size > 0
+
+
+def test_ile_postprocess_rejects_empty_composite(tmp_path):
+    wrapper = _copy_wrapper(tmp_path, "util_ILEdagPostprocess.sh")
+    _write_helper(tmp_path, "util_CleanILE.py", "exit 0\n")
+    result, base_out = _run_ile(wrapper, tmp_path)
+    assert result.returncode != 0
+    assert "produced an empty composite" in result.stderr
+    assert not base_out.with_suffix(".composite").exists()
+
+
+def test_ile_postprocess_propagates_cleaner_failure(tmp_path):
+    wrapper = _copy_wrapper(tmp_path, "util_ILEdagPostprocess.sh")
+    _write_helper(tmp_path, "util_CleanILE.py", "exit 42\n")
+    result, base_out = _run_ile(wrapper, tmp_path)
+    assert result.returncode == 42
+    assert not base_out.with_suffix(".composite").exists()
+
+
+def test_nr_postprocess_rejects_empty_relabel_output(tmp_path):
+    wrapper = _copy_wrapper(tmp_path, "util_NRdagPostprocess.sh")
+    _write_helper(tmp_path, "util_CleanILE.py", 'cat "$1"\n')
+    _write_helper(tmp_path, "util_NRRelabelILE.py", "exit 0\n")
+    result, base_out = _run_nr(wrapper, tmp_path)
+    assert result.returncode != 0
+    assert "NR relabeling failed" in result.stderr
+    assert not base_out.with_suffix(".indexed").exists()