diff --git a/apex/core/calculator/Lammps.py b/apex/core/calculator/Lammps.py
index 224a9050..661607e5 100644
--- a/apex/core/calculator/Lammps.py
+++ b/apex/core/calculator/Lammps.py
@@ -446,6 +446,15 @@ def make_input_file(self, output_dir, task_type, task_param):
                 fp.write(fc)
 
     def compute(self, output_dir):
+        task_json = os.path.join(output_dir, "task.json")
+        try:
+            task_param = loadfn(task_json) if os.path.isfile(task_json) else {}
+        except Exception:
+            task_param = {}
+        task_type = task_param.get("type", task_param.get("cal_type"))
+        if task_type in ["annealing", "Annealing"]:
+            return None
+
         log_lammps = os.path.join(output_dir, "log.lammps")
         dump_lammps = os.path.join(output_dir, "dump.relax")
         if not os.path.isfile(log_lammps) or not os.path.isfile(dump_lammps):
@@ -724,12 +733,15 @@ def backward_files(self, property_type="relaxation"):
                 "log.lammps",
                 "outlog",
                 *debug_files,
-                "dump.anneal_ramp",
-                "dump.anneal_cool",
-                "heating_interval.dat",
-                "cooling_interval.dat",
-                "rdf_ramp.dat",
-                "rdf_cool.dat",
+                "dump.init.*",
+                "dump.eq_*",
+                "dump.T_ramp_*",
+                "dump.T_decline_*",
+                "dump.final_eq_*",
+                "heating_interval_*.dat",
+                "cooling_interval_*.dat",
+                "rdf.*.txt",
+                "msd.*.txt",
                 "restart.*",
             ]
         elif property_type == "finite_t_elastic":
diff --git a/apex/core/calculator/lib/lammps_utils.py b/apex/core/calculator/lib/lammps_utils.py
index ca3e12d2..4cadf8de 100644
--- a/apex/core/calculator/lib/lammps_utils.py
+++ b/apex/core/calculator/lib/lammps_utils.py
@@ -690,7 +690,7 @@ def make_lammps_press_relax(
     return ret
 
 def make_lammps_annealing(conf, type_map, interaction, param, cal_setting):
-    """LAMMPS input for annealing: equilibrate -> heat (ramp) -> optional hold -> cool.
+    """LAMMPS input for annealing using the same stage controls as annealing/.
 
     Uses variables provided by `variable_Annealing.in` in the task directory.
     - thermostat: nose_hoover | langevin
@@ -698,12 +698,77 @@ def make_lammps_annealing(conf, type_map, interaction, param, cal_setting):
     """
 
     type_map_list = element_list(type_map)
-    dump_step = int(cal_setting.get("dump_step", 1000))
-    tdamp = cal_setting.get("tdamp", 100)
-    pdamp = cal_setting.get("pdamp", 1000)
+    dump_interval = int(cal_setting.get("dump_interval", cal_setting.get("dump_step", 2000)))
+    thermo_interval = int(cal_setting.get("thermo_interval", 2000))
+    restart_interval = int(cal_setting.get("restart_interval", 20000))
+    tdamp = cal_setting.get("tdamp", "${tdamp}")
+    pdamp = cal_setting.get("pdamp", "${pdamp}")
     thermostat = cal_setting.get("thermostat", "nose_hoover")
     ensemble = cal_setting.get("ensemble", "npt")
-    vseed = int(cal_setting.get("velocity_seed", 12345))
+    vseed = int(cal_setting.get("velocity_seed", cal_setting.get("init_v_seed", 123457)))
+    lgv_seed = int(cal_setting.get("lgv_seed", vseed))
+    req_lgv_damping = cal_setting.get("req_lgv_damping", False)
+    req_compute_rdf = cal_setting.get("req_compute_rdf", True)
+    req_compute_msd = cal_setting.get("req_compute_msd", True)
+    req_write_restart = cal_setting.get("req_write_restart", True)
+    req_dump_init_atom = cal_setting.get("req_dump_init_atom", True)
+    req_dump_ave_atom = cal_setting.get("req_dump_ave_atom", False)
+    rdf_nevery = int(cal_setting.get("rdf_nevery", cal_setting.get("rdf_interval", 100)))
+    rdf_nrepeat = int(cal_setting.get("rdf_nrepeat", 1))
+    rdf_nfreq = int(cal_setting.get("rdf_nfreq", cal_setting.get("rdf_interval", 200)))
+    msd_nevery = int(cal_setting.get("msd_nevery", 100))
+    msd_nrepeat = int(cal_setting.get("msd_nrepeat", 1))
+    msd_nfreq = int(cal_setting.get("msd_nfreq", 200))
+
+    def _truthy(value):
+        if isinstance(value, bool):
+            return value
+        if isinstance(value, str):
+            return value.strip().lower() in {"1", "true", "yes", "on"}
+        return bool(value)
+
+    def _thermo_fix(name, t_start, t_stop):
+        if thermostat == "langevin":
+            if ensemble == "nve":
+                fix = f"fix {name}_int all nve\n"
+            else:
+                fix = f"fix {name}_int all nph aniso 0.0 0.0 {pdamp} drag 1.0\n"
+            fix += f"fix {name}_lgv all langevin {t_start} {t_stop} {tdamp} {lgv_seed}\n"
+            return fix
+        if ensemble == "nvt":
+            return f"fix {name}_nh all nvt temp {t_start} {t_stop} {tdamp}\n"
+        return (
+            f"fix {name}_nh all npt temp {t_start} {t_stop} {tdamp} "
+            f"x 0.0 0.0 {pdamp} y 0.0 0.0 {pdamp} z 0.0 0.0 {pdamp}\n"
+        )
+
+    def _unfix_thermo(name):
+        if thermostat == "langevin":
+            return f"unfix {name}_lgv\nunfix {name}_int\n"
+        return f"unfix {name}_nh\n"
+
+    def _stage_analysis(stage, rdf_file, msd_file):
+        ret = ""
+        if _truthy(req_compute_rdf):
+            ret += (
+                f"fix rdf_{stage} all ave/time {rdf_nevery} {rdf_nrepeat} {rdf_nfreq} "
+                f"c_myRDF[*] file {rdf_file} mode vector\n"
+            )
+        if _truthy(req_compute_msd):
+            ret += f"compute myMSD_{stage} all msd com yes\n"
+            ret += (
+                f"fix msd_{stage} all ave/time {msd_nevery} {msd_nrepeat} {msd_nfreq} "
+                f"c_myMSD_{stage} file {msd_file} mode vector\n"
+            )
+        return ret
+
+    def _unfix_stage_analysis(stage):
+        ret = ""
+        if _truthy(req_compute_rdf):
+            ret += f"unfix rdf_{stage}\n"
+        if _truthy(req_compute_msd):
+            ret += f"unfix msd_{stage}\nuncompute myMSD_{stage}\n"
+        return ret
 
     ret = ""
     ret += "include  variable_Annealing.in\n"
@@ -720,8 +785,9 @@ def make_lammps_annealing(conf, type_map, interaction, param, cal_setting):
     ret += "neigh_modify    every 1 delay 0 check no\n"
     ret += interaction(param)
     ret += "compute         mype all pe\n"
-    ret += "thermo          100\n"
-    ret += ("thermo_style    custom step temp pe pxx pyy pzz pxy pxz pyz lx ly lz vol c_mype\n")
+    ret += f"thermo          {thermo_interval}\n"
+    ret += ("thermo_style    custom step temp atoms epair pe ke etotal press vol pxx pyy pzz pxy pxz pyz lx ly lz xlo xhi ylo yhi zlo zhi fnorm fmax\n")
+    ret += "thermo_modify   lost error flush yes format 4 %.8f\n"
     ret += "timestep ${timestep}\n"
     ret += "variable        N equal count(all)\n"
     ret += "variable        V equal vol\n"
@@ -731,80 +797,99 @@ def make_lammps_annealing(conf, type_map, interaction, param, cal_setting):
     ret += "variable        Etotal equal etotal\n"
     ret += "variable        Press equal press\n"
     ret += "variable        stepVal equal step\n"
-    ret += "compute         myRDF all rdf ${rdf_bins} cutoff ${rdf_cutoff}\n"
+    if _truthy(req_compute_rdf):
+        ret += "variable        rdf_comm_cutoff equal ${rdf_cutoff}+2.0\n"
+        ret += "comm_modify    cutoff ${rdf_comm_cutoff}\n"
+        ret += "compute         myRDF all rdf ${rdf_bins} cutoff ${rdf_cutoff}\n"
+
+    if _truthy(req_dump_init_atom):
+        ret += "dump init_dump all atom 1 dump.init.*\n"
+        ret += "run 0\n"
+        ret += "undump init_dump\n"
 
     # Initialize velocities and equilibrate at start_temp
     ret += f"velocity all create ${{start_temp}} {vseed} mom yes rot yes dist gaussian\n"
 
-    if thermostat == "langevin":
-        # Langevin + barostat (nph) or without (nve)
-        if ensemble == "nve":
-            ret += "fix 1 all nve\n"
-        else:
-            ret += f"fix 1 all nph aniso 0.0 0.0 {pdamp} drag 1.0\n"
-        ret += f"fix tg all langevin ${{start_temp}} ${{start_temp}} {tdamp} {vseed}\n"
-    else:
-        # Nose-Hoover NPT or NVT
-        if ensemble == "nvt":
-            ret += f"fix 1 all nvt temp ${{start_temp}} ${{start_temp}} {tdamp}\n"
-        else:
-            ret += f"fix 1 all npt temp ${{start_temp}} ${{start_temp}} {tdamp} x 0.0 0.0 {pdamp} y 0.0 0.0 {pdamp} z 0.0 0.0 {pdamp}\n"
-
-    ret += "run ${equi_step}\n"
-    ret += "unfix 1\n"
-    if thermostat == "langevin":
-        ret += "unfix tg\n"
+    if _truthy(req_lgv_damping):
+        ret += "fix eq_lgv_int all nve\n"
+        ret += f"fix eq_lgv all langevin ${{start_temp}} ${{start_temp}} {tdamp} {lgv_seed}\n"
+        ret += f"dump eq_lgv_dump all atom {dump_interval} dump.eq_lgv_${{start_temp}}K.*\n"
+        if _truthy(req_write_restart):
+            ret += f"restart {restart_interval} restart.eq_lgv.*\n"
+        ret += "run ${init_lgv_thermo_equil_step}\n"
+        ret += "restart 0\n"
+        ret += "undump eq_lgv_dump\n"
+        ret += "unfix eq_lgv\n"
+        ret += "unfix eq_lgv_int\n"
+        ret += "write_restart restart.eq_lgv_final\n"
+
+    ret += _stage_analysis("eq", "rdf.eq_${start_temp}K.txt", "msd.eq_${start_temp}K.txt")
+    ret += _thermo_fix("eq", "${start_temp}", "${start_temp}")
+    ret += f"dump eq_nh_dump all atom {dump_interval} dump.eq_nh_${{start_temp}}K.*\n"
+    if _truthy(req_dump_ave_atom):
+        ret += "compute atom_u_pos all property/atom xu yu zu\n"
+        ret += "fix ave_atom_u_pos all ave/atom ${ave_atom_sample_feq} ${ave_atom_sample_length} ${dump_interval} c_atom_u_pos[*]\n"
+        ret += f"dump eq_nh_ave_dump all custom {dump_interval} dump.eq_nh_${{start_temp}}K_ave.* id type f_ave_atom_u_pos[1] f_ave_atom_u_pos[2] f_ave_atom_u_pos[3]\n"
+    if _truthy(req_write_restart):
+        ret += f"restart {restart_interval} restart.eq_nh.*\n"
+    ret += "run ${init_thermo_equil_step}\n"
+    ret += "restart 0\n"
+    ret += "undump eq_nh_dump\n"
+    if _truthy(req_dump_ave_atom):
+        ret += "undump eq_nh_ave_dump\nunfix ave_atom_u_pos\nuncompute atom_u_pos\n"
+    ret += _unfix_thermo("eq")
+    ret += _unfix_stage_analysis("eq")
+    ret += "reset_timestep 0\n"
+    ret += "write_restart restart.eq_final\n"
 
     # Temperature ramp to target_temp
-    if thermostat == "langevin":
-        if ensemble == "nve":
-            ret += "fix 1 all nve\n"
-        else:
-            ret += f"fix 1 all nph aniso 0.0 0.0 {pdamp} drag 1.0\n"
-        ret += f"fix tg all langevin ${{start_temp}} ${{target_temp}} {tdamp} {vseed}\n"
-    else:
-        if ensemble == "nvt":
-            ret += f"fix 1 all nvt temp ${{start_temp}} ${{target_temp}} {tdamp}\n"
-        else:
-            ret += f"fix 1 all npt temp ${{start_temp}} ${{target_temp}} {tdamp} x 0.0 0.0 {pdamp} y 0.0 0.0 {pdamp} z 0.0 0.0 {pdamp}\n"
-    ret += f"dump            1 all custom  {dump_step} dump.anneal_ramp id type xs ys zs fx fy fz\n"
-    ret += "fix rdf_ramp all ave/time ${rdf_interval} 1 ${rdf_interval} c_myRDF[*] file rdf_ramp.dat mode vector\n"
-    ret += "fix heat_log all ave/time ${rdf_interval} 1 ${rdf_interval} v_stepVal v_N v_Temp v_Vatom v_pote v_Etotal v_Press file heating_interval.dat\n"
-    ret += "run ${ramp_step}\n"
+    ret += _stage_analysis("ramp", "rdf.T_ramp_${start_temp}K_${temp}K.txt", "msd.T_ramp_${start_temp}K_${temp}K.txt")
+    ret += _thermo_fix("ramp", "${start_temp}", "${temp}")
+    ret += f"dump T_ramp_nh_dump all atom {dump_interval} dump.T_ramp_nh_${{start_temp}}K_${{temp}}K.*\n"
+    ret += "fix heat_log all ave/time 1 100 ${thermo_interval} v_N v_Temp v_Vatom v_pote v_Etotal v_Press file heating_interval_${thermo_interval}.dat\n"
+    if _truthy(req_write_restart):
+        ret += f"restart {restart_interval} restart.T_ramp_nh.*\n"
+    ret += "run ${temp_ramp_remain_step}\n"
+    ret += "restart 0\n"
     ret += "unfix heat_log\n"
-    ret += "unfix rdf_ramp\n"
-    ret += "undump 1\n"
-    ret += "unfix 1\n"
-    if thermostat == "langevin":
-        ret += "unfix tg\n"
-
-        # Optional hold at target_temp
-    ret += f'if "${{hold_step}} > 0" then "fix 1 all nvt temp ${{target_temp}} ${{target_temp}} {tdamp}" "run ${{hold_step}}" "unfix 1"\n'
+    ret += "undump T_ramp_nh_dump\n"
+    ret += _unfix_thermo("ramp")
+    ret += _unfix_stage_analysis("ramp")
+    ret += "reset_timestep 0\n"
+    ret += "write_restart restart.T_ramp_nh_final\n"
 
     # Cool to end_temp
-    if thermostat == "langevin":
-        if ensemble == "nve":
-            ret += "fix 1 all nve\n"
-        else:
-            ret += f"fix 1 all nph aniso 0.0 0.0 {pdamp} drag 1.0\n"
-        ret += f"fix tg all langevin ${{target_temp}} ${{end_temp}} {tdamp} {vseed}\n"
-    else:
-        if ensemble == "nvt":
-            ret += f"fix 1 all nvt temp ${{target_temp}} ${{end_temp}} {tdamp}\n"
-        else:
-            ret += f"fix 1 all npt temp ${{target_temp}} ${{end_temp}} {tdamp} x 0.0 0.0 {pdamp} y 0.0 0.0 {pdamp} z 0.0 0.0 {pdamp}\n"
-    ret += f"dump            2 all custom  {dump_step} dump.anneal_cool id type xs ys zs fx fy fz\n"
-    ret += "fix rdf_cool all ave/time ${rdf_interval} 1 ${rdf_interval} c_myRDF[*] file rdf_cool.dat mode vector\n"
-    ret += "fix cool_log all ave/time ${rdf_interval} 1 ${rdf_interval} v_stepVal v_N v_Temp v_Vatom v_pote v_Etotal v_Press file cooling_interval.dat\n"
-    ret += "run ${cool_step}\n"
+    ret += _stage_analysis("decline", "rdf.T_decline_${temp}K_${end_temp}K.txt", "msd.T_decline_${temp}K_${end_temp}K.txt")
+    ret += _thermo_fix("decline", "${temp}", "${end_temp}")
+    ret += f"dump T_decline_nh_dump all atom {dump_interval} dump.T_decline_nh_${{temp}}K_${{end_temp}}K.*\n"
+    ret += "fix cool_log all ave/time 1 100 ${thermo_interval} v_N v_Temp v_Vatom v_pote v_Etotal v_Press file cooling_interval_${thermo_interval}.dat\n"
+    if _truthy(req_write_restart):
+        ret += f"restart {restart_interval} restart.T_decline_nh.*\n"
+    ret += "run ${temp_decline_remain_step}\n"
+    ret += "restart 0\n"
     ret += "unfix cool_log\n"
-    ret += "unfix rdf_cool\n"
-    ret += "undump 2\n"
-    ret += "unfix 1\n"
-    if thermostat == "langevin":
-        ret += "unfix tg\n"
-
-    ret += 'print "All done"\n'
+    ret += "undump T_decline_nh_dump\n"
+    ret += _unfix_thermo("decline")
+    ret += _unfix_stage_analysis("decline")
+    ret += "reset_timestep 0\n"
+    ret += "write_restart restart.T_decline_final\n"
+
+    ret += 'if "${final_thermo_equil_step} <= 0" then "jump SELF end_of_run"\n'
+    ret += _stage_analysis("final_eq", "rdf.final_eq_${end_temp}K.txt", "msd.final_eq_${end_temp}K.txt")
+    ret += _thermo_fix("final_eq", "${end_temp}", "${end_temp}")
+    ret += f"dump final_eq_nh_dump all atom {dump_interval} dump.final_eq_nh_${{end_temp}}K.*\n"
+    if _truthy(req_write_restart):
+        ret += f"restart {restart_interval} restart.final_eq_nh.*\n"
+    ret += "run ${final_thermo_equil_remain_step}\n"
+    ret += "restart 0\n"
+    ret += "undump final_eq_nh_dump\n"
+    ret += _unfix_thermo("final_eq")
+    ret += _unfix_stage_analysis("final_eq")
+    ret += "reset_timestep 0\n"
+    ret += "write_restart restart.final_eq_final\n"
+
+    ret += 'print "__end_of_lmp_annealing_calculation__"\n'
+    ret += 'label end_of_run\n'
     return ret
 
 """
diff --git a/apex/core/property/Annealing.py b/apex/core/property/Annealing.py
index 2082f33b..bf87d415 100644
--- a/apex/core/property/Annealing.py
+++ b/apex/core/property/Annealing.py
@@ -1,14 +1,23 @@
+import glob
 import os
 import logging
 from typing import List, Dict, Any
 
-from monty.serialization import dumpfn
+from monty.serialization import dumpfn, loadfn
 from pymatgen.core.structure import Structure
 
 from apex.core.property.Property import Property
 from apex.core.calculator.lib import vasp_utils
 
 
+def _as_bool(value) -> bool:
+    if isinstance(value, bool):
+        return value
+    if isinstance(value, str):
+        return value.strip().lower() in {"1", "true", "yes", "on"}
+    return bool(value)
+
+
 class Annealing(Property):
 
     def __init__(self, parameter: Dict[str, Any], inter_param=None):
@@ -22,32 +31,64 @@ def __init__(self, parameter: Dict[str, Any], inter_param=None):
 
         # MD controls (independent knobs only)
         cal = parameter.setdefault("cal_setting", {})
-        # required temps
-        self.start_temp = float(cal.get("start_temp", 300))
-        _tgt = cal.get("target_temp", 800)
+        # Schedule defaults mirror annealing/spec.
+        self.start_temp = float(cal.get("start_temp", 4))
+        _tgt = cal.get("target_temp", cal.get("temp", 300))
         self.target_temp = float(_tgt if not isinstance(_tgt, list) else _tgt[0])
-        self.end_temp = float(cal.get("end_temp", 300))
-        # required steps
-        self.equi_step = int(cal.get("equi_step", 10000))
-        # Allow specifying ramp/cool by rate (K/step); derive steps per task if provided
-        self.ramp_rate = cal.pop("ramp_rate", None)
-        self.cool_rate = cal.pop("cool_rate", None)
-        self.ramp_step = int(cal.get("ramp_step", 20000))
-        self.hold_step = int(cal.get("hold_step", 0))
-        self.cool_step = int(cal.get("cool_step", 20000))
+        self.end_temp = float(cal.get("end_temp", 4))
+        self._has_ramp_rate = "temp_ramp_rate" in cal or "ramp_rate" in cal
+        self._has_cool_rate = "cool_rate" in cal
+        self.temp_ramp_rate = cal.get("temp_ramp_rate", cal.get("ramp_rate", 1000))
+        self.cool_rate = cal.get("cool_rate", self.temp_ramp_rate)
+        self.equi_step = int(cal.get("equi_step", cal.get("init_thermo_equil_step", 20000)))
+        self.init_lgv_thermo_equil_step = int(cal.get("init_lgv_thermo_equil_step", 20000))
+        self.init_thermo_equil_step = int(cal.get("init_thermo_equil_step", self.equi_step))
+        self.final_thermo_equil_step = int(cal.get("final_thermo_equil_step", cal.get("hold_step", 20000)))
+        # Explicit step counts override rate-derived counts when provided.
+        self.ramp_step = int(cal.get("ramp_step", cal.get("temp_ramp_step", 0)))
+        self.cool_step = int(cal.get("cool_step", cal.get("temp_decline_step", 0)))
+        self.hold_step = int(cal.get("hold_step", self.final_thermo_equil_step))
         # options
         self.thermostat = cal.get("thermostat", "nose_hoover")
         self.ensemble = cal.get("ensemble", "npt")
-        self.tdamp = cal.get("tdamp", 100)
-        self.pdamp = cal.get("pdamp", 1000)
-        self.velocity_seed = cal.get("velocity_seed", 12345)
-        self.dump_step = int(cal.get("dump_step", 1000))
-        # timestep (ps, units metal); default 0.002 ps (2 fs)
-        self.timestep = float(cal.get("timestep", 0.002))
+        self.timestep = float(cal.get("timestep", 0.001))
+        self.tdamp_factor = cal.get("tdamp_factor", 100)
+        self.pdamp_factor = cal.get("pdamp_factor", 1000)
+        self.tdamp = cal.get("tdamp")
+        self.pdamp = cal.get("pdamp")
+        self.velocity_seed = cal.get("velocity_seed", cal.get("init_v_seed", 123457))
+        self.lgv_seed = cal.get("lgv_seed", self.velocity_seed)
+        self.req_lgv_damping = _as_bool(cal.get("req_lgv_damping", False))
+        self.req_opti_init_structure = _as_bool(cal.get("req_opti_init_structure", True))
+        self.req_write_restart = _as_bool(cal.get("req_write_restart", True))
+        self.req_dump_init_atom = _as_bool(cal.get("req_dump_init_atom", True))
+        self.req_dump_ave_atom = _as_bool(cal.get("req_dump_ave_atom", False))
+        self.dump_step = int(cal.get("dump_step", cal.get("dump_interval", 2000)))
+        self.dump_interval = int(cal.get("dump_interval", self.dump_step))
+        self.thermo_interval = int(cal.get("thermo_interval", 2000))
+        self.restart_interval = int(cal.get("restart_interval", 20000))
+        self.ave_atom_sample_feq = int(cal.get("ave_atom_sample_feq", 1))
+        self.ave_atom_num_sample = int(cal.get("ave_atom_num_sample", self.dump_interval))
+        self.ave_atom_sample_length = int(cal.get(
+            "ave_atom_sample_length",
+            self.ave_atom_sample_feq * self.ave_atom_num_sample,
+        ))
+        self.init_opt_loop_size = int(cal.get("init_opt_loop_size", 10))
+        self.init_fmax_tol = cal.get("init_fmax_tol", 1.0e-8)
+        self.init_stress_tol = cal.get("init_stress_tol", 1.0e-2)
         # RDF settings
-        self.rdf_bins = int(cal.get("rdf_bins", 200))
-        self.rdf_cutoff = float(cal.get("rdf_cutoff", 10.0))
-        self.rdf_interval = int(cal.get("rdf_interval", 100))
+        self.req_compute_rdf = _as_bool(cal.get("req_compute_rdf", True))
+        self.rdf_bins = int(cal.get("rdf_bins", 100))
+        self.rdf_cutoff = float(cal.get("rdf_cutoff", 6.0))
+        self.rdf_nevery = int(cal.get("rdf_nevery", cal.get("rdf_interval", 100)))
+        self.rdf_nrepeat = int(cal.get("rdf_nrepeat", 1))
+        self.rdf_nfreq = int(cal.get("rdf_nfreq", cal.get("rdf_interval", 200)))
+        self.rdf_interval = int(cal.get("rdf_interval", self.rdf_nfreq))
+        # MSD settings
+        self.req_compute_msd = _as_bool(cal.get("req_compute_msd", True))
+        self.msd_nevery = int(cal.get("msd_nevery", 100))
+        self.msd_nrepeat = int(cal.get("msd_nrepeat", 1))
+        self.msd_nfreq = int(cal.get("msd_nfreq", 200))
 
     def task_type(self):
         return self.parameter["type"]
@@ -59,23 +100,57 @@ def task_param(self):
             {
                 "start_temp": self.start_temp,
                 "target_temp": self.target_temp,
+                "temp": self.target_temp,
                 "end_temp": self.end_temp,
+                "temp_ramp_rate": self.temp_ramp_rate,
                 "equi_step": self.equi_step,
+                "init_lgv_thermo_equil_step": self.init_lgv_thermo_equil_step,
+                "init_thermo_equil_step": self.init_thermo_equil_step,
+                "final_thermo_equil_step": self.final_thermo_equil_step,
                 "ramp_step": self.ramp_step,
+                "temp_ramp_step": self.ramp_step,
                 "hold_step": self.hold_step,
                 "cool_step": self.cool_step,
+                "temp_decline_step": self.cool_step,
                 "thermostat": self.thermostat,
                 "ensemble": self.ensemble,
-                "tdamp": self.tdamp,
-                "pdamp": self.pdamp,
+                "tdamp_factor": self.tdamp_factor,
+                "pdamp_factor": self.pdamp_factor,
                 "velocity_seed": self.velocity_seed,
+                "lgv_seed": self.lgv_seed,
+                "req_lgv_damping": self.req_lgv_damping,
+                "req_opti_init_structure": self.req_opti_init_structure,
+                "req_write_restart": self.req_write_restart,
+                "req_dump_init_atom": self.req_dump_init_atom,
+                "req_dump_ave_atom": self.req_dump_ave_atom,
                 "dump_step": self.dump_step,
+                "dump_interval": self.dump_interval,
+                "thermo_interval": self.thermo_interval,
+                "restart_interval": self.restart_interval,
+                "ave_atom_sample_feq": self.ave_atom_sample_feq,
+                "ave_atom_num_sample": self.ave_atom_num_sample,
+                "ave_atom_sample_length": self.ave_atom_sample_length,
+                "init_opt_loop_size": self.init_opt_loop_size,
+                "init_fmax_tol": self.init_fmax_tol,
+                "init_stress_tol": self.init_stress_tol,
                 "timestep": self.timestep,
+                "req_compute_rdf": self.req_compute_rdf,
                 "rdf_bins": self.rdf_bins,
                 "rdf_cutoff": self.rdf_cutoff,
                 "rdf_interval": self.rdf_interval,
+                "rdf_nevery": self.rdf_nevery,
+                "rdf_nrepeat": self.rdf_nrepeat,
+                "rdf_nfreq": self.rdf_nfreq,
+                "req_compute_msd": self.req_compute_msd,
+                "msd_nevery": self.msd_nevery,
+                "msd_nrepeat": self.msd_nrepeat,
+                "msd_nfreq": self.msd_nfreq,
             }
         )
+        if self.tdamp is not None:
+            cal["tdamp"] = self.tdamp
+        if self.pdamp is not None:
+            cal["pdamp"] = self.pdamp
         self.parameter["supercell_size"] = self.supercell_size
         if self.supercell_length is not None:
             self.parameter["supercell_length"] = self.supercell_length
@@ -95,7 +170,10 @@ def make_confs(self, path_to_work: str, path_to_equi: str, refine=False) -> List
 
         task_list: List[str] = []
         # One task per target_temp (allow list), else single
-        targets = self.parameter.get("cal_setting", {}).get("target_temp", self.target_temp)
+        targets = self.parameter.get("cal_setting", {}).get(
+            "target_temp",
+            self.parameter.get("cal_setting", {}).get("temp", self.target_temp),
+        )
         if not isinstance(targets, list):
             targets = [targets]
 
@@ -127,7 +205,9 @@ def make_confs(self, path_to_work: str, path_to_equi: str, refine=False) -> List
             anneal_task = {
                 "start_temp": self.start_temp,
                 "target_temp": float(tgt),
+                "temp": float(tgt),
                 "end_temp": self.end_temp,
+                "temp_ramp_rate": self.temp_ramp_rate,
                 "supercell_size": self.supercell_size,
             }
             dumpfn(anneal_task, os.path.join(task_dir, "Annealing.json"), indent=4)
@@ -140,36 +220,100 @@ def make_confs(self, path_to_work: str, path_to_equi: str, refine=False) -> List
             var.append(f"variable nz equal {self.supercell_size[2]}")
             var.append(f"variable start_temp equal {self.start_temp:.2f}")
             var.append(f"variable target_temp equal {float(tgt):.2f}")
+            var.append(f"variable temp equal {float(tgt):.2f}")
             var.append(f"variable end_temp equal {self.end_temp:.2f}")
+            var.append(f"variable temp_ramp_rate equal {self.temp_ramp_rate}")
             var.append(f"variable equi_step equal {self.equi_step}")
             # derive ramp/cool steps if rates are provided (K/step); else use defaults
             import math
-            if self.ramp_rate is not None:
+            if self._has_ramp_rate and self.temp_ramp_rate is not None:
                 try:
                     # convert K/ns -> steps using timestep (ps): dt_ns = dt_ps/1000
-                    rstep = max(1, int(math.ceil(abs(float(tgt) - self.start_temp) * 1000.0 / (float(self.ramp_rate) * self.timestep))))
+                    rstep = max(1, int(math.ceil(abs(float(tgt) - self.start_temp) * 1000.0 / (float(self.temp_ramp_rate) * self.timestep))))
+                except Exception:
+                    rstep = self.ramp_step
+            elif self.ramp_step > 0:
+                rstep = self.ramp_step
+            elif self.temp_ramp_rate is not None:
+                try:
+                    rstep = max(1, int(math.ceil(abs(float(tgt) - self.start_temp) * 1000.0 / (float(self.temp_ramp_rate) * self.timestep))))
                 except Exception:
                     rstep = self.ramp_step
             else:
                 rstep = self.ramp_step
-            if self.cool_rate is not None:
+            if (self._has_cool_rate or self._has_ramp_rate) and self.cool_rate is not None:
                 try:
                     cr = self.cool_rate[idx] if isinstance(self.cool_rate, (list, tuple)) else float(self.cool_rate)
                     cstep = max(1, int(math.ceil(abs(float(tgt) - self.end_temp) * 1000.0 / (float(cr) * self.timestep))))
                 except Exception:
                     cstep = self.cool_step
+            elif self.cool_step > 0:
+                cstep = self.cool_step
+            elif self.cool_rate is not None:
+                try:
+                    cstep = max(1, int(math.ceil(abs(float(tgt) - self.end_temp) * 1000.0 / (float(self.cool_rate) * self.timestep))))
+                except Exception:
+                    cstep = self.cool_step
             else:
                 cstep = self.cool_step
             var.append(f"variable ramp_step equal {rstep}")
+            var.append(f"variable temp_ramp_step equal {rstep}")
+            var.append(f"variable temp_ramp_remain_step equal {rstep}")
             var.append(f"variable hold_step equal {self.hold_step}")
             var.append(f"variable cool_step equal {cstep}")
+            var.append(f"variable temp_decline_step equal {cstep}")
+            var.append(f"variable temp_decline_remain_step equal {cstep}")
+            var.append(f"variable init_lgv_thermo_equil_step equal {self.init_lgv_thermo_equil_step}")
+            var.append(f"variable init_thermo_equil_step equal {self.init_thermo_equil_step}")
+            var.append(f"variable final_thermo_equil_step equal {self.final_thermo_equil_step}")
+            var.append(f"variable final_thermo_equil_remain_step equal {self.final_thermo_equil_step}")
             var.append(f"variable timestep equal {self.timestep}")
+            var.append(f"variable thermo_interval equal {self.thermo_interval}")
+            var.append(f"variable dump_interval equal {self.dump_interval}")
+            var.append(f"variable restart_interval equal {self.restart_interval}")
+            var.append(f"variable req_lgv_damping equal {str(self.req_lgv_damping).lower()}")
+            var.append(f"variable req_opti_init_structure equal {str(self.req_opti_init_structure).lower()}")
+            var.append(f"variable req_write_restart equal {str(self.req_write_restart).lower()}")
+            var.append(f"variable req_dump_init_atom equal {str(self.req_dump_init_atom).lower()}")
+            var.append(f"variable req_dump_ave_atom equal {str(self.req_dump_ave_atom).lower()}")
+            var.append(f"variable ave_atom_sample_feq equal {self.ave_atom_sample_feq}")
+            var.append(f"variable ave_atom_num_sample equal {self.ave_atom_num_sample}")
+            var.append(f"variable ave_atom_sample_length equal {self.ave_atom_sample_length}")
+            var.append(f"variable init_opt_loop_size equal {self.init_opt_loop_size}")
+            var.append(f"variable init_fmax_tol equal {self.init_fmax_tol}")
+            var.append(f"variable init_stress_tol equal {self.init_stress_tol}")
+            var.append(f"variable req_compute_rdf equal {str(self.req_compute_rdf).lower()}")
             var.append(f"variable rdf_bins equal {self.rdf_bins}")
             var.append(f"variable rdf_cutoff equal {self.rdf_cutoff}")
             var.append(f"variable rdf_interval equal {self.rdf_interval}")
-            var.append(f"variable tdamp equal {self.tdamp}")
-            var.append(f"variable pdamp equal {self.pdamp}")
+            var.append(f"variable rdf_nevery equal {self.rdf_nevery}")
+            var.append(f"variable rdf_nrepeat equal {self.rdf_nrepeat}")
+            var.append(f"variable rdf_nfreq equal {self.rdf_nfreq}")
+            var.append("variable rdf_file_eq string rdf.eq_${start_temp}K.txt")
+            var.append("variable rdf_file_ramp string rdf.T_ramp_${start_temp}K_${temp}K.txt")
+            var.append("variable rdf_file_decline string rdf.T_decline_${temp}K_${end_temp}K.txt")
+            var.append("variable rdf_file_final_eq string rdf.final_eq_${end_temp}K.txt")
+            var.append(f"variable req_compute_msd equal {str(self.req_compute_msd).lower()}")
+            var.append(f"variable msd_nevery equal {self.msd_nevery}")
+            var.append(f"variable msd_nrepeat equal {self.msd_nrepeat}")
+            var.append(f"variable msd_nfreq equal {self.msd_nfreq}")
+            var.append("variable msd_file_eq string msd.eq_${start_temp}K.txt")
+            var.append("variable msd_file_ramp string msd.T_ramp_${start_temp}K_${temp}K.txt")
+            var.append("variable msd_file_decline string msd.T_decline_${temp}K_${end_temp}K.txt")
+            var.append("variable msd_file_final_eq string msd.final_eq_${end_temp}K.txt")
+            var.append(f"variable tdamp_factor equal {self.tdamp_factor}")
+            var.append(f"variable pdamp_factor equal {self.pdamp_factor}")
+            if self.tdamp is not None:
+                var.append(f"variable tdamp equal {self.tdamp}")
+            else:
+                var.append("variable tdamp equal v_tdamp_factor*${timestep}")
+            if self.pdamp is not None:
+                var.append(f"variable pdamp equal {self.pdamp}")
+            else:
+                var.append("variable pdamp equal v_pdamp_factor*${timestep}")
             var.append(f"variable velocity_seed equal {int(self.velocity_seed)}")
+            var.append(f"variable init_v_seed equal {int(self.velocity_seed)}")
+            var.append(f"variable lgv_seed equal {int(self.lgv_seed)}")
             var.append(f"variable dump_step equal {self.dump_step}")
             var.append(f"variable thermostat string {self.thermostat}")
             var.append(f"variable ensemble string {self.ensemble}")
@@ -185,13 +329,271 @@ def post_process(self, task_list: List[str]):
         pass
 
     def _compute_lower(self, output_file, all_tasks, all_res):
-        # Minimal aggregator: return basic info per task; users inspect dumps/logs
-        res_data = {}
-        ptr_data = os.path.dirname(output_file) + "\n"
-        for t in all_tasks:
-            name = os.path.basename(t)
-            res_data[name] = {
-                "task": name,
-                "note": "annealing run; inspect log.lammps and dump files",
-            }
+        res_data = {
+            "property": "annealing",
+            "tasks": {},
+        }
+        ptr_lines = [os.path.dirname(output_file)]
+        for task_dir in all_tasks:
+            name = os.path.basename(task_dir)
+            task_result = self._collect_task_result(task_dir)
+            res_data["tasks"][name] = task_result
+            ptr_lines.append(self._task_summary_line(name, task_result))
+        ptr_data = "\n".join(ptr_lines) + "\n"
+        dumpfn(res_data, output_file, indent=4)
         return res_data, ptr_data
+
+    @classmethod
+    def _collect_task_result(cls, task_dir: str) -> Dict[str, Any]:
+        task_path = os.path.abspath(task_dir)
+        metadata_path = os.path.join(task_path, "Annealing.json")
+        status_path = os.path.join(task_path, "apex_task_status.json")
+        result = {
+            "task": os.path.basename(task_path),
+            "path": task_path,
+            "metadata": cls._safe_load_json(metadata_path),
+            "status": cls._safe_load_json(status_path),
+            "rdf": cls._collect_rdf(task_path),
+            "msd": cls._collect_msd(task_path),
+            "volume_temperature": cls._collect_volume_temperature(task_path),
+        }
+        result["summary"] = cls._build_summary(result)
+        return result
+
+    @staticmethod
+    def _safe_load_json(path: str):
+        try:
+            if os.path.isfile(path):
+                return loadfn(path)
+        except Exception as exc:
+            return {"error": f"failed to read {os.path.basename(path)}: {exc}"}
+        return {}
+
+    @classmethod
+    def _collect_rdf(cls, task_path: str) -> Dict[str, Any]:
+        stages = {}
+        for path in sorted(glob.glob(os.path.join(task_path, "rdf.*.txt"))):
+            stage = cls._stage_from_analysis_file(path, prefix="rdf", suffix=".txt")
+            parsed = cls._parse_rdf_file(path)
+            if parsed:
+                stages[stage] = parsed
+        return stages
+
+    @classmethod
+    def _collect_msd(cls, task_path: str) -> Dict[str, Any]:
+        stages = {}
+        for path in sorted(glob.glob(os.path.join(task_path, "msd.*.txt"))):
+            stage = cls._stage_from_analysis_file(path, prefix="msd", suffix=".txt")
+            parsed = cls._parse_msd_file(path)
+            if parsed:
+                stages[stage] = parsed
+        return stages
+
+    @classmethod
+    def _collect_volume_temperature(cls, task_path: str) -> Dict[str, Any]:
+        stages = {}
+        for path in sorted(glob.glob(os.path.join(task_path, "heating_interval_*.dat"))):
+            parsed = cls._parse_thermo_interval_file(path)
+            if parsed:
+                stages["heating"] = parsed
+        for path in sorted(glob.glob(os.path.join(task_path, "cooling_interval_*.dat"))):
+            parsed = cls._parse_thermo_interval_file(path)
+            if parsed:
+                stages["cooling"] = parsed
+        return stages
+
+    @staticmethod
+    def _stage_from_analysis_file(path: str, prefix: str, suffix: str) -> str:
+        name = os.path.basename(path)
+        if name.startswith(prefix + "."):
+            name = name[len(prefix) + 1:]
+        if suffix and name.endswith(suffix):
+            name = name[:-len(suffix)]
+        return name
+
+    @staticmethod
+    def _numeric_tokens(line: str) -> List[float]:
+        values = []
+        for token in line.split():
+            try:
+                values.append(float(token))
+            except ValueError:
+                return []
+        return values
+
+    @classmethod
+    def _parse_ave_time_blocks(cls, path: str):
+        column_names = []
+        blocks = []
+        try:
+            with open(path, "r", errors="replace") as fp:
+                lines = fp.readlines()
+        except OSError:
+            return column_names, blocks
+
+        idx = 0
+        while idx < len(lines):
+            line = lines[idx].strip()
+            if not line:
+                idx += 1
+                continue
+            if line.startswith("#"):
+                if line.startswith("# Row "):
+                    column_names = line[2:].split()[1:]
+                idx += 1
+                continue
+
+            header = line.split()
+            if len(header) != 2:
+                idx += 1
+                continue
+            try:
+                timestep = int(float(header[0]))
+                nrows = int(float(header[1]))
+            except ValueError:
+                idx += 1
+                continue
+
+            rows = []
+            for raw_row in lines[idx + 1: idx + 1 + nrows]:
+                values = cls._numeric_tokens(raw_row.strip())
+                if values:
+                    rows.append(values)
+            if rows:
+                blocks.append({"timestep": timestep, "rows": rows})
+            idx += 1 + nrows
+        return column_names, blocks
+
+    @classmethod
+    def _parse_rdf_file(cls, path: str) -> Dict[str, Any]:
+        column_names, blocks = cls._parse_ave_time_blocks(path)
+        if not blocks:
+            return {}
+        last_block = blocks[-1]
+        rows = [
+            row[1:] if column_names and len(row) == len(column_names) + 1 else row
+            for row in last_block["rows"]
+        ]
+        columns = {}
+        ncols = max(len(row) for row in rows)
+        if len(column_names) < ncols:
+            column_names = column_names + [f"column_{idx}" for idx in range(len(column_names), ncols)]
+        for idx in range(ncols):
+            columns[column_names[idx]] = [row[idx] for row in rows if len(row) > idx]
+
+        radius = columns.get("c_myRDF[1]", columns.get("column_1", []))
+        g_r = columns.get("c_myRDF[2]", columns.get("column_2", []))
+        coordination = columns.get("c_myRDF[3]", columns.get("column_3", []))
+        return {
+            "source": os.path.basename(path),
+            "timestep": last_block["timestep"],
+            "nblocks": len(blocks),
+            "columns": columns,
+            "radius": radius,
+            "g_r": g_r,
+            "coordination": coordination,
+        }
+
+    @classmethod
+    def _parse_msd_file(cls, path: str) -> Dict[str, Any]:
+        _column_names, blocks = cls._parse_ave_time_blocks(path)
+        if not blocks:
+            return {}
+        timesteps = []
+        msd_x = []
+        msd_y = []
+        msd_z = []
+        msd_total = []
+        for block in blocks:
+            values = [row[1] if len(row) > 1 else row[0] for row in block["rows"]]
+            if len(values) < 4:
+                continue
+            timesteps.append(block["timestep"])
+            msd_x.append(values[0])
+            msd_y.append(values[1])
+            msd_z.append(values[2])
+            msd_total.append(values[3])
+        return {
+            "source": os.path.basename(path),
+            "timestep": timesteps,
+            "msd_x": msd_x,
+            "msd_y": msd_y,
+            "msd_z": msd_z,
+            "msd_total": msd_total,
+        }
+
+    @classmethod
+    def _parse_thermo_interval_file(cls, path: str) -> Dict[str, Any]:
+        header = []
+        rows = []
+        try:
+            with open(path, "r", errors="replace") as fp:
+                for raw_line in fp:
+                    line = raw_line.strip()
+                    if not line:
+                        continue
+                    if line.startswith("#"):
+                        if line.startswith("# TimeStep"):
+                            header = line[2:].split()
+                        continue
+                    values = cls._numeric_tokens(line)
+                    if values:
+                        rows.append(values)
+        except OSError:
+            return {}
+        if not header or not rows:
+            return {}
+
+        columns = {}
+        for idx, name in enumerate(header):
+            columns[name] = [row[idx] for row in rows if len(row) > idx]
+
+        atom_count = columns.get("v_N", [])
+        volume_per_atom = columns.get("v_Vatom", [])
+        total_volume = [
+            n * v
+            for n, v in zip(atom_count, volume_per_atom)
+        ]
+        return {
+            "source": os.path.basename(path),
+            "timestep": columns.get("TimeStep", []),
+            "temperature": columns.get("v_Temp", []),
+            "volume_per_atom": volume_per_atom,
+            "total_volume": total_volume,
+            "potential_energy": columns.get("v_pote", []),
+            "total_energy": columns.get("v_Etotal", []),
+            "pressure": columns.get("v_Press", []),
+        }
+
+    @staticmethod
+    def _build_summary(task_result: Dict[str, Any]) -> Dict[str, Any]:
+        rdf_points = {
+            stage: len(data.get("radius", []))
+            for stage, data in task_result.get("rdf", {}).items()
+        }
+        msd_points = {
+            stage: len(data.get("timestep", []))
+            for stage, data in task_result.get("msd", {}).items()
+        }
+        volume_points = {
+            stage: len(data.get("temperature", []))
+            for stage, data in task_result.get("volume_temperature", {}).items()
+        }
+        return {
+            "rdf_stages": sorted(task_result.get("rdf", {}).keys()),
+            "msd_stages": sorted(task_result.get("msd", {}).keys()),
+            "volume_temperature_stages": sorted(task_result.get("volume_temperature", {}).keys()),
+            "rdf_points": rdf_points,
+            "msd_points": msd_points,
+            "volume_temperature_points": volume_points,
+        }
+
+    @staticmethod
+    def _task_summary_line(name: str, task_result: Dict[str, Any]) -> str:
+        summary = task_result.get("summary", {})
+        return (
+            f"{name}: "
+            f"rdf={summary.get('rdf_stages', [])}, "
+            f"msd={summary.get('msd_stages', [])}, "
+            f"volume_temperature={summary.get('volume_temperature_stages', [])}"
+        )
diff --git a/apex/flow.py b/apex/flow.py
index d34f23e8..d63cab76 100644
--- a/apex/flow.py
+++ b/apex/flow.py
@@ -151,13 +151,12 @@ def _format_step_failure(
             return f"{fallback_label}: no step details available"
 
         detail_keys = [
+            "id",
+            "key",
             "phase",
             "message",
             "reason",
-            "podName",
-            "pod_name",
             "displayName",
-            "name",
             "finishedAt",
             "startedAt",
         ]
@@ -172,25 +171,148 @@ def _format_step_failure(
                 lines.append(f"  {key}: {value}")
 
         if main_log_path:
+            log_paths = main_log_path
             if isinstance(main_log_path, list):
                 main_log_path = ", ".join(str(item) for item in main_log_path)
             lines.append(f"  main_logs: {main_log_path}")
+            log_excerpt = FlowGenerator._failure_log_excerpt(log_paths)
+            if log_excerpt:
+                cause = FlowGenerator._failure_cause_from_excerpt(log_excerpt)
+                if cause:
+                    lines.append(f"  cause: {cause}")
+                lines.append("  main_logs_excerpt:")
+                lines.extend(f"    {line}" for line in log_excerpt.splitlines())
         elif main_log_error:
             lines.append(f"  main_logs: unavailable ({main_log_error})")
 
         if diagnostic_artifacts:
             artifact_text = ", ".join(str(item) for item in diagnostic_artifacts)
             lines.append(f"  failed_artifacts: {artifact_text}")
-
-        if isinstance(step, dict):
-            try:
-                lines.append("  raw_step: " + json.dumps(step, default=str, sort_keys=True))
-            except TypeError:
+            artifact_excerpt = FlowGenerator._failure_log_excerpt(diagnostic_artifacts)
+            if artifact_excerpt:
+                cause = FlowGenerator._failure_cause_from_excerpt(artifact_excerpt)
+                if cause:
+                    lines.append(f"  cause: {cause}")
+                lines.append("  failed_artifacts_excerpt:")
+                lines.extend(f"    {line}" for line in artifact_excerpt.splitlines())
+
+        if os.environ.get("APEX_SHOW_RAW_STEP", "").lower() in {"1", "true", "yes"}:
+            if isinstance(step, dict):
+                try:
+                    lines.append("  raw_step: " + json.dumps(step, default=str, sort_keys=True))
+                except TypeError:
+                    lines.append(f"  raw_step: {step!r}")
+            else:
                 lines.append(f"  raw_step: {step!r}")
-        else:
-            lines.append(f"  raw_step: {step!r}")
         return "\n".join(lines)
 
+    @staticmethod
+    def _failure_log_excerpt(paths, max_lines: int = 80, max_chars: int = 12000) -> str:
+        if paths is None:
+            return ""
+        if isinstance(paths, (str, os.PathLike)):
+            path_list = [paths]
+        else:
+            path_list = list(paths)
+
+        chunks = []
+        for raw_path in path_list:
+            for path in FlowGenerator._failure_log_candidates(raw_path):
+                text = FlowGenerator._read_failure_log_tail(path, max_lines=max_lines)
+                if not text:
+                    continue
+                rel_path = os.path.relpath(path, os.getcwd())
+                chunks.append(f"--- {rel_path} ---\n{text}")
+                if sum(len(chunk) for chunk in chunks) >= max_chars:
+                    break
+            if sum(len(chunk) for chunk in chunks) >= max_chars:
+                break
+        result = "\n".join(chunks)
+        if len(result) > max_chars:
+            result = "<truncated>\n" + result[-max_chars:]
+        return result
+
+    @staticmethod
+    def _failure_cause_from_excerpt(excerpt: str) -> str:
+        if not excerpt:
+            return ""
+        lines = [line.strip() for line in excerpt.splitlines() if line.strip()]
+        exception_re = re.compile(r"^(?:[A-Za-z_][A-Za-z0-9_.]*)(?:Error|Exception|Warning): .+")
+        for line in reversed(lines):
+            if exception_re.match(line):
+                return line
+        markers = ("ERROR", "Error", "error", "Failed", "failed")
+        for line in reversed(lines):
+            if any(marker in line for marker in markers):
+                return line
+        return lines[-1] if lines else ""
+
+    @staticmethod
+    def _failure_log_candidates(raw_path) -> List[str]:
+        if isinstance(raw_path, list):
+            candidates = []
+            for item in raw_path:
+                candidates.extend(FlowGenerator._failure_log_candidates(item))
+            return candidates
+
+        path = os.fspath(raw_path)
+        if not os.path.exists(path):
+            return []
+        if os.path.isfile(path):
+            return [path]
+
+        preferred_names = {
+            "main.log",
+            "executor.log",
+            "stderr",
+            "stdout",
+            ".debug.log",
+            ".debug.stderr",
+            ".debug.stdout",
+            "apex_task_status.json",
+            "failed_lammps_tasks.json",
+            "errlog",
+            "outlog",
+            "log.lammps",
+            "run.log",
+        }
+        preferred_suffixes = (".log", ".out", ".err")
+        candidates = []
+        for root, _dirs, files in os.walk(path):
+            for name in files:
+                file_path = os.path.join(root, name)
+                if name in preferred_names or name.endswith(preferred_suffixes):
+                    candidates.append(file_path)
+        candidates.sort(key=lambda item: (
+            os.path.basename(item) not in preferred_names,
+            item.count(os.sep),
+            item,
+        ))
+        return candidates[:12]
+
+    @staticmethod
+    def _read_failure_log_tail(path: str, max_lines: int = 80) -> str:
+        try:
+            with open(path, "r", errors="replace") as fp:
+                lines = fp.readlines()
+        except Exception as exc:
+            return f"<unable to read {path}: {exc}>"
+        cleaned = [line.rstrip("\n") for line in lines if line.strip()]
+        if not cleaned:
+            return ""
+
+        traceback_start = None
+        for index, line in enumerate(cleaned):
+            if line.startswith("Traceback (most recent call last):"):
+                traceback_start = index
+        if traceback_start is not None:
+            selected = cleaned[traceback_start:]
+        else:
+            selected = cleaned[-max_lines:]
+        if len(selected) > max_lines:
+            selected = selected[-max_lines:]
+        return "\n".join(selected)
+
     @staticmethod
     def _safe_get(obj, key, default=None):
         if isinstance(obj, dict):
@@ -341,7 +463,8 @@ def _download_step_main_logs(self, step, step_label: str, step_info=None):
         )
         os.makedirs(log_dir, exist_ok=True)
         try:
-            return self._download_artifact_with_retry(artifact=artifact, path=log_dir), None
+            downloaded_path = self._download_artifact_with_retry(artifact=artifact, path=log_dir)
+            return downloaded_path or log_dir, None
         except Exception as exc:
             return None, str(exc)
 
diff --git a/apex/main.py b/apex/main.py
index c4f5e9d5..1d30565d 100644
--- a/apex/main.py
+++ b/apex/main.py
@@ -1087,6 +1087,11 @@ def _download_failure_artifacts_for_step(wf_info, root_step, key, work_dir):
     return downloaded
 
 
+def _is_workflow_failure_summary(exc: Exception) -> bool:
+    text = str(exc)
+    return " failed with " in text and " failed step(s):" in text
+
+
 def main():
     # logging
     logging.basicConfig(level=logging.INFO)
@@ -1094,16 +1099,21 @@ def main():
     parser, args = parse_args()
     if args.cmd == 'submit':
         header()
-        submit_from_args(
-            parameters=args.parameter,
-            config_file=args.config,
-            work_dirs=args.work,
-            indicated_flow_type=args.flow,
-            flow_name=args.name,
-            submit_only=args.submit_only,
-            is_debug=args.debug,
-            labels=args.label,
-        )
+        try:
+            submit_from_args(
+                parameters=args.parameter,
+                config_file=args.config,
+                work_dirs=args.work,
+                indicated_flow_type=args.flow,
+                flow_name=args.name,
+                submit_only=args.submit_only,
+                is_debug=args.debug,
+                labels=args.label,
+            )
+        except RuntimeError as exc:
+            if _is_workflow_failure_summary(exc):
+                raise SystemExit(str(exc)) from None
+            raise
     elif args.cmd == "list":
             config_dflow(args.config)
             if args.label is not None:
diff --git a/apex/reporter/DashReportApp.py b/apex/reporter/DashReportApp.py
index 65f64800..69262779 100644
--- a/apex/reporter/DashReportApp.py
+++ b/apex/reporter/DashReportApp.py
@@ -44,6 +44,8 @@ def return_prop_class(prop_type: str):
         return FiniteTlattReport
     elif prop_type == 'finite_t_elastic':
         return FiniteTelasticReport
+    elif prop_type == 'annealing':
+        return AnnealingReport
 
 
 def return_prop_type(prop: str):
diff --git a/apex/reporter/property_report.py b/apex/reporter/property_report.py
index 2e8b8f4c..a402c3ce 100644
--- a/apex/reporter/property_report.py
+++ b/apex/reporter/property_report.py
@@ -433,6 +433,117 @@ def dash_table(res_data: dict, decimal: int = 3, **kwargs) -> dash_table.DataTab
         return build_table(df, cell_style={"width": "120px"}), df
 
 
+class AnnealingReport(PropertyReport):
+    """Report annealing RDF, MSD, and temperature-volume response."""
+
+    @staticmethod
+    def _iter_tasks(res_data: dict):
+        tasks = res_data.get("tasks", {})
+        if tasks:
+            return sorted(tasks.items())
+        return []
+
+    @staticmethod
+    def plotly_graph(res_data: dict, name: str, **kwargs):
+        traces = []
+        color = kwargs.get("color")
+        for task_name, task_data in AnnealingReport._iter_tasks(res_data):
+            for stage, rdf_data in sorted(task_data.get("rdf", {}).items()):
+                radius = rdf_data.get("radius", [])
+                g_r = rdf_data.get("g_r", [])
+                if not radius or not g_r:
+                    continue
+                traces.append(
+                    go.Scatter(
+                        name=f"{name} {task_name} RDF {stage}",
+                        x=radius,
+                        y=g_r,
+                        mode="lines",
+                        xaxis="x",
+                        yaxis="y",
+                        line=dict(color=color) if color else None,
+                    )
+                )
+
+            for stage, msd_data in sorted(task_data.get("msd", {}).items()):
+                timesteps = msd_data.get("timestep", [])
+                total = msd_data.get("msd_total", [])
+                if not timesteps or not total:
+                    continue
+                traces.append(
+                    go.Scatter(
+                        name=f"{name} {task_name} MSD {stage}",
+                        x=timesteps,
+                        y=total,
+                        mode="lines",
+                        xaxis="x2",
+                        yaxis="y2",
+                    )
+                )
+
+            for stage, vt_data in sorted(task_data.get("volume_temperature", {}).items()):
+                temps = vt_data.get("temperature", [])
+                volume = vt_data.get("volume_per_atom", [])
+                if not temps or not volume:
+                    continue
+                traces.append(
+                    go.Scatter(
+                        name=f"{name} {task_name} {stage} V(T)",
+                        x=temps,
+                        y=volume,
+                        mode="lines+markers",
+                        xaxis="x3",
+                        yaxis="y3",
+                    )
+                )
+
+        layout = go.Layout(
+            title="Annealing RDF, MSD, and Volume-Temperature Response",
+            showlegend=True,
+            xaxis=dict(title="r (Å)", domain=[0.0, 1.0], anchor="y"),
+            yaxis=dict(title="g(r)", domain=[0.70, 1.0], anchor="x"),
+            xaxis2=dict(title="Timestep", domain=[0.0, 1.0], anchor="y2"),
+            yaxis2=dict(title="MSD (Ų)", domain=[0.36, 0.64], anchor="x2"),
+            xaxis3=dict(title="Temperature (K)", domain=[0.0, 1.0], anchor="y3"),
+            yaxis3=dict(title="Volume/atom (ų)", domain=[0.0, 0.30], anchor="x3"),
+            height=850,
+        )
+        return traces, layout
+
+    @staticmethod
+    def dash_table(res_data: dict, decimal: int = 3, **kwargs) -> dash_table.DataTable:
+        rows = []
+        for task_name, task_data in AnnealingReport._iter_tasks(res_data):
+            summary = task_data.get("summary", {})
+            for stage in sorted(set(
+                    list(task_data.get("rdf", {}).keys())
+                    + list(task_data.get("msd", {}).keys())
+                    + list(task_data.get("volume_temperature", {}).keys())
+            )):
+                rdf_points = summary.get("rdf_points", {}).get(stage, 0)
+                msd_points = summary.get("msd_points", {}).get(stage, 0)
+                volume_points = summary.get("volume_temperature_points", {}).get(stage, 0)
+                volume_data = task_data.get("volume_temperature", {}).get(stage, {})
+                temps = volume_data.get("temperature", [])
+                volumes = volume_data.get("volume_per_atom", [])
+                row = {
+                    "Task": task_name,
+                    "Stage": stage,
+                    "RDF points": rdf_points,
+                    "MSD timesteps": msd_points,
+                    "V(T) points": volume_points,
+                }
+                if temps:
+                    row["T min (K)"] = round(float(min(temps)), decimal)
+                    row["T max (K)"] = round(float(max(temps)), decimal)
+                if volumes:
+                    row["V/atom min (ų)"] = round(float(min(volumes)), decimal)
+                    row["V/atom max (ų)"] = round(float(max(volumes)), decimal)
+                rows.append(row)
+        df = pd.DataFrame(rows)
+        return build_table(df, cell_style={"width": "130px"}), df
+
+
 class ElasticReport(PropertyReport):
     @staticmethod
     def plotly_graph(res_data: dict, name: str, **kwargs):
diff --git a/examples/annealing/confs/mo-bcc/POSCAR b/examples/annealing/confs/mo-bcc/POSCAR
new file mode 100644
index 00000000..97240a37
--- /dev/null
+++ b/examples/annealing/confs/mo-bcc/POSCAR
@@ -0,0 +1,13 @@
+Mo2                                     
+   1.0000000000000000     
+     3.1623672675177916   -0.0000000000000000   -0.0000000000000000
+    -0.0000000000000000    3.1623672675177916   -0.0000000000000000
+     0.0000000000000000    0.0000000000000000    3.1623672675177916
+   Mo
+     2
+Direct
+  0.0000000000000000  0.0000000000000000  0.0000000000000000
+  0.5000000000000000  0.5000000000000000  0.5000000000000000
+
+  0.00000000E+00  0.00000000E+00  0.00000000E+00
+  0.00000000E+00  0.00000000E+00  0.00000000E+00
diff --git a/examples/annealing/global.json b/examples/annealing/global.json
new file mode 100644
index 00000000..704aa2fa
--- /dev/null
+++ b/examples/annealing/global.json
@@ -0,0 +1,7 @@
+{
+    "lammps_image_name": "registry.dp.tech/dptech/deepmd-kit:3.1.1",
+    "lammps_run_command": "lmp -in in.lammps -log log.lammps -screen outlog",
+    "group_size": 1,
+    "pool_size": 1,
+    "scass_type": "c8_m31_1 * NVIDIA T4"
+}
diff --git a/examples/annealing/param_joint.json b/examples/annealing/param_joint.json
new file mode 100644
index 00000000..a3547a4f
--- /dev/null
+++ b/examples/annealing/param_joint.json
@@ -0,0 +1,45 @@
+{
+    "structures": ["confs/*"],
+    "interaction": {
+        "type": "deepmd",
+        "model": "frozen_model.pb",
+        "type_map": {
+            "Mo": 0
+        }
+    },
+    "relaxation": {
+        "cal_type": "relaxation",
+        "req_recal": false,
+        "cal_setting": {
+            "etol": 0,
+            "ftol": 1e-10,
+            "maxiter": 5000,
+            "maxeval": 500000
+        }
+    },
+    "properties": [
+        {
+            "type": "annealing",
+            "supercell_size": [4, 4, 4],
+            "cal_setting": {
+                "start_temp": 300,
+                "target_temp": 1500,
+                "end_temp": 300,
+                "equi_step": 10000,
+                "ramp_step": 20000,
+                "hold_step": 10000,
+                "cool_step": 20000,
+                "thermostat": "nose_hoover",
+                "ensemble": "npt",
+                "tdamp": 0.1,
+                "pdamp": 1.0,
+                "velocity_seed": 12345,
+                "dump_step": 1000,
+                "timestep": 0.002,
+                "rdf_bins": 200,
+                "rdf_cutoff": 5.0,
+                "rdf_interval": 100
+            }
+        }
+    ]
+}
diff --git a/tests/test_annealing.py b/tests/test_annealing.py
index b3f98928..425e88c7 100644
--- a/tests/test_annealing.py
+++ b/tests/test_annealing.py
@@ -5,10 +5,13 @@
 from pathlib import Path
 
 import pytest
-from monty.serialization import loadfn
+from monty.serialization import dumpfn, loadfn
 
+from apex.archive import ResultStorage
 from apex.core.calculator.lib import lammps_utils
 from apex.core.property.Annealing import Annealing
+from apex.reporter.DashReportApp import DashReportApp, return_prop_class, return_prop_type
+from apex.reporter.property_report import AnnealingReport
 
 
 TEST_CONTCAR = os.path.join(
@@ -33,21 +36,27 @@ def test_annealing_default_parameter_parsing():
     prop = Annealing({"type": "annealing"})
 
     assert prop.task_type() == "annealing"
-    assert prop.start_temp == 300.0
-    assert prop.target_temp == 800.0
-    assert prop.end_temp == 300.0
-    assert prop.equi_step == 10000
-    assert prop.ramp_step == 20000
-    assert prop.hold_step == 0
-    assert prop.cool_step == 20000
+    assert prop.start_temp == 4.0
+    assert prop.target_temp == 300.0
+    assert prop.end_temp == 4.0
+    assert prop.temp_ramp_rate == 1000
+    assert prop.equi_step == 20000
+    assert prop.init_thermo_equil_step == 20000
+    assert prop.final_thermo_equil_step == 20000
+    assert prop.ramp_step == 0
+    assert prop.hold_step == 20000
+    assert prop.cool_step == 0
     assert prop.thermostat == "nose_hoover"
     assert prop.ensemble == "npt"
     assert prop.supercell_size == [2, 2, 2]
 
     task_param = prop.task_param()
     assert task_param["cal_type"] == "annealing"
-    assert task_param["cal_setting"]["rdf_bins"] == 200
-    assert task_param["cal_setting"]["timestep"] == 0.002
+    assert task_param["cal_setting"]["rdf_bins"] == 100
+    assert task_param["cal_setting"]["rdf_cutoff"] == 6.0
+    assert task_param["cal_setting"]["req_compute_rdf"] is True
+    assert task_param["cal_setting"]["req_compute_msd"] is True
+    assert task_param["cal_setting"]["timestep"] == 0.001
 
 
 def test_annealing_custom_cal_setting_override():
@@ -127,12 +136,12 @@ def test_annealing_make_confs_writes_task_files_and_variables(tmp_path):
 
     first_meta = loadfn(first_task / "Annealing.json")
     second_meta = loadfn(second_task / "Annealing.json")
-    assert first_meta == {
-        "start_temp": 300.0,
-        "target_temp": 700.0,
-        "end_temp": 400.0,
-        "supercell_size": [2, 3, 4],
-    }
+    assert first_meta["start_temp"] == 300.0
+    assert first_meta["target_temp"] == 700.0
+    assert first_meta["temp"] == 700.0
+    assert first_meta["end_temp"] == 400.0
+    assert first_meta["temp_ramp_rate"] == 1000
+    assert first_meta["supercell_size"] == [2, 3, 4]
     assert second_meta["target_temp"] == 900.0
 
     variable_text = (first_task / "variable_Annealing.in").read_text()
@@ -141,14 +150,23 @@ def test_annealing_make_confs_writes_task_files_and_variables(tmp_path):
     assert "variable nz equal 4" in variable_text
     assert "variable start_temp equal 300.00" in variable_text
     assert "variable target_temp equal 700.00" in variable_text
+    assert "variable temp equal 700.00" in variable_text
     assert "variable end_temp equal 400.00" in variable_text
+    assert "variable temp_ramp_rate equal 1000" in variable_text
     assert "variable equi_step equal 10" in variable_text
     assert "variable ramp_step equal 20" in variable_text
+    assert "variable temp_ramp_step equal 20" in variable_text
     assert "variable hold_step equal 30" in variable_text
     assert "variable cool_step equal 40" in variable_text
+    assert "variable temp_decline_step equal 40" in variable_text
+    assert "variable init_thermo_equil_step equal 10" in variable_text
+    assert "variable final_thermo_equil_step equal 30" in variable_text
     assert "variable rdf_bins equal 64" in variable_text
     assert "variable rdf_cutoff equal 8.0" in variable_text
     assert "variable rdf_interval equal 5" in variable_text
+    assert "variable rdf_nevery equal 5" in variable_text
+    assert "variable rdf_nfreq equal 5" in variable_text
+    assert "variable req_compute_msd equal true" in variable_text
     assert "variable tdamp equal 0.5" in variable_text
     assert "variable pdamp equal 5.0" in variable_text
     assert "variable velocity_seed equal 13579" in variable_text
@@ -242,9 +260,172 @@ def test_annealing_compute_lower_returns_task_notes(tmp_path):
         {},
     )
 
-    assert ptr_data == str(tmp_path) + "\n"
-    assert res_data["task.000000"]["task"] == "task.000000"
-    assert "inspect log.lammps" in res_data["task.000001"]["note"]
+    assert ptr_data.startswith(str(tmp_path) + "\n")
+    assert "task.000000: rdf=[], msd=[], volume_temperature=[]" in ptr_data
+    assert res_data["property"] == "annealing"
+    assert res_data["tasks"]["task.000000"]["task"] == "task.000000"
+    assert res_data["tasks"]["task.000001"]["summary"]["rdf_stages"] == []
+    assert (tmp_path / "result.json").is_file()
+
+
+def write_annealing_analysis_files(task_dir: Path):
+    task_dir.mkdir(parents=True, exist_ok=True)
+    (task_dir / "Annealing.json").write_text(
+        '{"start_temp": 300, "target_temp": 1500, "end_temp": 300}',
+        encoding="utf-8",
+    )
+    (task_dir / "rdf.T_ramp_300K_1500K.txt").write_text(
+        "# Time-averaged data for fix rdf_ramp\n"
+        "# TimeStep Number-of-rows\n"
+        "# Row c_myRDF[1] c_myRDF[2] c_myRDF[3]\n"
+        "0 2\n"
+        "1 0.1 1.0 0.2\n"
+        "2 0.2 2.0 0.4\n"
+        "100 2\n"
+        "1 0.1 1.5 0.3\n"
+        "2 0.2 2.5 0.5\n",
+        encoding="utf-8",
+    )
+    (task_dir / "msd.T_ramp_300K_1500K.txt").write_text(
+        "# Time-averaged data for fix msd_ramp\n"
+        "# TimeStep Number-of-rows\n"
+        "# Row c_myMSD_ramp\n"
+        "0 4\n"
+        "1 0.0\n"
+        "2 0.0\n"
+        "3 0.0\n"
+        "4 0.0\n"
+        "200 4\n"
+        "1 0.1\n"
+        "2 0.2\n"
+        "3 0.3\n"
+        "4 0.6\n",
+        encoding="utf-8",
+    )
+    (task_dir / "heating_interval_2000.dat").write_text(
+        "# Time-averaged data for fix heat_log\n"
+        "# TimeStep v_N v_Temp v_Vatom v_pote v_Etotal v_Press\n"
+        "2000 128 400 15.5 -1 -2 10\n"
+        "4000 128 800 16.0 -1 -2 20\n",
+        encoding="utf-8",
+    )
+    (task_dir / "cooling_interval_2000.dat").write_text(
+        "# Time-averaged data for fix cool_log\n"
+        "# TimeStep v_N v_Temp v_Vatom v_pote v_Etotal v_Press\n"
+        "2000 128 1200 16.5 -1 -2 30\n"
+        "4000 128 600 15.8 -1 -2 40\n",
+        encoding="utf-8",
+    )
+
+
+def test_annealing_compute_lower_extracts_rdf_msd_and_volume_temperature(tmp_path):
+    task_dir = tmp_path / "task.000000"
+    write_annealing_analysis_files(task_dir)
+    prop = Annealing({"type": "annealing"})
+
+    res_data, ptr_data = prop._compute_lower(
+        str(tmp_path / "result.json"),
+        [str(task_dir)],
+        {},
+    )
+
+    task = res_data["tasks"]["task.000000"]
+    assert "T_ramp_300K_1500K" in task["rdf"]
+    assert task["rdf"]["T_ramp_300K_1500K"]["radius"] == [0.1, 0.2]
+    assert task["rdf"]["T_ramp_300K_1500K"]["g_r"] == [1.5, 2.5]
+    assert task["msd"]["T_ramp_300K_1500K"]["msd_total"] == [0.0, 0.6]
+    assert task["volume_temperature"]["heating"]["temperature"] == [400.0, 800.0]
+    assert task["volume_temperature"]["heating"]["total_volume"] == [1984.0, 2048.0]
+    assert "volume_temperature=['cooling', 'heating']" in ptr_data
+    assert loadfn(tmp_path / "result.json")["tasks"]["task.000000"]["summary"]["rdf_points"] == {
+        "T_ramp_300K_1500K": 2
+    }
+
+
+def test_annealing_report_registered_and_builds_graph_table(tmp_path):
+    task_dir = tmp_path / "task.000000"
+    write_annealing_analysis_files(task_dir)
+    prop = Annealing({"type": "annealing"})
+    res_data, _ptr_data = prop._compute_lower(
+        str(tmp_path / "result.json"),
+        [str(task_dir)],
+        {},
+    )
+
+    assert return_prop_type("annealing_00") == "annealing"
+    assert return_prop_class("annealing") is AnnealingReport
+    traces, layout = AnnealingReport.plotly_graph(res_data, "work")
+    table, df = AnnealingReport.dash_table(res_data)
+
+    assert len(traces) == 4
+    assert layout.title.text == "Annealing RDF, MSD, and Volume-Temperature Response"
+    assert "RDF points" in df.columns
+    assert set(df["Stage"]) == {"T_ramp_300K_1500K", "cooling", "heating"}
+    assert table.data
+
+
+def test_annealing_archive_sync_props_extracts_result(tmp_path):
+    prop_dir = tmp_path / "confs" / "mo-bcc" / "annealing_00"
+    prop_dir.mkdir(parents=True)
+    result_payload = {
+        "property": "annealing",
+        "tasks": {
+            "task.000000": {
+                "summary": {
+                    "rdf_stages": ["eq_300K"],
+                    "msd_stages": ["eq_300K"],
+                    "volume_temperature_stages": ["heating"],
+                }
+            }
+        },
+    }
+    param_payload = {"type": "annealing"}
+    dumpfn(result_payload, prop_dir / "result.json")
+    dumpfn(param_payload, prop_dir / "param.json")
+
+    storage = ResultStorage(tmp_path)
+    storage.sync_props(
+        {
+            "structures": ["confs/*"],
+            "interaction": {"type": "deepmd"},
+            "properties": [{"type": "annealing"}],
+        }
+    )
+
+    archived = storage.result_data["confs/mo-bcc"]["annealing_00"]
+    assert archived["parameter"] == param_payload
+    assert archived["result"]["tasks"]["task.000000"]["summary"]["rdf_stages"] == ["eq_300K"]
+
+
+def test_annealing_dash_report_app_builds_graph_and_table(tmp_path):
+    task_dir = tmp_path / "task.000000"
+    write_annealing_analysis_files(task_dir)
+    res_data, _ptr_data = Annealing({"type": "annealing"})._compute_lower(
+        str(tmp_path / "result.json"),
+        [str(task_dir)],
+        {},
+    )
+    app = DashReportApp(
+        {
+            "work": {
+                "conf": {
+                    "annealing_00": {
+                        "result": res_data,
+                    }
+                }
+            }
+        }
+    )
+
+    figure = app.update_graph("annealing_00", "conf")
+    table_container = app.update_table("annealing_00", "conf")
+    table_div = table_container.children[0]
+    table = table_div.children[2]
+
+    assert len(figure.data) == 4
+    assert figure.layout.title.text == "Annealing RDF, MSD, and Volume-Temperature Response"
+    assert table.id == {"type": "table", "index": 0}
+    assert "RDF points" in app.csv_copy(1, table.data)
 
 
 def test_annealing_lammps_input_contains_all_md_stages_and_outputs():
@@ -278,23 +459,25 @@ def test_annealing_lammps_input_contains_all_md_stages_and_outputs():
         prop.task_param()["cal_setting"],
     )
 
-    assert "fix 1 all npt temp ${start_temp} ${start_temp} tdamp_var" in script
-    assert "fix 1 all npt temp ${start_temp} ${target_temp} tdamp_var" in script
-    assert (
-        'if "${hold_step} > 0" then "fix 1 all nvt temp ${target_temp} '
-        '${target_temp} tdamp_var" "run ${hold_step}" "unfix 1"'
-    ) in script
-    assert "fix 1 all npt temp ${target_temp} ${end_temp} tdamp_var" in script
-    assert "run ${equi_step}" in script
-    assert "run ${ramp_step}" in script
-    assert "run ${cool_step}" in script
+    assert "fix eq_nh all npt temp ${start_temp} ${start_temp} tdamp_var" in script
+    assert "fix ramp_nh all npt temp ${start_temp} ${temp} tdamp_var" in script
+    assert "fix decline_nh all npt temp ${temp} ${end_temp} tdamp_var" in script
+    assert "fix final_eq_nh all npt temp ${end_temp} ${end_temp} tdamp_var" in script
+    assert "run ${init_thermo_equil_step}" in script
+    assert "run ${temp_ramp_remain_step}" in script
+    assert "run ${temp_decline_remain_step}" in script
+    assert "run ${final_thermo_equil_remain_step}" in script
+    assert "variable        rdf_comm_cutoff equal ${rdf_cutoff}+2.0" in script
+    assert "comm_modify    cutoff ${rdf_comm_cutoff}" in script
     assert "compute         myRDF all rdf ${rdf_bins} cutoff ${rdf_cutoff}" in script
-    assert "file rdf_ramp.dat mode vector" in script
-    assert "file rdf_cool.dat mode vector" in script
-    assert "file heating_interval.dat" in script
-    assert "file cooling_interval.dat" in script
-    assert "dump.anneal_ramp" in script
-    assert "dump.anneal_cool" in script
+    assert "file rdf.T_ramp_${start_temp}K_${temp}K.txt mode vector" in script
+    assert "file rdf.T_decline_${temp}K_${end_temp}K.txt mode vector" in script
+    assert "file rdf.final_eq_${end_temp}K.txt mode vector" in script
+    assert "file msd.T_ramp_${start_temp}K_${temp}K.txt mode vector" in script
+    assert "file heating_interval_${thermo_interval}.dat" in script
+    assert "file cooling_interval_${thermo_interval}.dat" in script
+    assert "dump.T_ramp_nh_${start_temp}K_${temp}K.*" in script
+    assert "dump.T_decline_nh_${temp}K_${end_temp}K.*" in script
 
 
 class TestAnnealingCoverage(unittest.TestCase):
@@ -324,5 +507,21 @@ def test_annealing_compute_lower_returns_task_notes(self):
         with tempfile.TemporaryDirectory() as tmp:
             test_annealing_compute_lower_returns_task_notes(Path(tmp))
 
+    def test_annealing_compute_lower_extracts_rdf_msd_and_volume_temperature(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            test_annealing_compute_lower_extracts_rdf_msd_and_volume_temperature(Path(tmp))
+
+    def test_annealing_report_registered_and_builds_graph_table(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            test_annealing_report_registered_and_builds_graph_table(Path(tmp))
+
+    def test_annealing_archive_sync_props_extracts_result(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            test_annealing_archive_sync_props_extracts_result(Path(tmp))
+
+    def test_annealing_dash_report_app_builds_graph_and_table(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            test_annealing_dash_report_app_builds_graph_and_table(Path(tmp))
+
     def test_annealing_lammps_input_contains_all_md_stages_and_outputs(self):
         test_annealing_lammps_input_contains_all_md_stages_and_outputs()
diff --git a/tests/test_flow.py b/tests/test_flow.py
index c04009de..104dd2de 100644
--- a/tests/test_flow.py
+++ b/tests/test_flow.py
@@ -148,13 +148,25 @@ def test_flow_static_helpers_and_failure_formatting():
     assert "phase: Failed" in formatted
     assert "main_logs: log-a, log-b" in formatted
     assert "failed_artifacts: debug-artifact" in formatted
-    assert "raw_step:" in formatted
+    assert "raw_step:" not in formatted
 
     assert flow.FlowGenerator._failed_child_ids_from_step(
         {"message": "child 'a' failed", "outboundNodes": ["b"]}
     ) == ["a", "b"]
 
 
+def test_format_step_failure_can_include_raw_step_when_requested(monkeypatch):
+    monkeypatch.setenv("APEX_SHOW_RAW_STEP", "1")
+
+    formatted = flow.FlowGenerator._format_step_failure(
+        {"phase": "Failed", "displayName": "PropsMake", "inputs": {"large": "payload"}},
+        "fallback",
+    )
+
+    assert "raw_step:" in formatted
+    assert "large" in formatted
+
+
 def test_download_artifact_with_retry_retries_transient_and_stops_on_missing(monkeypatch):
     calls = []
 
@@ -216,6 +228,92 @@ def fake_download(artifact, path, retries=3, delay=10):
     assert "main-logs artifact not found" in no_error
 
 
+def test_main_log_download_uses_target_dir_when_download_returns_none(tmp_path, monkeypatch):
+    generator = make_generator(debug_mode=True)
+    generator.download_path = str(tmp_path)
+    step = {
+        "id": "failed",
+        "displayName": "PropsMake",
+        "outputs": {"artifacts": {"main-logs": "main-log-artifact"}},
+    }
+
+    def fake_download(artifact, path, retries=3, delay=10):
+        os.makedirs(path, exist_ok=True)
+        with open(os.path.join(path, "main.log"), "w", encoding="utf-8") as fp:
+            fp.write("ValueError: bad annealing temperature\n")
+        return None
+
+    monkeypatch.setattr(flow.FlowGenerator, "_download_artifact_with_retry", staticmethod(fake_download))
+
+    path, error = generator._download_step_main_logs(step, "prop-key")
+    formatted = flow.FlowGenerator._format_step_failure(
+        {"phase": "Failed", "displayName": "PropsMake"},
+        "property failed",
+        main_log_path=path,
+        main_log_error=error,
+    )
+
+    assert error is None
+    assert path.endswith(os.path.join("main-logs", "prop-key"))
+    assert "cause: ValueError: bad annealing temperature" in formatted
+    assert "main_logs_excerpt:" in formatted
+
+
+def test_format_step_failure_includes_traceback_excerpt(tmp_path):
+    log_dir = tmp_path / "main-logs"
+    log_dir.mkdir()
+    (log_dir / "main.log").write_text(
+        "setup line\n"
+        "Traceback (most recent call last):\n"
+        "  File \"/work/apex/core/property/Phonon.py\", line 360, in make_confs\n"
+        "    raise RuntimeError('phonopy failed')\n"
+        "RuntimeError: phonopy failed\n",
+        encoding="utf-8",
+    )
+
+    formatted = flow.FlowGenerator._format_step_failure(
+        {"phase": "Failed", "displayName": "PropsMake"},
+        "property failed",
+        main_log_path=str(log_dir),
+    )
+
+    assert "main_logs_excerpt:" in formatted
+    assert "cause: RuntimeError: phonopy failed" in formatted
+    assert "Traceback (most recent call last):" in formatted
+    assert "apex/core/property/Phonon.py" in formatted
+    assert "RuntimeError: phonopy failed" in formatted
+
+
+def test_format_step_failure_includes_lammps_diagnostic_excerpt(tmp_path):
+    task_dir = tmp_path / "failed-artifacts" / "prop-key" / "RunLAMMPS" / "backward_dir" / "task.000000"
+    task_dir.mkdir(parents=True)
+    (task_dir / "apex_task_status.json").write_text(
+        '{\n'
+        '    "state": "failed",\n'
+        '    "reason": "command_not_found",\n'
+        '    "exit_code": 127\n'
+        '}\n',
+        encoding="utf-8",
+    )
+    (task_dir / ".debug.log").write_text(
+        "# APEX LAMMPS debug log\n"
+        "## Command\n"
+        "lmp -in in.lammps\n",
+        encoding="utf-8",
+    )
+
+    formatted = flow.FlowGenerator._format_step_failure(
+        {"phase": "Failed", "displayName": "PropsPost"},
+        "property failed",
+        diagnostic_artifacts=[str(tmp_path / "failed-artifacts")],
+    )
+
+    assert "failed_artifacts_excerpt:" in formatted
+    assert "apex_task_status.json" in formatted
+    assert '"reason": "command_not_found"' in formatted
+    assert "lmp -in in.lammps" in formatted
+
+
 def test_diagnostic_artifact_downloads_only_allowed_debug_artifacts(tmp_path, monkeypatch):
     generator = make_generator(debug_mode=True)
     generator.download_path = str(tmp_path)
@@ -442,6 +540,19 @@ def run_with_monkeypatch(self, func):
     def test_flow_static_helpers_and_failure_formatting(self):
         test_flow_static_helpers_and_failure_formatting()
 
+    def test_format_step_failure_can_include_raw_step_when_requested(self):
+        self.run_with_monkeypatch(
+            test_format_step_failure_can_include_raw_step_when_requested
+        )
+
+    def test_format_step_failure_includes_traceback_excerpt(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            test_format_step_failure_includes_traceback_excerpt(Path(tmp))
+
+    def test_format_step_failure_includes_lammps_diagnostic_excerpt(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            test_format_step_failure_includes_lammps_diagnostic_excerpt(Path(tmp))
+
     def test_download_artifact_with_retry_retries_transient_and_stops_on_missing(self):
         self.run_with_monkeypatch(
             test_download_artifact_with_retry_retries_transient_and_stops_on_missing
@@ -452,6 +563,11 @@ def test_step_artifact_lookup_and_main_log_download_from_child(self):
             test_step_artifact_lookup_and_main_log_download_from_child
         )
 
+    def test_main_log_download_uses_target_dir_when_download_returns_none(self):
+        self.run_with_tmp_and_monkeypatch(
+            test_main_log_download_uses_target_dir_when_download_returns_none
+        )
+
     def test_diagnostic_artifact_downloads_only_allowed_debug_artifacts(self):
         self.run_with_tmp_and_monkeypatch(
             test_diagnostic_artifact_downloads_only_allowed_debug_artifacts
diff --git a/tests/test_lammps.py b/tests/test_lammps.py
index abd9d856..e07917c4 100644
--- a/tests/test_lammps.py
+++ b/tests/test_lammps.py
@@ -3,7 +3,9 @@
 import os
 import shutil
 import sys
+import tempfile
 import unittest
+import warnings
 
 import dpdata
 import numpy as np
@@ -128,3 +130,14 @@ def test_backward_files(self):
         ]
         self.assertEqual(self.Lammps.backward_files(), backward_files)
 
+    def test_compute_skips_annealing_without_relax_dump_warning(self):
+        with tempfile.TemporaryDirectory() as tmpdir:
+            dumpfn({"type": "annealing"}, os.path.join(tmpdir, "task.json"))
+            with warnings.catch_warnings(record=True) as caught:
+                warnings.simplefilter("always")
+                result = self.Lammps.compute(tmpdir)
+
+        self.assertIsNone(result)
+        self.assertFalse(
+            any("dump.relax" in str(item.message) for item in caught)
+        )
diff --git a/tests/test_lammps_utils.py b/tests/test_lammps_utils.py
index 5df8d4de..d62f7d5f 100644
--- a/tests/test_lammps_utils.py
+++ b/tests/test_lammps_utils.py
@@ -375,7 +375,8 @@ def test_make_lammps_finite_t_elastic_invalid_role_raises(tmp_path):
 
 def make_annealing_input(cal_setting):
     defaults = {
-        "dump_step": 500,
+        "dump_interval": 500,
+        "thermo_interval": 250,
         "tdamp": "tdamp_var",
         "pdamp": "pdamp_var",
         "velocity_seed": 24680,
@@ -395,27 +396,29 @@ def assert_common_annealing_script(script):
     assert "read_data   conf.lmp" in script
     assert "replicate   ${nx} ${ny} ${nz}" in script
     assert "pair_style dummy" in script
+    assert "variable        rdf_comm_cutoff equal ${rdf_cutoff}+2.0" in script
+    assert "comm_modify    cutoff ${rdf_comm_cutoff}" in script
     assert "compute         myRDF all rdf ${rdf_bins} cutoff ${rdf_cutoff}" in script
-    assert "run ${equi_step}" in script
-    assert "run ${ramp_step}" in script
-    assert "run ${cool_step}" in script
-    assert (
-        'if "${hold_step} > 0" then "fix 1 all nvt temp ${target_temp} '
-        '${target_temp} tdamp_var" "run ${hold_step}" "unfix 1"'
-    ) in script
-    assert "dump            1 all custom  500 dump.anneal_ramp id type xs ys zs fx fy fz" in script
-    assert "dump            2 all custom  500 dump.anneal_cool id type xs ys zs fx fy fz" in script
+    assert "run ${init_thermo_equil_step}" in script
+    assert "run ${temp_ramp_remain_step}" in script
+    assert "run ${temp_decline_remain_step}" in script
+    assert "run ${final_thermo_equil_remain_step}" in script
+    assert "dump eq_nh_dump all atom 500 dump.eq_nh_${start_temp}K.*" in script
+    assert "dump T_ramp_nh_dump all atom 500 dump.T_ramp_nh_${start_temp}K_${temp}K.*" in script
+    assert "dump T_decline_nh_dump all atom 500 dump.T_decline_nh_${temp}K_${end_temp}K.*" in script
+    assert "dump final_eq_nh_dump all atom 500 dump.final_eq_nh_${end_temp}K.*" in script
     assert (
-        "fix rdf_ramp all ave/time ${rdf_interval} 1 ${rdf_interval} c_myRDF[*] "
-        "file rdf_ramp.dat mode vector"
+        "fix rdf_ramp all ave/time 100 1 200 c_myRDF[*] "
+        "file rdf.T_ramp_${start_temp}K_${temp}K.txt mode vector"
     ) in script
     assert (
-        "fix rdf_cool all ave/time ${rdf_interval} 1 ${rdf_interval} c_myRDF[*] "
-        "file rdf_cool.dat mode vector"
+        "fix rdf_decline all ave/time 100 1 200 c_myRDF[*] "
+        "file rdf.T_decline_${temp}K_${end_temp}K.txt mode vector"
     ) in script
-    assert "file heating_interval.dat" in script
-    assert "file cooling_interval.dat" in script
-    assert 'print "All done"' in script
+    assert "fix msd_final_eq all ave/time 100 1 200 c_myMSD_final_eq file msd.final_eq_${end_temp}K.txt mode vector" in script
+    assert "file heating_interval_${thermo_interval}.dat" in script
+    assert "file cooling_interval_${thermo_interval}.dat" in script
+    assert 'print "__end_of_lmp_annealing_calculation__"' in script
 
 
 def test_make_lammps_annealing_default_nose_hoover_npt():
@@ -424,53 +427,56 @@ def test_make_lammps_annealing_default_nose_hoover_npt():
     assert_common_annealing_script(script)
     assert "velocity all create ${start_temp} 24680 mom yes rot yes dist gaussian" in script
     assert (
-        "fix 1 all npt temp ${start_temp} ${start_temp} tdamp_var "
+        "fix eq_nh all npt temp ${start_temp} ${start_temp} tdamp_var "
         "x 0.0 0.0 pdamp_var y 0.0 0.0 pdamp_var z 0.0 0.0 pdamp_var"
     ) in script
     assert (
-        "fix 1 all npt temp ${start_temp} ${target_temp} tdamp_var "
+        "fix ramp_nh all npt temp ${start_temp} ${temp} tdamp_var "
         "x 0.0 0.0 pdamp_var y 0.0 0.0 pdamp_var z 0.0 0.0 pdamp_var"
     ) in script
     assert (
-        "fix 1 all npt temp ${target_temp} ${end_temp} tdamp_var "
+        "fix decline_nh all npt temp ${temp} ${end_temp} tdamp_var "
         "x 0.0 0.0 pdamp_var y 0.0 0.0 pdamp_var z 0.0 0.0 pdamp_var"
     ) in script
-    assert "fix tg all langevin" not in script
+    assert "fix final_eq_nh all npt temp ${end_temp} ${end_temp} tdamp_var" in script
+    assert "_lgv all langevin" not in script
 
 
 def test_make_lammps_annealing_nose_hoover_nvt():
     script = make_annealing_input({"ensemble": "nvt"})
 
     assert_common_annealing_script(script)
-    assert "fix 1 all nvt temp ${start_temp} ${start_temp} tdamp_var" in script
-    assert "fix 1 all nvt temp ${start_temp} ${target_temp} tdamp_var" in script
-    assert "fix 1 all nvt temp ${target_temp} ${end_temp} tdamp_var" in script
-    assert "fix 1 all npt temp" not in script
-    assert "fix tg all langevin" not in script
+    assert "fix eq_nh all nvt temp ${start_temp} ${start_temp} tdamp_var" in script
+    assert "fix ramp_nh all nvt temp ${start_temp} ${temp} tdamp_var" in script
+    assert "fix decline_nh all nvt temp ${temp} ${end_temp} tdamp_var" in script
+    assert "fix final_eq_nh all nvt temp ${end_temp} ${end_temp} tdamp_var" in script
+    assert "all npt temp" not in script
+    assert "_lgv all langevin" not in script
 
 
 def test_make_lammps_annealing_langevin_nph():
     script = make_annealing_input({"thermostat": "langevin", "ensemble": "nph"})
 
     assert_common_annealing_script(script)
-    assert script.count("fix 1 all nph aniso 0.0 0.0 pdamp_var drag 1.0") == 3
-    assert "fix tg all langevin ${start_temp} ${start_temp} tdamp_var 24680" in script
-    assert "fix tg all langevin ${start_temp} ${target_temp} tdamp_var 24680" in script
-    assert "fix tg all langevin ${target_temp} ${end_temp} tdamp_var 24680" in script
-    assert script.count("unfix tg") == 3
-    assert "fix 1 all npt temp" not in script
+    assert script.count("all nph aniso 0.0 0.0 pdamp_var drag 1.0") == 4
+    assert "fix eq_lgv all langevin ${start_temp} ${start_temp} tdamp_var 24680" in script
+    assert "fix ramp_lgv all langevin ${start_temp} ${temp} tdamp_var 24680" in script
+    assert "fix decline_lgv all langevin ${temp} ${end_temp} tdamp_var 24680" in script
+    assert "fix final_eq_lgv all langevin ${end_temp} ${end_temp} tdamp_var 24680" in script
+    assert script.count("_lgv") >= 4
+    assert "all npt temp" not in script
 
 
 def test_make_lammps_annealing_langevin_nve():
     script = make_annealing_input({"thermostat": "langevin", "ensemble": "nve"})
 
     assert_common_annealing_script(script)
-    assert script.count("fix 1 all nve") == 3
-    assert "fix 1 all nph" not in script
-    assert "fix tg all langevin ${start_temp} ${start_temp} tdamp_var 24680" in script
-    assert "fix tg all langevin ${start_temp} ${target_temp} tdamp_var 24680" in script
-    assert "fix tg all langevin ${target_temp} ${end_temp} tdamp_var 24680" in script
-    assert script.count("unfix tg") == 3
+    assert script.count("all nve") == 4
+    assert "all nph" not in script
+    assert "fix eq_lgv all langevin ${start_temp} ${start_temp} tdamp_var 24680" in script
+    assert "fix ramp_lgv all langevin ${start_temp} ${temp} tdamp_var 24680" in script
+    assert "fix decline_lgv all langevin ${temp} ${end_temp} tdamp_var 24680" in script
+    assert "fix final_eq_lgv all langevin ${end_temp} ${end_temp} tdamp_var 24680" in script
 
 
 class TestLammpsUtils(unittest.TestCase):
diff --git a/tests/test_main_workflow_errors.py b/tests/test_main_workflow_errors.py
index f1e568c4..4c469aab 100644
--- a/tests/test_main_workflow_errors.py
+++ b/tests/test_main_workflow_errors.py
@@ -70,6 +70,18 @@ def test_failure_artifact_retrieval_requires_debug_mode(self):
         finally:
             apex_main.config["mode"] = old_mode
 
+    def test_workflow_failure_summary_is_cli_concise_error(self):
+        self.assertTrue(
+            apex_main._is_workflow_failure_summary(
+                RuntimeError("Joint workflow failed with 1 failed step(s):\ndetail")
+            )
+        )
+        self.assertFalse(
+            apex_main._is_workflow_failure_summary(
+                RuntimeError("local parameter parsing failed")
+            )
+        )
+
     def test_formats_dflow_workflow_not_found_error(self):
         message = apex_main._format_workflow_query_error(
             "guipro-joint-svgzz",