Add more detailed failure statuses

Author: yingyue2030699

.gitignore

@@ -208,3 +208,5 @@ test_parse_8ip3.py
 /pdb_benchmark
 /7cqv_dir
 /tests/troubleshoot/8IP3_validation
+/ionerdss/benchmark/trials
+/ionerdss/benchmark_output

benchmark/run_validation_benchmark.py

@@ -13,13 +13,71 @@
 import logging
 import traceback
 from pathlib import Path
+from typing import Optional
 
 from ionerdss.model.pdb import PDBModelBuilder
 from ionerdss.model import pdb
+from ionerdss.model.pdb.structure_validation import get_disconnected_design_message
 
 FAST_VALIDATION_ITERATIONS = 100000
 
 
+def _classify_failed_assembly(sim_result, target_assembly_size: int) -> str:
+    """Return UA when the largest observed assembly is smaller than target, otherwise OA."""
+    if sim_result.largest_observed_assembly_size < target_assembly_size:
+        return "UA"
+    return "OA"
+
+
+def _contains_nerdss_crash_signature(message: Optional[str]) -> bool:
+    """Return True when the text clearly indicates a NERDSS runtime crash."""
+    if not message:
+        return False
+
+    normalized = message.lower()
+    crash_signatures = (
+        "segmentation fault",
+        "segfault",
+        "core dumped",
+        "signal 11",
+        "sigsegv",
+        "abort trap",
+        "stack trace",
+    )
+    return any(signature in normalized for signature in crash_signatures)
+
+
+def _detect_nerdss_crash(sim_result) -> bool:
+    """Return True when the simulation result includes clear evidence of a NERDSS crash."""
+    if _contains_nerdss_crash_signature(getattr(sim_result, "warning_message", None)):
+        return True
+
+    simulation_dir = getattr(sim_result, "simulation_dir", None)
+    if simulation_dir is None:
+        return False
+
+    log_path = Path(simulation_dir) / "output.log"
+    if not log_path.exists():
+        return False
+
+    try:
+        return _contains_nerdss_crash_signature(log_path.read_text(encoding="utf-8", errors="replace"))
+    except OSError:
+        return False
+
+
+def _status_for_failed_validation(sim_result, target_assembly_size: int) -> str:
+    """Prefer specific failure codes and fall back to the older ambiguous status when needed."""
+    if _detect_nerdss_crash(sim_result):
+        return "NC"
+
+    largest_observed_assembly_size = getattr(sim_result, "largest_observed_assembly_size", None)
+    if largest_observed_assembly_size is not None:
+        return _classify_failed_assembly(sim_result, target_assembly_size)
+
+    return "Failed_Assembly"
+
+
 def _run_validation_attempt(
     system,
     workspace_manager,
@@ -136,6 +194,20 @@ def main():
             chain_types_count = len(rep_instances)
 
             print(f"  -> Extracted {chains_count} chains across {chain_types_count} unique types.")
+
+            if chains_count < 2:
+                status = "FP"
+                print("  -> Too few protein chains remain after coarse-graining; marking as FP.")
+                continue
+
+            disconnected_design_message = get_disconnected_design_message(
+                system,
+                prefix="Validation preflight warning",
+            )
+            if disconnected_design_message is not None:
+                status = "DC"
+                print(f"  -> {disconnected_design_message}")
+                continue
 
             # Create a range of titration rates, scaled for each molecular species
             base_rate = 0.0 # 0.25e-3
@@ -187,20 +259,23 @@ def main():
                 print(f"  -> Validation Complete! RMSD: {rmsd:.4f} nm")
             else:
                 print("  -> Simulation ran but did not yield a full matching assembly.")
+                target_assembly_size = sum(artifacts.target_counts.values())
+                status = _status_for_failed_validation(sim_result, target_assembly_size)
                 if sim_result.warning_message:
                     print(f"     Warning: {sim_result.warning_message}")
                     if "[Errno 2] No such file" in sim_result.warning_message or "invalid literal format" in sim_result.warning_message or "invalid literal for int" in sim_result.warning_message:
                         status = "Crashed"
-                    else:
-                        status = "Failed_Assembly"
-                        print("     Error: target composition did not appear in the histogram after both validation runs.")
-                else:
-                    status = "Failed_Assembly"
+                    elif status in {"UA", "OA"}:
+                        print(
+                            "     Error: target assembly was not found after both validation runs. "
+                            f"Largest observed assembly size was {sim_result.largest_observed_assembly_size} "
+                            f"vs target size {target_assembly_size}."
+                        )
 
                 _dump_nerdss_log(Path(sim_result.simulation_dir), pdb_id)
 
         except Exception as e:
-            status = "Crashed"
+            status = "NC" if _contains_nerdss_crash_signature(str(e)) else "Crashed"
             print(f"  -> Error encountered benchmarking {pdb_id}: {e}")
             if "DATA/restart.dat or any RESTART snapshot" in str(e):
                 print("     Error: target composition appeared in the histogram, but no restart snapshot contained the full assembly.")

ionerdss/model/pdb/main.py

@@ -11,6 +11,7 @@
 from pathlib import Path
 import logging
 import math
+import warnings
 import numpy as np
 
 from ionerdss.model.components.system import System
@@ -241,10 +242,11 @@ def notice(self, message, *args, **kwargs):
 
             disconnected_design_message = get_disconnected_design_message(
                 system,
-                prefix="Preflight error",
+                prefix="Preflight warning",
             )
             if disconnected_design_message is not None:
-                raise ValueError(disconnected_design_message)
+                self.workspace_manager.logger.warning(disconnected_design_message)
+                warnings.warn(disconnected_design_message, RuntimeWarning)
 
             # Calculate default molecule counts if not provided (using stoichiometry)
             if molecule_counts is None:

ionerdss/model/pdb/structure_validation.py

@@ -80,6 +80,7 @@ class StructureValidationSimulationResult:
     full_assembly_found: bool
     warning_message: Optional[str]
     first_full_assembly_time: Optional[float]
+    largest_observed_assembly_size: int
     observed_coordinates: Optional[Dict[str, Tuple[float, float, float]]]
 
 
@@ -1013,6 +1014,75 @@ def _find_observed_com_coordinates_in_restart_snapshots(
     )
 
 
+def _get_largest_complex_json_size(complex_json_file: Union[str, Path]) -> int:
+    """Return the largest complex size present in a COMPLEXES JSON snapshot."""
+    payload = json.loads(Path(complex_json_file).read_text(encoding="utf-8"))
+    if not isinstance(payload, list):
+        raise ValueError(f"Malformed COMPLEXES JSON snapshot: {complex_json_file}")
+
+    largest_size = 0
+    for complex_record in payload:
+        if not isinstance(complex_record, dict):
+            continue
+        names = complex_record.get("names")
+        coords = complex_record.get("coords")
+        if not isinstance(names, list) or not isinstance(coords, list) or len(names) != len(coords):
+            continue
+        largest_size = max(largest_size, len(names))
+
+    return largest_size
+
+
+def _get_largest_complex_json_size_in_snapshots(complexes_dir: Union[str, Path]) -> int:
+    """Return the largest complex size seen across COMPLEXES JSON snapshots."""
+    largest_size = 0
+    for candidate_json in _iter_complex_json_candidates(complexes_dir):
+        try:
+            largest_size = max(largest_size, _get_largest_complex_json_size(candidate_json))
+        except Exception:
+            continue
+    return largest_size
+
+
+def _get_largest_restart_component_size(restart_file: Union[str, Path]) -> int:
+    """Return the largest connected component size in a restart snapshot."""
+    adjacency, restart_coords, _restart_mol_names = _parse_restart_snapshot(restart_file)
+    for mol_id in restart_coords:
+        adjacency.setdefault(mol_id, set())
+
+    largest_size = 0
+    visited: set[int] = set()
+    for mol_id in sorted(restart_coords):
+        if mol_id in visited:
+            continue
+
+        stack = [mol_id]
+        visited.add(mol_id)
+        component_size = 0
+        while stack:
+            current = stack.pop()
+            component_size += 1
+            for neighbor in sorted(adjacency.get(current, ())):
+                if neighbor in restart_coords and neighbor not in visited:
+                    visited.add(neighbor)
+                    stack.append(neighbor)
+
+        largest_size = max(largest_size, component_size)
+
+    return largest_size
+
+
+def _get_largest_restart_component_size_in_snapshots(primary_restart_file: Union[str, Path]) -> int:
+    """Return the largest connected component size seen across DATA and RESTART snapshots."""
+    largest_size = 0
+    for candidate_restart in _iter_restart_snapshot_candidates(primary_restart_file):
+        try:
+            largest_size = max(largest_size, _get_largest_restart_component_size(candidate_restart))
+        except Exception:
+            continue
+    return largest_size
+
+
 def run_structure_validation_simulation(
     artifacts: StructureValidationArtifacts,
     nerdss_dir: Union[str, Path],
@@ -1065,8 +1135,10 @@ def run_structure_validation_simulation(
     warning_message = None
     observed_coordinates = None
     selected_restart_file = restart_file
+    largest_observed_assembly_size = 0
     complex_json_candidates = _iter_complex_json_candidates(complexes_dir)
     if complex_json_candidates:
+        largest_observed_assembly_size = _get_largest_complex_json_size_in_snapshots(complexes_dir)
         try:
             observed_coordinates, selected_restart_file = _find_observed_com_coordinates_in_complex_json_snapshots(
                 complexes_dir=complexes_dir,
@@ -1086,6 +1158,7 @@ def run_structure_validation_simulation(
         )
         warnings.warn(fallback_warning, RuntimeWarning)
         warning_message = fallback_warning
+        largest_observed_assembly_size = _get_largest_restart_component_size_in_snapshots(restart_file)
         try:
             observed_coordinates, selected_restart_file = _find_observed_com_coordinates_in_restart_snapshots(
                 system_psf_file=system_psf_file,
@@ -1122,5 +1195,6 @@ def run_structure_validation_simulation(
         full_assembly_found=full_assembly_found,
         warning_message=warning_message,
         first_full_assembly_time=first_full_assembly_time,
+        largest_observed_assembly_size=largest_observed_assembly_size,
         observed_coordinates=observed_coordinates,
     )

ionerdss/tests/test_structure_validation.py

@@ -452,7 +452,7 @@ def get_system(self):
 
     builder = PDBModelBuilder("test_input.pdb")
 
-    with pytest.raises(ValueError, match="Preflight error: the designed assembly graph is disconnected"):
+    with pytest.warns(RuntimeWarning, match="Preflight warning: the designed assembly graph is disconnected"):
         builder.build_system(
             workspace_path=str(tmp_path / "workspace"),
             generate_visualizations=False,

ionerdss/tests/test_validation_benchmark_status.py

@@ -0,0 +1,71 @@
+from importlib.util import module_from_spec, spec_from_file_location
+from pathlib import Path
+from types import SimpleNamespace
+
+
+BENCHMARK_SCRIPT = Path(__file__).resolve().parents[2] / "benchmark" / "run_validation_benchmark.py"
+_SPEC = spec_from_file_location("run_validation_benchmark", BENCHMARK_SCRIPT)
+assert _SPEC is not None and _SPEC.loader is not None
+benchmark = module_from_spec(_SPEC)
+_SPEC.loader.exec_module(benchmark)
+
+
+def test_status_for_failed_validation_returns_ua_when_largest_assembly_is_too_small():
+    sim_result = SimpleNamespace(
+        largest_observed_assembly_size=3,
+        warning_message="Validation warning: no full assembly matching target.",
+        simulation_dir=Path("/tmp/does_not_exist"),
+    )
+
+    assert benchmark._status_for_failed_validation(sim_result, target_assembly_size=4) == "UA"
+
+
+def test_status_for_failed_validation_returns_oa_when_largest_assembly_meets_or_exceeds_target():
+    sim_result = SimpleNamespace(
+        largest_observed_assembly_size=4,
+        warning_message="Validation warning: no full assembly matching target.",
+        simulation_dir=Path("/tmp/does_not_exist"),
+    )
+
+    assert benchmark._status_for_failed_validation(sim_result, target_assembly_size=4) == "OA"
+
+
+def test_status_for_failed_validation_returns_nc_when_warning_contains_crash_signature():
+    sim_result = SimpleNamespace(
+        largest_observed_assembly_size=2,
+        warning_message="NERDSS terminated with Segmentation fault (core dumped).",
+        simulation_dir=Path("/tmp/does_not_exist"),
+    )
+
+    assert benchmark._status_for_failed_validation(sim_result, target_assembly_size=4) == "NC"
+
+
+def test_status_for_failed_validation_returns_nc_when_output_log_contains_crash_signature(tmp_path):
+    simulation_dir = tmp_path / "validation_output" / "1"
+    simulation_dir.mkdir(parents=True)
+    (simulation_dir / "output.log").write_text(
+        "fatal error: abort trap\nSegmentation fault\n",
+        encoding="utf-8",
+    )
+    sim_result = SimpleNamespace(
+        largest_observed_assembly_size=2,
+        warning_message="Validation warning: simulation did not produce a full assembly.",
+        simulation_dir=simulation_dir,
+    )
+
+    assert benchmark._status_for_failed_validation(sim_result, target_assembly_size=4) == "NC"
+
+
+def test_status_for_failed_validation_falls_back_to_old_failed_assembly_label_without_size_metadata():
+    sim_result = SimpleNamespace(
+        warning_message="Validation warning: simulation did not produce a full assembly.",
+        simulation_dir=Path("/tmp/does_not_exist"),
+    )
+
+    assert benchmark._status_for_failed_validation(sim_result, target_assembly_size=4) == "Failed_Assembly"
+
+
+def test_contains_nerdss_crash_signature_matches_common_runtime_crash_messages():
+    assert benchmark._contains_nerdss_crash_signature("Segmentation fault (core dumped)")
+    assert benchmark._contains_nerdss_crash_signature("Received signal 11 while running NERDSS")
+    assert not benchmark._contains_nerdss_crash_signature("Validation warning: no full assembly was found")