fix(ecc): stop recording failed RCX extraction as Success

.gitignore

@@ -182,6 +182,7 @@ chipcompiler/tools/ecc_dreamplace/dreamplace
 
 .humanize/
 humanize-*
+.zcode/
 docs/superpowers/
 findings.md
 progress.md

chipcompiler/engine/flow.py

@@ -304,11 +304,10 @@ def check_step_result(self, workspace_step: WorkspaceStep):
                 ):
                     success = True
             case StepEnum.RCX.value:
-                success = True
-                for spef in ecc_output.spef if ecc_output else []:
-                    if not os.path.exists(spef):
-                        success = False
-                        break
+                spef_list = ecc_output.spef if ecc_output else []
+                success = bool(spef_list) and all(
+                    os.path.isfile(spef) and os.path.getsize(spef) > 0 for spef in spef_list
+                )
             case StepEnum.TIMING_OPT.value:
                 if os.path.exists(output.def_ or "") and os.path.exists(output.verilog or ""):
                     success = True

chipcompiler/engine/step_execution.py

@@ -4,7 +4,7 @@
 from dataclasses import dataclass
 from threading import Event, Thread
 
-from chipcompiler.data import Workspace, WorkspaceStep
+from chipcompiler.data import StateEnum, Workspace, WorkspaceStep
 from chipcompiler.engine.db import EngineDB
 from chipcompiler.utility.log import capture_stdio_to_file, flush_cstdio
 
@@ -87,6 +87,8 @@ def execute_tool_step(
                     ecc_module=engine_db.engine,
                 )
                 workspace.logger.info(f"[STEP] {step_tag} finished result={result}")
+                if result is not True and result is not StateEnum.Success:
+                    step_error = f"{step_tag} reported failure (run_step returned {result!r})."
             except (Exception, SystemExit) as exc:
                 step_error = record_tool_failure(workspace.logger, step_tag, exc)
     except (Exception, SystemExit) as exc:

chipcompiler/tools/ecc/rcx_artifacts.py

@@ -0,0 +1,126 @@
+"""Publication policy for RCX extraction artifacts.
+
+iRCX writes extraction results to ``<data_dir>/spef_writer``. This module
+resolves the RCX step directories, clears stale extraction artifacts before a
+run, and publishes the fresh SPEF set to the step output directory
+all-or-nothing: a failed pass restores the previously published SPEFs instead
+of destroying them, and a successful pass leaves no backups behind.
+"""
+
+import os
+import shutil
+from pathlib import Path
+
+from chipcompiler.data import EccStep, Workspace
+
+
+def _workspace_rcx_dir(path_text: str, workspace_dir: Path) -> Path:
+    if path_text.startswith("/"):
+        relative_path = path_text[1:]
+        if relative_path.split("/", 1)[0] in ("RCX_ecc", "rcx_ecc"):
+            return workspace_dir / relative_path
+    return Path(path_text)
+
+
+def resolve_rcx_dirs(workspace: Workspace, step: EccStep) -> tuple[Path | None, Path | None]:
+    """Resolve the RCX extraction data and output directories for a step."""
+    workspace_dir = workspace.directory
+    if workspace_dir is None:
+        return None, None
+
+    data_dir_text = os.fspath(step.data.dir or "")
+    output_dir_text = os.fspath(step.output.dir or "")
+    if not data_dir_text or not output_dir_text:
+        return None, None
+
+    return (
+        _workspace_rcx_dir(data_dir_text, workspace_dir),
+        _workspace_rcx_dir(output_dir_text, workspace_dir),
+    )
+
+
+def wipe_stale_spef_artifacts(data_dir: Path) -> None:
+    for stale_path in (data_dir / "spef_writer").glob("*.spef"):
+        stale_path.unlink(missing_ok=True)
+
+
+def copy_rcx_spef_outputs(workspace: Workspace, step: EccStep) -> bool:
+    data_dir, output_dir = resolve_rcx_dirs(workspace, step)
+    if data_dir is None or output_dir is None:
+        workspace.logger.error("RCX data or output directory is not configured")
+        return False
+
+    spef_writer_dir = data_dir / "spef_writer"
+    if not spef_writer_dir.is_dir():
+        workspace.logger.error("RCX extraction artifacts are missing: %s", spef_writer_dir)
+        return False
+
+    declared_paths = [spef_path for spef_path in step.output.spef if spef_path]
+    output_paths = [
+        output_dir / spef_path.name
+        for spef_path in (declared_paths or sorted(spef_writer_dir.glob("*.spef")))
+    ]
+
+    if not output_paths:
+        workspace.logger.error("RCX extraction produced no SPEF artifacts to publish")
+        return False
+
+    # Validate the whole source set before touching any destination, so a
+    # partially copied set or a stale destination left by an earlier run can
+    # never pass for fresh extraction output.
+    for output_path in output_paths:
+        source_path = spef_writer_dir / output_path.name
+        if not (source_path.is_file() and os.path.getsize(source_path) > 0):
+            workspace.logger.error("RCX extraction artifact is missing or empty: %s", source_path)
+            return False
+
+    processed: list[tuple[Path, Path | None]] = []
+    temp_paths: list[Path] = []
+    try:
+        for output_path in output_paths:
+            source_path = spef_writer_dir / output_path.name
+            output_path.parent.mkdir(parents=True, exist_ok=True)
+            backup_path = (
+                output_path.with_name(f".{output_path.name}.prev") if output_path.exists() else None
+            )
+            if backup_path is not None:
+                # Preserve the previously published SPEF so a failure later
+                # in this pass can restore the last known-good artifact.
+                output_path.replace(backup_path)
+            temp_path = output_path.with_name(f".{output_path.name}.tmp")
+            processed.append((output_path, backup_path))
+            temp_paths.append(temp_path)
+            shutil.copy2(source_path, temp_path)
+            temp_path.replace(output_path)
+            workspace.logger.info("Copied RCX SPEF %s to %s", source_path, output_path)
+
+        for output_path in output_paths:
+            if not (os.path.isfile(output_path) and os.path.getsize(output_path) > 0):
+                raise OSError(f"Published RCX SPEF is missing or empty: {output_path}")
+    except Exception as exc:
+        # Publication is all-or-nothing: a failed pass restores the SPEFs it
+        # replaced, removes the destinations it created, and drops partial
+        # temporary copies, leaving the step outputs exactly as before.
+        for output_path, backup_path in processed:
+            if backup_path is not None:
+                backup_path.replace(output_path)
+            else:
+                output_path.unlink(missing_ok=True)
+        for temp_path in temp_paths:
+            temp_path.unlink(missing_ok=True)
+        workspace.logger.error("Failed to publish RCX SPEF artifacts: %s", exc)
+        return False
+
+    # The pass is committed: drop the preserved copies so a successful rerun
+    # does not accumulate a second SPEF set on disk. A removal failure must
+    # not fail the already-published step, so it is only reported.
+    for _output_path, backup_path in processed:
+        if backup_path is not None:
+            try:
+                backup_path.unlink(missing_ok=True)
+            except OSError as exc:
+                workspace.logger.warning("Failed to remove RCX SPEF backup: %s", exc)
+
+    if isinstance(step.output.spef, list):
+        step.output.spef[:] = output_paths
+    return True

chipcompiler/tools/ecc/runner.py

@@ -18,6 +18,11 @@
     save_rcx_spef_feature_facts,
 )
 from chipcompiler.tools.ecc.module import ECCToolsModule
+from chipcompiler.tools.ecc.rcx_artifacts import (
+    copy_rcx_spef_outputs,
+    resolve_rcx_dirs,
+    wipe_stale_spef_artifacts,
+)
 from chipcompiler.tools.ecc.sta_artifacts import discard_sta_outputs
 from chipcompiler.tools.ecc.sta_qor import (
     POST_SYNTHESIS_STA_CORNER,
@@ -65,49 +70,6 @@ def _workspace_sta_config_path(workspace: Workspace) -> str | None:
     return os.fspath(config_path) if config_path is not None else None
 
 
-def copy_rcx_spef_outputs(workspace: Workspace, step: EccStep):
-    data_dir_text = os.fspath(step.data.dir or "")
-    output_dir_text = os.fspath(step.output.dir or "")
-    workspace_dir = workspace.directory
-    if not data_dir_text or not output_dir_text or workspace_dir is None:
-        return
-
-    data_dir = Path(data_dir_text)
-    if data_dir_text.startswith("/"):
-        relative_data_dir = data_dir_text[1:]
-        if relative_data_dir.split("/", 1)[0] in ("RCX_ecc", "rcx_ecc"):
-            data_dir = workspace_dir / relative_data_dir
-
-    output_dir = Path(output_dir_text)
-    if output_dir_text.startswith("/"):
-        relative_output_dir = output_dir_text[1:]
-        if relative_output_dir.split("/", 1)[0] in ("RCX_ecc", "rcx_ecc"):
-            output_dir = workspace_dir / relative_output_dir
-
-    spef_writer_dir = data_dir / "spef_writer"
-    if not spef_writer_dir.is_dir():
-        return
-
-    output_paths = [output_dir / spef_path.name for spef_path in step.output.spef if spef_path]
-
-    if not output_paths:
-        output_paths = [
-            output_dir / spef_path.name for spef_path in sorted(spef_writer_dir.glob("*.spef"))
-        ]
-
-    for output_path in output_paths:
-        source_path = spef_writer_dir / output_path.name
-        if not source_path.is_file():
-            continue
-
-        output_path.parent.mkdir(parents=True, exist_ok=True)
-        shutil.copy2(source_path, output_path)
-        workspace.logger.info("Copied RCX SPEF %s to %s", source_path, output_path)
-
-    if isinstance(step.output.spef, list):
-        step.output.spef[:] = output_paths
-
-
 def copy_lvs_outputs(workspace: Workspace, step: EccStep):
     output_dir_text = os.fspath((step.data.steps or {}).get(StepEnum.LVS.value, ""))
     if not output_dir_text:
@@ -841,12 +803,37 @@ def run_rcx(workspace: Workspace, step: EccStep, ecc_module: ECCToolsModule | No
     if ecc_module is not None:
         sub_flow.update_step(step_name=EccSubFlowEnum.load_data.value, state=StateEnum.Success)
 
-        ecc_module.init_rcx(
-            config=workspace.config.get(StepEnum.RCX.value, ""), pdk=workspace.pdk.name
-        )
-        ecc_module.run_rcx()
-        ecc_module.destroy_rcx()
-        copy_rcx_spef_outputs(workspace, step)
+        # A rerun keeps the step directory, so drop previously extracted SPEFs
+        # first; stale artifacts left behind by an earlier run must not be
+        # mistaken for fresh extraction output.
+        data_dir, _ = resolve_rcx_dirs(workspace, step)
+        if data_dir is not None:
+            wipe_stale_spef_artifacts(data_dir)
+
+        try:
+            if not ecc_module.init_rcx(
+                config=workspace.config.get(StepEnum.RCX.value, ""), pdk=workspace.pdk.name
+            ):
+                workspace.logger.error("Failed to initialize RCX extraction")
+                sub_flow.update_step(
+                    step_name=EccSubFlowEnum.run_rcx.value, state=StateEnum.Imcomplete
+                )
+                return False
+            if not ecc_module.run_rcx():
+                workspace.logger.error("RCX extraction failed")
+                sub_flow.update_step(
+                    step_name=EccSubFlowEnum.run_rcx.value, state=StateEnum.Imcomplete
+                )
+                return False
+        finally:
+            try:
+                ecc_module.destroy_rcx()
+            except Exception as exc:
+                workspace.logger.error("Failed to release the RCX extractor: %s", exc)
+
+        if not copy_rcx_spef_outputs(workspace, step):
+            sub_flow.update_step(step_name=EccSubFlowEnum.run_rcx.value, state=StateEnum.Imcomplete)
+            return False
         sub_flow.update_step(step_name=EccSubFlowEnum.run_rcx.value, state=StateEnum.Success)
 
         if not save_data(