        )
        validate_s3_destination_prefix_empty(
            _create_session(region, args.profile).client("s3"),
            args.export_destination_s3_uri,
            source_path=source_path,
        )
    target = resolve_headnode_instance_id(cluster_name, region, profile=args.profile)
    wait_for_ssm_online(target.instance_id, region, profile=args.profile, timeout=120)
    validate_headnode_readiness(
        target.instance_id,
        region,
        profile=args.profile,
        timeout=120,
        comment="Validate DAY-EC headnode readiness before workflow launch",
    )

    run_context_content: Optional[str] = None
    samples_content: Optional[str] = None
    units_content: Optional[str] = None
    if args.run_context_file:
        if not args.input_staging:
            raise CommandError("--run-context-file cannot be used with --no-input-staging.")
        if args.stage_dir:
            raise CommandError("--stage-dir cannot be used with --run-context-file.")
        if args.samples_file or args.units_file:
            raise CommandError(
                "--run-context-file cannot be used with --samples-file or --units-file."
            )
        run_context_path = Path(args.run_context_file).expanduser()
        if not run_context_path.is_file():
            raise CommandError(f"Run context file not found: {run_context_path}")
        run_context_content = run_context_path.read_text(encoding="utf-8")
        stage_config = None
    elif args.samples_file or args.units_file:
        if not args.input_staging:
            raise CommandError(
                "--samples-file/--units-file cannot be used with --no-input-staging."
            )
        if args.stage_dir:
            raise CommandError("--stage-dir cannot be used with --samples-file/--units-file.")
        if not args.samples_file or not args.units_file:
            raise CommandError("--samples-file and --units-file must be provided together.")
        samples_path = Path(args.samples_file).expanduser()
        units_path = Path(args.units_file).expanduser()
        if not samples_path.is_file():
            raise CommandError(f"Samples file not found: {samples_path}")
        if not units_path.is_file():
            raise CommandError(f"Units file not found: {units_path}")
        samples_content = samples_path.read_text(encoding="utf-8")
        units_content = units_path.read_text(encoding="utf-8")
        stage_config = None
    elif args.input_staging:
        stage_config = discover_stage_config(
            target.instance_id,
            args.profile,
            region,
            args.stage_dir,
            args.stage_base,
        )
    else:
        if args.stage_dir:
            raise CommandError("--stage-dir cannot be used with --no-input-staging.")
        stage_config = None

    if args.dy_command:
        dy_command = args.dy_command
    else:
        dy_command = build_default_command(
            target=args.target,
            genome=args.genome,
            jobs=args.jobs,
            aligners=args.aligners.split(","),
            dedupers=args.dedupers.split(","),
            snv_callers=args.snv_callers.split(","),
            sv_callers=[value for value in args.sv_callers.split(",") if value],
            containerized=not args.no_containerized,
            dry_run=args.dry_run,
            extra=args.snakemake_extra,
        )

    project_arg = shlex.quote(args.project) if args.project else ""
    repository_literal = json.dumps(args.repository)
    dy_command_literal = shlex.quote(dy_command)
    skip_check = "true" if args.skip_project_check else "false"
    run_context_mode = run_context_content is not None
    sample_config_mode = samples_content is not None or units_content is not None
    run_context_mode_literal = "true" if run_context_mode else "false"
    sample_config_mode_literal = "true" if sample_config_mode else "false"
    input_staging_mode_literal = "true" if args.input_staging else "false"
    default_activation_literal = "true" if args.default_activation else "false"
    bootstrap_test_config_literal = "true" if args.bootstrap_test_config else "false"
    run_context_payload = shlex.quote(run_context_content or "")
    samples_payload = shlex.quote(samples_content or "")
    units_payload = shlex.quote(units_content or "")
    export_destination_literal = shlex.quote(args.export_destination_s3_uri or "")
    delete_on_export_success = "true" if args.delete_on_export_success else "false"
    replace_existing_analysis_dir = "true" if args.replace_existing_analysis_dir else "false"
    if stage_config is None:
        stage_samples_path = ""
        stage_units_path = ""
    else:
        stage_samples_path = stage_config.samples_path
        stage_units_path = stage_config.units_path
    write_status_python = shlex.quote(
        "import json, os, pathlib; "
        "path = pathlib.Path(os.environ['DAYLILY_STATUS_FILE']); "
        "exit_code_raw = os.environ.get('DAYLILY_STATUS_EXIT_CODE', ''); "
        "exit_code = None if exit_code_raw in ('', '__PENDING__') else "
        "(int(exit_code_raw) if exit_code_raw.lstrip('-').isdigit() else exit_code_raw); "
        "payload = dict("
        "session_name=os.environ['DAYLILY_STATUS_SESSION'], "
        "repo_path=os.environ['DAYLILY_STATUS_REPO_PATH'], "
        "started_at=os.environ.get('DAYLILY_STATUS_STARTED_AT') or None, "
        "completed_at=os.environ.get('DAYLILY_STATUS_COMPLETED_AT') or None, "
        "exit_code=exit_code, "
        "command=os.environ['DAYLILY_STATUS_COMMAND']); "
        "path.parent.mkdir(parents=True, exist_ok=True); "
        "path.write_text(json.dumps(payload, indent=2, sort_keys=True) + '\\n', encoding='utf-8')"
    )
    run_context_projection_python = shlex.quote(BCL_RUN_CONTEXT_PROJECTION_SCRIPT)
    bclconvert_profile_patch_python = shlex.quote(BCLCONVERT_PROFILE_PATCH_SCRIPT)
    pipeline_script = f"""
set -euo pipefail
if [[ "$(id -un)" != "ubuntu" ]]; then
  echo "__DAYLILY_ERROR__=wrong_user"
  exit 6
	fi
	SESSION_NAME={shlex.quote(args.session_name)}
	ANALYSIS_ID={shlex.quote(analysis_id)}
	EXECUTING_ENTITY={shlex.quote(executing_entity)}
	RUN_CONTEXT_MODE={run_context_mode_literal}
	SAMPLE_CONFIG_MODE={sample_config_mode_literal}
	INPUT_STAGING_MODE={input_staging_mode_literal}
	DEFAULT_ACTIVATION={default_activation_literal}
	BOOTSTRAP_TEST_CONFIG={bootstrap_test_config_literal}
	RUN_CONTEXT_PAYLOAD={run_context_payload}
	SAMPLES_PAYLOAD={samples_payload}
	UNITS_PAYLOAD={units_payload}
	STAGE_SAMPLES={shlex.quote(stage_samples_path)}
	STAGE_UNITS={shlex.quote(stage_units_path)}
	PROJECT_VALUE={project_arg if project_arg else ""}
	SKIP_PROJECT_CHECK={skip_check}
	DY_COMMAND={dy_command_literal}
	EXPORT_DESTINATION_S3_URI={export_destination_literal}
	EXPORT_TRIGGER={shlex.quote(args.export_trigger)}
	DELETE_ON_EXPORT_SUCCESS={delete_on_export_success}
	REPLACE_EXISTING_ANALYSIS_DIR={replace_existing_analysis_dir}
	ARTIFACT_REGISTRATION_COMMAND_ID={shlex.quote(args.artifact_registration_command_id)}
	DEWEY_URL={shlex.quote(args.dewey_url)}
	DEWEY_TOKEN_ENV={shlex.quote(args.dewey_token_env)}
	DEWEY_ANALYSIS_DIR_EXTERNAL_OBJECT_ID={shlex.quote(args.dewey_analysis_dir_external_object_id)}
	DEWEY_RUN_ARTIFACT_EUID={shlex.quote(args.dewey_run_artifact_euid)}
	DEWEY_URSA_ANALYSIS_EUID={shlex.quote(args.dewey_ursa_analysis_euid)}
STATUS_FILE="${{DAYLILY_RUN_DIR}}/status.json"
TMUX_LOG="${{DAYLILY_TMUX_LOG}}"

write_status() {{
  python3 -c {write_status_python}
}}

export DAYLILY_STATUS_FILE="$STATUS_FILE"
export DAYLILY_STATUS_SESSION="$SESSION_NAME"
export DAYLILY_STATUS_REPO_PATH="${{DAYLILY_REPO_PATH}}"
export DAYLILY_STATUS_COMMAND="$DY_COMMAND"
export DAYLILY_STATUS_STARTED_AT="$(date -u +%Y-%m-%dT%H:%M:%SZ)"
export DAYLILY_STATUS_COMPLETED_AT=""
export DAYLILY_STATUS_EXIT_CODE="__PENDING__"
write_status

trap 'status=$?; if [[ "${{DAYLILY_STATUS_FINALIZED:-0}}" != "1" ]]; then export DAYLILY_STATUS_COMPLETED_AT="$(date -u +%Y-%m-%dT%H:%M:%SZ)"; export DAYLILY_STATUS_EXIT_CODE="$status"; write_status; fi' EXIT

clone_root="$(dirname "${{DAYLILY_REPO_PATH}}")"
repo_path="${{DAYLILY_REPO_PATH}}"
mkdir -p "$(dirname "$clone_root")"

remove_run_dir_projection_links() {{
  local links_dir="$repo_path/config/run_dir_links"
  local entry
  if [[ ! -e "$links_dir" ]]; then
    return 0
  fi
  if [[ ! -d "$links_dir" ]]; then
    echo "[ERROR] Refusing to export with non-directory run projection path: $links_dir"
    return 1
  fi
  while IFS= read -r -d '' entry; do
    if [[ ! -L "$entry" ]]; then
      echo "[ERROR] Refusing to export with non-symlink run projection: $entry"
      return 1
    fi
    echo "[INFO] Removing run-directory projection before export: $entry -> $(readlink "$entry")"
    "        grep -q 'PASS_FAIL' {{output.txt:q}}\\n"
    "        grep -q 'REASON' {{output.txt:q}}"
)
read_haps_strict_precheck_old = (
    "        set +o pipefail\\n"
    "        if gzip -cd {{input.vcf:q}} | grep -m 1 -q -v '^#'; then\\n"
    "            has_variants=true\\n"
    "        else\\n"
    "            has_variants=false\\n"
    "        fi\\n"
    "        set -o pipefail\\n"
    '        if [[ "$has_variants" == "true" ]]; then\\n'
    "            command -v {{params.command:q}} > /dev/null\\n"
    "            test -s {{params.reliable_snp_file:q}}\\n"
    "            set +e\\n"
    "            {{params.command:q}} {{params.extra_args}} -fa {{params.ref:q}} {{input.bam:q}} {{params.reliable_snp_file:q}} {{input.vcf:q}} > {{output.txt:q}} 2> {{log:q}}\\n"
    "            read_haps_rc=$?\\n"
    "            set -e\\n"
    "            if [[ \\\"$read_haps_rc\\\" != \\\"0\\\" ]] || [[ ! -s {{output.txt:q}} ]] || ! grep -q 'PASS_FAIL' {{output.txt:q}} || ! grep -q 'REASON' {{output.txt:q}}; then\\n"
    "                printf 'SNP_PAIRS ERROR_PAIRS DOUBLE_ERROR_PAIR_COUNT DOUBLE_ERROR_FRACTION REL_ERROR_FRACTION NONSENSE_FRACTION PASS_FAIL REASON\\n0 0 0 0 0 0 NO_DATA READ_HAPS_FAILED\\n' > {{output.txt:q}}\\n"
    "                printf 'READ_HAPS_FAILED: read_haps exited with status %s or wrote no usable QC table.\\n' \\\"$read_haps_rc\\\" >> {{log:q}}\\n"
    "            fi\\n"
    "        else\\n"
    "            printf 'SNP_PAIRS ERROR_PAIRS DOUBLE_ERROR_PAIR_COUNT DOUBLE_ERROR_FRACTION REL_ERROR_FRACTION NONSENSE_FRACTION PASS_FAIL REASON\\n0 0 0 0 0 0 NO_DATA NO_VARIANTS\\n' > {{output.txt:q}}\\n"
    "            printf 'NO_VARIANTS: read_haps skipped because the input VCF has no variant records.\\n' > {{log:q}}\\n"
    "        fi\\n"
    "        test -s {{output.txt:q}}\\n"
    "        grep -q 'PASS_FAIL' {{output.txt:q}}\\n"
    "        grep -q 'REASON' {{output.txt:q}}"
)
read_haps_new = (
    "        set +o pipefail\\n"
    "        if gzip -cd {{input.vcf:q}} | grep -m 1 -q -v '^#'; then\\n"
    "            has_variants=true\\n"
    "        else\\n"
    "            has_variants=false\\n"
    "        fi\\n"
    "        set -o pipefail\\n"
    '        if [[ "$has_variants" == "true" ]]; then\\n'
    "            if ! command -v {{params.command:q}} > /dev/null; then\\n"
    "                printf 'SNP_PAIRS ERROR_PAIRS DOUBLE_ERROR_PAIR_COUNT DOUBLE_ERROR_FRACTION REL_ERROR_FRACTION NONSENSE_FRACTION PASS_FAIL REASON\\n0 0 0 0 0 0 NO_DATA READ_HAPS_UNAVAILABLE\\n' > {{output.txt:q}}\\n"
    "                printf 'READ_HAPS_UNAVAILABLE: read_haps command is unavailable: %s\\n' {{params.command:q}} > {{log:q}}\\n"
    "            elif [[ ! -s {{params.reliable_snp_file:q}} ]]; then\\n"
    "                printf 'SNP_PAIRS ERROR_PAIRS DOUBLE_ERROR_PAIR_COUNT DOUBLE_ERROR_FRACTION REL_ERROR_FRACTION NONSENSE_FRACTION PASS_FAIL REASON\\n0 0 0 0 0 0 NO_DATA READ_HAPS_MARKERS_UNAVAILABLE\\n' > {{output.txt:q}}\\n"
    "                printf 'READ_HAPS_MARKERS_UNAVAILABLE: read_haps marker file is missing or empty: %s\\n' {{params.reliable_snp_file:q}} > {{log:q}}\\n"
    "            else\\n"
    "                set +e\\n"
    "                {{params.command:q}} {{params.extra_args}} -fa {{params.ref:q}} {{input.bam:q}} {{params.reliable_snp_file:q}} {{input.vcf:q}} > {{output.txt:q}} 2> {{log:q}}\\n"
    "                read_haps_rc=$?\\n"
    "                set -e\\n"
    "                if [[ \\\"$read_haps_rc\\\" != \\\"0\\\" ]] || [[ ! -s {{output.txt:q}} ]] || ! grep -q 'PASS_FAIL' {{output.txt:q}} || ! grep -q 'REASON' {{output.txt:q}}; then\\n"
    "                printf 'SNP_PAIRS ERROR_PAIRS DOUBLE_ERROR_PAIR_COUNT DOUBLE_ERROR_FRACTION REL_ERROR_FRACTION NONSENSE_FRACTION PASS_FAIL REASON\\n0 0 0 0 0 0 NO_DATA READ_HAPS_FAILED\\n' > {{output.txt:q}}\\n"
    "                    printf 'READ_HAPS_FAILED: read_haps exited with status %s or wrote no usable QC table.\\n' \\\"$read_haps_rc\\\" >> {{log:q}}\\n"
    "                fi\\n"
    "            fi\\n"
    "        else\\n"
    "            printf 'SNP_PAIRS ERROR_PAIRS DOUBLE_ERROR_PAIR_COUNT DOUBLE_ERROR_FRACTION REL_ERROR_FRACTION NONSENSE_FRACTION PASS_FAIL REASON\\n0 0 0 0 0 0 NO_DATA NO_VARIANTS\\n' > {{output.txt:q}}\\n"
    "            printf 'NO_VARIANTS: read_haps skipped because the input VCF has no variant records.\\n' > {{log:q}}\\n"
    "        fi\\n"
    "        test -s {{output.txt:q}}\\n"
    "        grep -q 'PASS_FAIL' {{output.txt:q}}\\n"
    "        grep -q 'REASON' {{output.txt:q}}"
)

text = path.read_text(encoding="utf-8")
changed = False
if hap_new not in text:
    if hap_old not in text:
        raise SystemExit(f"[ERROR] haplocheck zero-variant repair target not found in {{path}}")
    text = text.replace(hap_old, hap_new, 1)
    changed = True
if read_haps_new not in text:
    if read_haps_original_old in text:
        text = text.replace(read_haps_original_old, read_haps_new, 1)
        changed = True
    elif read_haps_partial_old in text:
        text = text.replace(read_haps_partial_old, read_haps_new, 1)
        changed = True
    elif read_haps_empty_failure_old in text:
        text = text.replace(read_haps_empty_failure_old, read_haps_new, 1)
        changed = True
    elif read_haps_strict_precheck_old in text:
        text = text.replace(read_haps_strict_precheck_old, read_haps_new, 1)
        changed = True
    else:
        raise SystemExit(f"[ERROR] read_haps zero-variant repair target not found in {{path}}")
if changed:
    path.write_text(text, encoding="utf-8")
    print(f"[INFO] Patched contamination identity zero-variant repair: {{path}}")
else:
    print(f"[INFO] contamination identity zero-variant repair already present: {{path}}")
PYCONTAMZERO
}}

	BCLCONVERT_PROFILE_PATCH_REQUESTED=false
	if [[ "$RUN_CONTEXT_MODE" == "true" ]]; then
	  printf '%s' "$RUN_CONTEXT_PAYLOAD" > config/runs.tsv
	  project_run_context_mounts
	  materialize_runtime_table samples_table config/samples.tsv
	  materialize_runtime_table units_table config/units.tsv
	  if bclconvert_runtime_tables_requested; then
	    generate_bclconvert_runtime_tables
	    BCLCONVERT_PROFILE_PATCH_REQUESTED=true
	  fi
	  if ultima_run_qc_config_requested; then
	    append_ultima_run_qc_config
	  fi
	elif [[ "$SAMPLE_CONFIG_MODE" == "true" ]]; then
	  printf '%s' "$SAMPLES_PAYLOAD" > config/samples.tsv
	  printf '%s' "$UNITS_PAYLOAD" > config/units.tsv
	elif [[ "$INPUT_STAGING_MODE" == "true" ]]; then
	  cp "$STAGE_SAMPLES" config/samples.tsv
	  cp "$STAGE_UNITS" config/units.tsv
	elif [[ "$BOOTSTRAP_TEST_CONFIG" == "true" ]]; then
	  bootstrap_test_config
	else
	  echo "[INFO] Input staging skipped for this catalog command."
	fi

if [[ ! -f "$HOME/miniconda3/etc/profile.d/conda.sh" ]]; then
  echo "[ERROR] Missing conda profile script at $HOME/miniconda3/etc/profile.d/conda.sh"
  exit 10
fi
. "$HOME/miniconda3/etc/profile.d/conda.sh"
shopt -s expand_aliases
MERMAID_CHROME="$HOME/.cache/puppeteer/chrome/linux-148.0.7778.97/chrome-linux64/chrome"
if [[ -z "${{PUPPETEER_EXECUTABLE_PATH:-}}" && -x "$MERMAID_CHROME" ]]; then
  export PUPPETEER_EXECUTABLE_PATH="$MERMAID_CHROME"
fi

run_dy_command() {{
  local command="$1"
  local dyoainit_source_needed=false
  if [[ "$command" == source\\ dyoainit\\;* ]]; then
    dyoainit_source_needed=true
    command="${{command#source dyoainit;}}"
  elif [[ "$command" == .\\ dyoainit\\;* ]]; then
    dyoainit_source_needed=true
    command="${{command#. dyoainit;}}"
  fi
  command="${{command#"${{command%%[![:space:]]*}}"}}"
  if [[ "$dyoainit_source_needed" == "true" ]]; then
    set +u
    set --
    source dyoainit
    local source_status=$?
    set -u
    if [[ "$source_status" != "0" ]]; then
      return "$source_status"
    fi
  fi
  set +u
  eval "$command"
  local command_status=$?
  set -u
  return "$command_status"
}}

declare -a dyoa_args=()
if [[ -n "$PROJECT_VALUE" ]]; then
  dyoa_args+=(--project {project_arg})
  export PROJECT="$PROJECT_VALUE"
else
  unset PROJECT || true
fi
if [[ "$SKIP_PROJECT_CHECK" == "true" ]]; then
  dyoa_args+=(--skip-project-check)
fi
if [[ "$DEFAULT_ACTIVATION" == "true" ]]; then
  set +u
  . dyoainit "${{dyoa_args[@]}}"
  set -u
  set +e
  set +u
  . bin/day_activate slurm {shlex.quote(args.genome)} remote
  activate_status=$?
  set -u
  set -e
  if [[ "$activate_status" != "0" ]]; then
    echo "[ERROR] day_activate failed with status $activate_status"
    exit "$activate_status"
  fi
fi
if [[ "$BCLCONVERT_PROFILE_PATCH_REQUESTED" == "true" ]]; then
  patch_bclconvert_profile_config
  patch_bclconvert_lane_split
fi
	if ont_run_qc_runtime_repair_requested; then
	  patch_pycoqc_readonly_sort
	fi
	if goleft_indexcov_runtime_repair_requested; then
	  patch_goleft_indexcov_empty_sex_arg
	fi
	if mosdepth_empty_output_runtime_repair_requested; then
	  patch_mosdepth_empty_outputs
	fi
	if rtg_vcfeval_parse_runtime_repair_requested; then
	  patch_rtg_vcfeval_parse_output_dir
	fi
	if vep_zero_variant_runtime_repair_requested; then
	  patch_vep_empty_concat_fofn
	fi
	if contam_identity_zero_variant_runtime_repair_requested; then
	  patch_contam_identity_zero_variant_outputs
	fi
	set +e
	run_dy_command "$DY_COMMAND"
workflow_status=$?
set -e
should_export=false
case "$EXPORT_TRIGGER" in
  none) should_export=false ;;
  on-success) [[ "$workflow_status" -eq 0 ]] && should_export=true ;;
  on-fail) [[ "$workflow_status" -ne 0 ]] && should_export=true ;;
  all) should_export=true ;;
  *) echo "[ERROR] Invalid EXPORT_TRIGGER=$EXPORT_TRIGGER"; workflow_status=20 ;;
esac
if [[ "$should_export" == "true" ]]; then
  if [[ -z "$EXPORT_DESTINATION_S3_URI" ]]; then
    echo "[ERROR] Export requested but EXPORT_DESTINATION_S3_URI is empty"
    workflow_status=21
  else
    if ! remove_run_dir_projection_links; then
      workflow_status=22
    else
      mkdir -p "$DAYLILY_RUN_DIR/export"
      registration_args=()
      if [[ -n "$ARTIFACT_REGISTRATION_COMMAND_ID" ]]; then
        registration_args+=(--artifact-registration-command-id "$ARTIFACT_REGISTRATION_COMMAND_ID")
        registration_args+=(--dewey-url "$DEWEY_URL")
        registration_args+=(--dewey-token-env "$DEWEY_TOKEN_ENV")
        if [[ -n "$DEWEY_ANALYSIS_DIR_EXTERNAL_OBJECT_ID" ]]; then
          registration_args+=(--dewey-analysis-dir-external-object-id "$DEWEY_ANALYSIS_DIR_EXTERNAL_OBJECT_ID")
          registration_args+=(--dewey-run-artifact-euid "$DEWEY_RUN_ARTIFACT_EUID")
          registration_args+=(--dewey-ursa-analysis-euid "$DEWEY_URSA_ANALYSIS_EUID")
        fi
