diff --git a/jobs/JGLOBAL_STAGE_IC b/jobs/JGLOBAL_STAGE_IC
index b8126f8efe..525e4c5caf 100755
--- a/jobs/JGLOBAL_STAGE_IC
+++ b/jobs/JGLOBAL_STAGE_IC
@@ -3,6 +3,11 @@
 source "${HOMEgfs}/ush/preamble.sh"
 source "${HOMEgfs}/ush/jjob_header.sh" -e "stage_ic" -c "base stage_ic"
 
+# Since ICs will download to DATA if DO_DOWNLOAD_ICS is YES, set ICSDIR to DATA
+if [[ "${DO_DOWNLOAD_ICS}" == "YES" ]]; then
+  export ICSDIR=${DATA}
+fi
+
 # Execute staging
 "${SCRgfs}/exglobal_stage_ic.py"
 err=$?
diff --git a/parm/config/gefs/config.base b/parm/config/gefs/config.base
index f89659d30b..6d6c306c85 100644
--- a/parm/config/gefs/config.base
+++ b/parm/config/gefs/config.base
@@ -338,6 +338,9 @@ export ARCH_CYC=00           # Archive data at this cycle for warm_start capabil
 export ARCH_WARMICFREQ=4     # Archive frequency in days for warm_start capability
 export ARCH_FCSTICFREQ=1     # Archive frequency in days for gdas and gfs forecast-only capability
 
+export DO_DOWNLOAD_ICS="YES" # Set to YES to download ICs from HPSS, set to NO to copy ICs locally
+export HPSSICARCH="/NCEPDEV/emc-marine/2year/Neil.Barton/ICS/HR4/C384mx025" # ICs directory on HPSS
+
 export DELETE_COM_IN_ARCHIVE_JOB="YES"   # NO=retain ROTDIR.  YES default in arch.sh and earc.sh.
 
 # Number of regional collectives to create soundings for
diff --git a/parm/config/gefs/config.resources b/parm/config/gefs/config.resources
index 6fc8a0eec5..5baa3a774b 100644
--- a/parm/config/gefs/config.resources
+++ b/parm/config/gefs/config.resources
@@ -61,11 +61,11 @@ export max_tasks_per_node
 case ${step} in
 
   "stage_ic")
-    export walltime="00:15:00"
+    export walltime="06:00:00"
     export ntasks=1
     export tasks_per_node=1
     export threads_per_task=1
-    export is_exclusive=True
+    export memory="4096M"
     ;;
 
   "waveinit")
diff --git a/scripts/exglobal_stage_ic.py b/scripts/exglobal_stage_ic.py
index d737d83b47..25a09f94ac 100755
--- a/scripts/exglobal_stage_ic.py
+++ b/scripts/exglobal_stage_ic.py
@@ -22,7 +22,8 @@ def main():
             'assim_freq', 'current_cycle', 'previous_cycle',
             'ROTDIR', 'ICSDIR', 'STAGE_IC_YAML_TMPL', 'DO_JEDIATMVAR',
             'OCNRES', 'waveGRD', 'ntiles', 'DOIAU', 'DO_JEDIOCNVAR',
-            'REPLAY_ICS', 'DO_WAVE', 'DO_OCN', 'DO_ICE', 'DO_NEST']
+            'REPLAY_ICS', 'DO_WAVE', 'DO_OCN', 'DO_ICE', 'DO_NEST',
+            'HPSSICARCH', 'DO_DOWNLOAD_ICS']
 
     stage_dict = AttrDict()
     for key in keys:
diff --git a/ush/python/pygfs/task/stage_ic.py b/ush/python/pygfs/task/stage_ic.py
index 37cc4138f3..01744d8842 100644
--- a/ush/python/pygfs/task/stage_ic.py
+++ b/ush/python/pygfs/task/stage_ic.py
@@ -6,8 +6,9 @@
 from typing import Any, Dict, List
 
 from wxflow import (AttrDict, FileHandler, Task, cast_strdict_as_dtypedict,
-                    logit, parse_j2yaml, strftime, to_YMD,
-                    add_to_datetime, to_timedelta, Template, TemplateConstants)
+                    logit, parse_j2yaml, strftime, to_YMD, to_YMDH,
+                    add_to_datetime, to_timedelta, Template, TemplateConstants,
+                    Hsi, Htar)
 
 logger = getLogger(__name__.split('.')[-1])
 
@@ -47,6 +48,14 @@ def execute_stage(self, stage_dict: Dict[str, Any]) -> None:
         None
         """
 
+        if stage_dict.DO_DOWNLOAD_ICS == True:
+            # Download ICs from HPSS to ICSDIR
+            self.tar_cmd = "htar"
+            self.hsi = Hsi()
+            self.htar = Htar()
+            self.xvf = self.htar.xvf
+            self.xvf(stage_dict.HPSSICARCH+"/"+to_YMDH(stage_dict.current_cycle)+".tar")
+
         if not os.path.isdir(stage_dict.ROTDIR):
             raise FileNotFoundError(f"FATAL ERROR: The ROTDIR ({stage_dict.ROTDIR}) does not exist!")
 
diff --git a/workflow/rocoto/tasks.py b/workflow/rocoto/tasks.py
index 92ceea73aa..1dfc443187 100644
--- a/workflow/rocoto/tasks.py
+++ b/workflow/rocoto/tasks.py
@@ -10,8 +10,8 @@
 
 
 class Tasks:
-    SERVICE_TASKS = ['arch', 'earc']
-    VALID_TASKS = ['aerosol_init', 'stage_ic',
+    SERVICE_TASKS = ['stage_ic', 'arch', 'earc']
+    VALID_TASKS = ['aerosol_init',
                    'prep', 'anal', 'sfcanl', 'analcalc', 'analdiag', 'arch', "cleanup",
                    'prepatmiodaobs', 'atmanlinit', 'atmanlvar', 'atmanlfv3inc', 'atmanlfinal',
                    'prepoceanobs',