From 58e2a7472ea2d4e316d51ca33edd4d5f73695a9f Mon Sep 17 00:00:00 2001 From: Emin Date: Sun, 6 Sep 2026 00:10:37 +0800 Subject: [PATCH 01/10] chore: add .zcode/ to .gitignore Signed-off-by: Emin --- .gitignore | 1 + 1 file changed, 1 insertion(+) diff --git a/.gitignore b/.gitignore index 4cb575680..9f761a358 100644 --- a/.gitignore +++ b/.gitignore @@ -182,6 +182,7 @@ chipcompiler/tools/ecc_dreamplace/dreamplace .humanize/ humanize-* +.zcode/ docs/superpowers/ findings.md progress.md From d2870139dce67930b2269945c1feb553e2ae78b9 Mon Sep 17 00:00:00 2001 From: Emin Date: Sun, 6 Sep 2026 00:26:57 +0800 Subject: [PATCH 02/10] fix(ecc): stop recording failed RCX extraction as Success A failed iRCX run could still be recorded as a successful flow step: native init_rcx()/run_rcx() return values were dropped, SPEF publication silently no-op'd, and the RCX output verdict treated an empty spef list as success. - run_rcx propagates native init_rcx()/run_rcx() False returns and marks the run_rcx substep Incomplete, mirroring the run_sta failure paths - destroy_rcx() is attempted on every path once init_rcx() is entered - stale spef_writer/*.spef artifacts are wiped before extraction so a rerun after a past success cannot glob stale outputs - copy_rcx_spef_outputs returns a boolean and publishes all-or-nothing: no fresh SPEF, a missing spef_writer dir, or a missing/zero-byte published SPEF fails - check_step_result requires a non-empty spef list with every entry an existing, non-zero-byte file Refs #229 --- chipcompiler/engine/flow.py | 7 +- chipcompiler/tools/ecc/runner.py | 104 +++++++++++++----- test/test_engine_flow.py | 31 +++++- test/tools/ecc/test_runner.py | 179 ++++++++++++++++++++++++++++++- 4 files changed, 289 insertions(+), 32 deletions(-) diff --git a/chipcompiler/engine/flow.py b/chipcompiler/engine/flow.py index 1f61036f8..02b86e5b7 100644 --- a/chipcompiler/engine/flow.py +++ b/chipcompiler/engine/flow.py @@ -304,9 +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): + spef_list = ecc_output.spef if ecc_output else [] + success = bool(spef_list) + for spef in spef_list: + if not (os.path.isfile(spef) and os.path.getsize(spef) > 0): success = False break case StepEnum.TIMING_OPT.value: diff --git a/chipcompiler/tools/ecc/runner.py b/chipcompiler/tools/ecc/runner.py index 8d4dd9c03..af1b2bdcc 100644 --- a/chipcompiler/tools/ecc/runner.py +++ b/chipcompiler/tools/ecc/runner.py @@ -65,36 +65,56 @@ 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): +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 "") - workspace_dir = workspace.directory - if not data_dir_text or not output_dir_text or workspace_dir is None: - return + 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), + ) - 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 +def _wipe_stale_rcx_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(): - return - - output_paths = [output_dir / spef_path.name for spef_path in step.output.spef if spef_path] + workspace.logger.error("RCX extraction artifacts are missing: %s", spef_writer_dir) + return False - if not output_paths: + declared_paths = [spef_path for spef_path in step.output.spef if spef_path] + if declared_paths: + output_paths = [output_dir / spef_path.name for spef_path in declared_paths] + else: output_paths = [ output_dir / spef_path.name for spef_path in sorted(spef_writer_dir.glob("*.spef")) ] + copied = False for output_path in output_paths: source_path = spef_writer_dir / output_path.name if not source_path.is_file(): @@ -102,10 +122,21 @@ def copy_rcx_spef_outputs(workspace: Workspace, step: EccStep): output_path.parent.mkdir(parents=True, exist_ok=True) shutil.copy2(source_path, output_path) + copied = True workspace.logger.info("Copied RCX SPEF %s to %s", source_path, output_path) + if not copied: + workspace.logger.error("RCX extraction produced no SPEF artifacts to publish") + return False + + for output_path in output_paths: + if not (os.path.isfile(output_path) and os.path.getsize(output_path) > 0): + workspace.logger.error("Published RCX SPEF is missing or empty: %s", output_path) + return False + if isinstance(step.output.spef, list): step.output.spef[:] = output_paths + return True def copy_lvs_outputs(workspace: Workspace, step: EccStep): @@ -841,12 +872,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_rcx_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( diff --git a/test/test_engine_flow.py b/test/test_engine_flow.py index 6f2d78dd2..d3f9efb61 100644 --- a/test/test_engine_flow.py +++ b/test/test_engine_flow.py @@ -332,20 +332,43 @@ def test_rcx_fails_when_spef_missing(self, tmp_path): def test_rcx_succeeds_when_all_spef_exist(self, tmp_path): spef1 = tmp_path / "corner1.spef" spef2 = tmp_path / "corner2.spef" - spef1.write_text("") - spef2.write_text("") + spef1.write_text("*SPEF\n") + spef2.write_text("*SPEF\n") step = EccStep( name=StepEnum.RCX.value, output=EccOutput(spef=[spef1, spef2]), ) assert EngineFlow(Workspace()).check_step_result(step) is True - def test_rcx_succeeds_with_empty_spef_list(self): + def test_rcx_fails_with_empty_spef_list(self): step = EccStep( name=StepEnum.RCX.value, output=EccOutput(spef=[]), ) - assert EngineFlow(Workspace()).check_step_result(step) is True + assert EngineFlow(Workspace()).check_step_result(step) is False + + def test_rcx_fails_when_spef_is_zero_byte(self, tmp_path): + spef = tmp_path / "corner1.spef" + spef.write_text("") + step = EccStep( + name=StepEnum.RCX.value, + output=EccOutput(spef=[spef]), + ) + assert EngineFlow(Workspace()).check_step_result(step) is False + + def test_rcx_fails_when_one_of_two_spef_missing(self, tmp_path): + spef1 = tmp_path / "corner1.spef" + spef1.write_text("*SPEF\n") + spef2 = tmp_path / "corner2.spef" + step = EccStep( + name=StepEnum.RCX.value, + output=EccOutput(spef=[spef1, spef2]), + ) + assert EngineFlow(Workspace()).check_step_result(step) is False + + def test_rcx_fails_when_output_is_not_ecc_output(self): + step = YosysStep(name=StepEnum.RCX.value, tool="yosys") + assert EngineFlow(Workspace()).check_step_result(step) is False class TestStepExceptionForcesIncomplete: diff --git a/test/tools/ecc/test_runner.py b/test/tools/ecc/test_runner.py index ccb9a7ed7..ea2df6063 100644 --- a/test/tools/ecc/test_runner.py +++ b/test/tools/ecc/test_runner.py @@ -13,6 +13,7 @@ EccStep, OriginDesign, Parameters, + StateEnum, StepEnum, StepInput, Workspace, @@ -21,6 +22,7 @@ from chipcompiler.tools.ecc import runner as ecc_runner from chipcompiler.tools.ecc.builder import build_step, build_step_space from chipcompiler.tools.ecc.checklist import EccRcxChecklist +from chipcompiler.tools.ecc.subflow import EccSubFlowEnum class FakeEccModule: @@ -86,6 +88,7 @@ class FakeLogger: def __init__(self): self.infos = [] self.warnings = [] + self.errors = [] def info(self, message, *args): self.infos.append((message, args)) @@ -93,6 +96,9 @@ def info(self, message, *args): def warning(self, message, *args): self.warnings.append((message, args)) + def error(self, message, *args): + self.errors.append((message, args)) + class FakeSubFlow: def __init__(self, *args, **kwargs): @@ -125,6 +131,27 @@ def test_run_analysis_switch(parameters, expected_calls, tmp_path, monkeypatch): assert checklist.return_value.check.call_count == expected_calls +class FakeRcxModule: + def __init__(self, *, init_ok=True, run_ok=True): + self.calls = [] + self.init_ok = init_ok + self.run_ok = run_ok + + def update_step_paths(self, **kwargs): + self.calls.append(("update_step_paths", kwargs)) + + def init_rcx(self, **kwargs): + self.calls.append(("init_rcx", kwargs)) + return self.init_ok + + def run_rcx(self): + self.calls.append(("run_rcx",)) + return self.run_ok + + def destroy_rcx(self): + self.calls.append(("destroy_rcx",)) + + class FakeCtsModule: def __init__(self, timing_quality): self.calls = [] @@ -649,7 +676,7 @@ def test_copy_rcx_spef_outputs_publishes_to_step_output_dir(tmp_path): ) workspace = Workspace(directory=tmp_path, logger=FakeLogger()) - ecc_runner.copy_rcx_spef_outputs(workspace, step) + assert ecc_runner.copy_rcx_spef_outputs(workspace, step) is True destination = output_dir / source_path.name assert destination.read_text(encoding="utf-8") == "*SPEF\n" @@ -658,6 +685,156 @@ def test_copy_rcx_spef_outputs_publishes_to_step_output_dir(tmp_path): assert step.output.spef == [destination] +def test_copy_rcx_spef_outputs_fails_when_declared_spef_source_is_missing(tmp_path): + data_dir = tmp_path / "RCX_ecc" / "data" + output_dir = tmp_path / "RCX_ecc" / "output" + spef_writer = data_dir / "spef_writer" + spef_writer.mkdir(parents=True) + (spef_writer / "present.spef").write_text("*SPEF\n", encoding="utf-8") + step = EccStep( + name=StepEnum.RCX.value, + data=EccData(dir=data_dir), + output=EccOutput(dir=output_dir, spef=[output_dir / "missing.spef"]), + ) + workspace = Workspace(directory=tmp_path, logger=FakeLogger()) + + assert ecc_runner.copy_rcx_spef_outputs(workspace, step) is False + assert not (output_dir / "missing.spef").exists() + assert step.output.spef == [output_dir / "missing.spef"] + + +def test_copy_rcx_spef_outputs_fails_when_published_spef_is_zero_byte(tmp_path): + data_dir = tmp_path / "RCX_ecc" / "data" + output_dir = tmp_path / "RCX_ecc" / "output" + spef_writer = data_dir / "spef_writer" + spef_writer.mkdir(parents=True) + (spef_writer / "empty.spef").write_text("", encoding="utf-8") + step = EccStep( + name=StepEnum.RCX.value, + data=EccData(dir=data_dir), + output=EccOutput(dir=output_dir, spef=[output_dir / "empty.spef"]), + ) + workspace = Workspace(directory=tmp_path, logger=FakeLogger()) + + assert ecc_runner.copy_rcx_spef_outputs(workspace, step) is False + assert (output_dir / "empty.spef").read_text(encoding="utf-8") == "" + assert step.output.spef == [output_dir / "empty.spef"] + + +def _make_rcx_step(tmp_path): + return EccStep( + name=StepEnum.RCX.value, + data=EccData(dir=tmp_path / "RCX_ecc" / "data"), + output=EccOutput(dir=tmp_path / "RCX_ecc" / "output"), + ) + + +def test_run_rcx_propagates_init_failure(tmp_path, monkeypatch): + workspace = Workspace(directory=tmp_path, logger=FakeLogger()) + step = _make_rcx_step(tmp_path) + module = FakeRcxModule(init_ok=False) + sub_flow = FakeSubFlow() + monkeypatch.setattr(ecc_runner, "EccSubFlow", lambda **kwargs: sub_flow) + + assert ecc_runner.run_rcx(workspace, step, module) is False + + call_names = [call[0] for call in module.calls] + assert "init_rcx" in call_names + assert "run_rcx" not in call_names + assert call_names.count("destroy_rcx") == 1 + assert { + "step_name": EccSubFlowEnum.run_rcx.value, + "state": StateEnum.Imcomplete, + } in sub_flow.updates + + +def test_run_rcx_propagates_native_run_failure(tmp_path, monkeypatch): + workspace = Workspace(directory=tmp_path, logger=FakeLogger()) + step = _make_rcx_step(tmp_path) + module = FakeRcxModule(run_ok=False) + sub_flow = FakeSubFlow() + monkeypatch.setattr(ecc_runner, "EccSubFlow", lambda **kwargs: sub_flow) + + assert ecc_runner.run_rcx(workspace, step, module) is False + + call_names = [call[0] for call in module.calls] + assert "init_rcx" in call_names + assert "run_rcx" in call_names + assert call_names.count("destroy_rcx") == 1 + assert { + "step_name": EccSubFlowEnum.run_rcx.value, + "state": StateEnum.Imcomplete, + } in sub_flow.updates + + +def test_run_rcx_publishes_fresh_spef_and_releases_extractor_once(tmp_path, monkeypatch): + workspace = Workspace(directory=tmp_path, logger=FakeLogger()) + step = _make_rcx_step(tmp_path) + spef_writer = tmp_path / "RCX_ecc" / "data" / "spef_writer" + spef_writer.mkdir(parents=True) + module = FakeRcxModule() + + def run_native_extraction(): + module.calls.append(("run_rcx",)) + (spef_writer / "gcd_Cworst_125C.spef").write_text("*SPEF\n", encoding="utf-8") + return True + + module.run_rcx = run_native_extraction + sub_flow = FakeSubFlow() + monkeypatch.setattr(ecc_runner, "EccSubFlow", lambda **kwargs: sub_flow) + monkeypatch.setattr(ecc_runner, "save_data", lambda **kwargs: True) + monkeypatch.setattr(ecc_runner, "save_rcx_spef_feature_facts", lambda **kwargs: True) + monkeypatch.setattr(ecc_runner, "run_analysis", lambda **kwargs: None) + + assert ecc_runner.run_rcx(workspace, step, module) is True + + published = tmp_path / "RCX_ecc" / "output" / "gcd_Cworst_125C.spef" + assert published.read_text(encoding="utf-8") == "*SPEF\n" + assert step.output.spef == [published] + call_names = [call[0] for call in module.calls] + assert call_names.count("destroy_rcx") == 1 + assert { + "step_name": EccSubFlowEnum.run_rcx.value, + "state": StateEnum.Success, + } in sub_flow.updates + + +def test_run_rcx_wipes_stale_spef_and_fails_when_extraction_produces_nothing(tmp_path, monkeypatch): + workspace = Workspace(directory=tmp_path, logger=FakeLogger()) + step = _make_rcx_step(tmp_path) + spef_writer = tmp_path / "RCX_ecc" / "data" / "spef_writer" + spef_writer.mkdir(parents=True) + stale_spef = spef_writer / "stale.spef" + stale_spef.write_text("*SPEF\nstale\n", encoding="utf-8") + module = FakeRcxModule() + sub_flow = FakeSubFlow() + monkeypatch.setattr(ecc_runner, "EccSubFlow", lambda **kwargs: sub_flow) + + assert ecc_runner.run_rcx(workspace, step, module) is False + + assert not stale_spef.exists() + assert not (tmp_path / "RCX_ecc" / "output" / "stale.spef").exists() + assert { + "step_name": EccSubFlowEnum.run_rcx.value, + "state": StateEnum.Imcomplete, + } in sub_flow.updates + + +def test_run_rcx_fails_when_spef_writer_dir_is_missing(tmp_path, monkeypatch): + workspace = Workspace(directory=tmp_path, logger=FakeLogger()) + step = _make_rcx_step(tmp_path) + module = FakeRcxModule() + sub_flow = FakeSubFlow() + monkeypatch.setattr(ecc_runner, "EccSubFlow", lambda **kwargs: sub_flow) + + assert ecc_runner.run_rcx(workspace, step, module) is False + + assert { + "step_name": EccSubFlowEnum.run_rcx.value, + "state": StateEnum.Imcomplete, + } in sub_flow.updates + + def test_run_sta_uses_matched_report_and_feature_corner_directories(tmp_path, monkeypatch): config_dir = tmp_path / "config" config_dir.mkdir() From bbf343d68e7cc66c51188236e26f20eafa5f0d92 Mon Sep 17 00:00:00 2001 From: Emin Date: Sun, 6 Sep 2026 08:09:38 +0800 Subject: [PATCH 03/10] fix(ecc): validate all RCX SPEF sources before publishing any Require every selected SPEF source to exist and be non-empty before copying anything to the step output, so a partial set or a stale destination from an earlier run can no longer pass as fresh extraction output. Treat any run_step() return other than True or StateEnum.Success as a step failure instead of recording Success. --- chipcompiler/engine/step_execution.py | 4 +- chipcompiler/tools/ecc/runner.py | 20 ++++++---- test/tools/ecc/test_runner.py | 53 ++++++++++++++++++++++++++- 3 files changed, 66 insertions(+), 11 deletions(-) diff --git a/chipcompiler/engine/step_execution.py b/chipcompiler/engine/step_execution.py index 1d7d27ec1..0495a8ae1 100644 --- a/chipcompiler/engine/step_execution.py +++ b/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: diff --git a/chipcompiler/tools/ecc/runner.py b/chipcompiler/tools/ecc/runner.py index af1b2bdcc..916ce45c6 100644 --- a/chipcompiler/tools/ecc/runner.py +++ b/chipcompiler/tools/ecc/runner.py @@ -114,21 +114,25 @@ def copy_rcx_spef_outputs(workspace: Workspace, step: EccStep) -> bool: output_dir / spef_path.name for spef_path in sorted(spef_writer_dir.glob("*.spef")) ] - copied = False + 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(): - continue + 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 + for output_path in output_paths: + source_path = spef_writer_dir / output_path.name output_path.parent.mkdir(parents=True, exist_ok=True) shutil.copy2(source_path, output_path) - copied = True workspace.logger.info("Copied RCX SPEF %s to %s", source_path, output_path) - if not copied: - workspace.logger.error("RCX extraction produced no SPEF artifacts to publish") - return False - for output_path in output_paths: if not (os.path.isfile(output_path) and os.path.getsize(output_path) > 0): workspace.logger.error("Published RCX SPEF is missing or empty: %s", output_path) diff --git a/test/tools/ecc/test_runner.py b/test/tools/ecc/test_runner.py index ea2df6063..f218f0d65 100644 --- a/test/tools/ecc/test_runner.py +++ b/test/tools/ecc/test_runner.py @@ -700,10 +700,11 @@ def test_copy_rcx_spef_outputs_fails_when_declared_spef_source_is_missing(tmp_pa assert ecc_runner.copy_rcx_spef_outputs(workspace, step) is False assert not (output_dir / "missing.spef").exists() + assert not (output_dir / "present.spef").exists() assert step.output.spef == [output_dir / "missing.spef"] -def test_copy_rcx_spef_outputs_fails_when_published_spef_is_zero_byte(tmp_path): +def test_copy_rcx_spef_outputs_fails_when_source_spef_is_zero_byte(tmp_path): data_dir = tmp_path / "RCX_ecc" / "data" output_dir = tmp_path / "RCX_ecc" / "output" spef_writer = data_dir / "spef_writer" @@ -717,10 +718,58 @@ def test_copy_rcx_spef_outputs_fails_when_published_spef_is_zero_byte(tmp_path): workspace = Workspace(directory=tmp_path, logger=FakeLogger()) assert ecc_runner.copy_rcx_spef_outputs(workspace, step) is False - assert (output_dir / "empty.spef").read_text(encoding="utf-8") == "" + assert not (output_dir / "empty.spef").exists() assert step.output.spef == [output_dir / "empty.spef"] +def test_copy_rcx_spef_outputs_fails_and_keeps_stale_destination_when_source_missing(tmp_path): + data_dir = tmp_path / "RCX_ecc" / "data" + output_dir = tmp_path / "RCX_ecc" / "output" + spef_writer = data_dir / "spef_writer" + spef_writer.mkdir(parents=True) + (spef_writer / "good.spef").write_text("*SPEF\nfresh\n", encoding="utf-8") + stale_destination = output_dir / "missing.spef" + output_dir.mkdir(parents=True) + stale_destination.write_text("*SPEF\nstale\n", encoding="utf-8") + spef_outputs = [output_dir / "good.spef", stale_destination] + step = EccStep( + name=StepEnum.RCX.value, + data=EccData(dir=data_dir), + output=EccOutput(dir=output_dir, spef=spef_outputs), + ) + workspace = Workspace(directory=tmp_path, logger=FakeLogger()) + + assert ecc_runner.copy_rcx_spef_outputs(workspace, step) is False + + assert stale_destination.read_text(encoding="utf-8") == "*SPEF\nstale\n" + assert not (output_dir / "good.spef").exists() + assert step.output.spef is spef_outputs + assert step.output.spef == [output_dir / "good.spef", stale_destination] + + +def test_copy_rcx_spef_outputs_fails_without_partial_copy_when_one_source_is_empty(tmp_path): + data_dir = tmp_path / "RCX_ecc" / "data" + output_dir = tmp_path / "RCX_ecc" / "output" + spef_writer = data_dir / "spef_writer" + spef_writer.mkdir(parents=True) + (spef_writer / "good.spef").write_text("*SPEF\n", encoding="utf-8") + (spef_writer / "empty.spef").write_text("", encoding="utf-8") + spef_outputs = [output_dir / "good.spef", output_dir / "empty.spef"] + step = EccStep( + name=StepEnum.RCX.value, + data=EccData(dir=data_dir), + output=EccOutput(dir=output_dir, spef=spef_outputs), + ) + workspace = Workspace(directory=tmp_path, logger=FakeLogger()) + + assert ecc_runner.copy_rcx_spef_outputs(workspace, step) is False + + assert not (output_dir / "good.spef").exists() + assert not (output_dir / "empty.spef").exists() + assert step.output.spef is spef_outputs + assert step.output.spef == [output_dir / "good.spef", output_dir / "empty.spef"] + + def _make_rcx_step(tmp_path): return EccStep( name=StepEnum.RCX.value, From fdc28bfa7bb3f4adae90da6b456153baa1befa1e Mon Sep 17 00:00:00 2001 From: Emin Date: Sun, 6 Sep 2026 08:12:32 +0800 Subject: [PATCH 04/10] fix(engine): fail flow steps when run_step reports failure execute_tool_step only logged the run_step() return value, so a runner that returned False (or None or any non-success StateEnum) after publishing valid outputs was still recorded as Success - the RCX save-data failure paths hit exactly this hole. - execute_tool_step sets step_error unless run_step returns True or StateEnum.Success (the ecc_sizer runner returns StateEnum), naming the raw returned value; the existing flow terminal path then records the step as Imcomplete - copy_rcx_spef_outputs now validates the whole selected source set (regular, non-zero-byte) before creating or overwriting any destination, so a missing or empty source can no longer be masked by a stale destination from an earlier run, and publication is all-or-nothing - engine tests: parametrized over all seven rejected return values plus True/StateEnum.Success positives (incl. the sizer Timing optimization contract) using real output files, the real verdict, persisted ledger assertions, and the exact on_step_completed error - runner tests: mixed valid/missing and valid/empty source sets fail without partial copies, stale destinations are left untouched, and preflight failure preserves the live SPEF list Refs #229 --- test/test_engine_flow.py | 113 +++++++++++++++++++++++++++++++++++++++ 1 file changed, 113 insertions(+) diff --git a/test/test_engine_flow.py b/test/test_engine_flow.py index d3f9efb61..d8ed989cf 100644 --- a/test/test_engine_flow.py +++ b/test/test_engine_flow.py @@ -25,6 +25,16 @@ from chipcompiler.tools.ecc.signoff_checklist import refresh_step_checklist +class CompletionObserver: + """Record on_step_completed deliveries for assertions.""" + + def __init__(self): + self.completed = [] + + def on_step_completed(self, _step, state, error=None): + self.completed.append((state, error)) + + def test_engine_flow_missing_path_is_not_initialized(): engine_flow = EngineFlow(Workspace()) @@ -609,6 +619,109 @@ def on_step_completed(self, _step, state, error=None): assert interrupted_step["peak memory (mb)"] >= 0 +class TestRunStepReturnContract: + """Regression: a non-success run_step() return must fail the step, never Success.""" + + @staticmethod + def _make_flow(tmp_path, name, tool, output): + workspace = Workspace() + workspace.flow.path = tmp_path / "flow.json" + flow_data = {"steps": [{"name": name, "tool": tool, "state": "Unstart"}]} + workspace.flow.path.write_text(json.dumps(flow_data), encoding="utf-8") + engine_flow = EngineFlow(workspace) + workspace_step = EccStep(name=name, directory=tmp_path, tool=tool, output=output) + engine_flow.workspace_steps = [workspace_step] + engine_flow.engine_db = SimpleNamespace(engine=None) + return engine_flow, workspace, workspace_step + + @staticmethod + def _valid_route_output(tmp_path): + output_dir = tmp_path / "output" + output_dir.mkdir() + for name in ("route.def", "route.v", "route.gds"): + (output_dir / name).write_text("content\n", encoding="utf-8") + return EccOutput( + def_=output_dir / "route.def", + verilog=output_dir / "route.v", + gds=output_dir / "route.gds", + ) + + @pytest.mark.parametrize( + "returned", + [ + False, + None, + StateEnum.Imcomplete, + StateEnum.Invalid, + StateEnum.Pending, + StateEnum.Unstart, + StateEnum.Ongoing, + ], + ) + def test_non_success_return_forces_incomplete(self, monkeypatch, tmp_path, returned): + engine_flow, workspace, workspace_step = self._make_flow( + tmp_path, "route", "ecc", self._valid_route_output(tmp_path) + ) + observer = CompletionObserver() + monkeypatch.setattr(tools, "run_step", lambda **_kwargs: returned) + monkeypatch.setattr(tools, "save_layout_image", lambda **_kwargs: True) + + state = engine_flow.run_step(workspace_step, observer=observer) + + assert state == StateEnum.Imcomplete + persisted_step = json.loads(workspace.flow.path.read_text(encoding="utf-8"))["steps"][0] + assert persisted_step["state"] == StateEnum.Imcomplete.value + assert observer.completed == [ + ( + StateEnum.Imcomplete, + f"route(ecc) reported failure (run_step returned {returned!r}).", + ) + ] + + @pytest.mark.parametrize("returned", [True, StateEnum.Success]) + def test_success_returns_reach_success(self, monkeypatch, tmp_path, returned): + engine_flow, workspace, workspace_step = self._make_flow( + tmp_path, "route", "ecc", self._valid_route_output(tmp_path) + ) + observer = CompletionObserver() + monkeypatch.setattr(tools, "run_step", lambda **_kwargs: returned) + monkeypatch.setattr(tools, "save_layout_image", lambda **_kwargs: True) + monkeypatch.setattr(tools, "build_step_metrics", lambda **_kwargs: StepMetrics(data={})) + + state = engine_flow.run_step(workspace_step, observer=observer) + + assert state == StateEnum.Success + persisted_step = json.loads(workspace.flow.path.read_text(encoding="utf-8"))["steps"][0] + assert persisted_step["state"] == StateEnum.Success.value + assert observer.completed == [(StateEnum.Success, None)] + + @pytest.mark.parametrize("returned", [True, StateEnum.Success]) + def test_sizer_state_enum_return_keeps_success_contract(self, monkeypatch, tmp_path, returned): + output_dir = tmp_path / "output" + output_dir.mkdir() + def_path = output_dir / "sized.def" + verilog_path = output_dir / "sized.v" + def_path.write_text("VERSION 5.8 ;\nDESIGN gcd ;\nEND DESIGN\n", encoding="utf-8") + verilog_path.write_text("module gcd; endmodule\n", encoding="utf-8") + engine_flow, workspace, workspace_step = self._make_flow( + tmp_path, + StepEnum.TIMING_OPT.value, + "sizer", + EccOutput(def_=def_path, verilog=verilog_path), + ) + observer = CompletionObserver() + monkeypatch.setattr(tools, "run_step", lambda **_kwargs: returned) + monkeypatch.setattr(tools, "save_layout_image", lambda **_kwargs: True) + monkeypatch.setattr(tools, "build_step_metrics", lambda **_kwargs: StepMetrics(data={})) + + state = engine_flow.run_step(workspace_step, observer=observer) + + assert state == StateEnum.Success + persisted_step = json.loads(workspace.flow.path.read_text(encoding="utf-8"))["steps"][0] + assert persisted_step["state"] == StateEnum.Success.value + assert observer.completed == [(StateEnum.Success, None)] + + class TestCreateStepFailureBreaksChain: """Regression: create_step(None) must break the flow chain and mark step Incomplete.""" From 1a77cbde97d6e22e6aa06c987388aef26410052b Mon Sep 17 00:00:00 2001 From: Emin Date: Sun, 6 Sep 2026 08:26:50 +0800 Subject: [PATCH 05/10] fix(ecc): clean up partial RCX SPEF publication when a copy fails A copy2 failure mid-publication left previously written destinations on disk, so a failed publish could still leave a partial SPEF set visible as current extraction output. Publication now removes every destination written by the failed pass before reporting failure, following the all-or-nothing policy used by STA artifact publication. Refs #229 --- chipcompiler/tools/ecc/runner.py | 20 +++++++++++++++----- test/tools/ecc/test_runner.py | 32 ++++++++++++++++++++++++++++++++ 2 files changed, 47 insertions(+), 5 deletions(-) diff --git a/chipcompiler/tools/ecc/runner.py b/chipcompiler/tools/ecc/runner.py index 916ce45c6..271678d48 100644 --- a/chipcompiler/tools/ecc/runner.py +++ b/chipcompiler/tools/ecc/runner.py @@ -127,11 +127,21 @@ def copy_rcx_spef_outputs(workspace: Workspace, step: EccStep) -> bool: workspace.logger.error("RCX extraction artifact is missing or empty: %s", source_path) return False - for output_path in output_paths: - source_path = spef_writer_dir / output_path.name - 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) + written: 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) + shutil.copy2(source_path, output_path) + written.append(output_path) + workspace.logger.info("Copied RCX SPEF %s to %s", source_path, output_path) + except Exception as exc: + # Publication is all-or-nothing: a partial SPEF set left behind by a + # failed copy must not remain visible as current extraction output. + for output_path in written: + output_path.unlink(missing_ok=True) + workspace.logger.error("Failed to publish RCX SPEF artifacts: %s", exc) + return False for output_path in output_paths: if not (os.path.isfile(output_path) and os.path.getsize(output_path) > 0): diff --git a/test/tools/ecc/test_runner.py b/test/tools/ecc/test_runner.py index f218f0d65..53ea45e4b 100644 --- a/test/tools/ecc/test_runner.py +++ b/test/tools/ecc/test_runner.py @@ -1,4 +1,5 @@ import json +import shutil from pathlib import Path from unittest.mock import Mock @@ -770,6 +771,37 @@ def test_copy_rcx_spef_outputs_fails_without_partial_copy_when_one_source_is_emp assert step.output.spef == [output_dir / "good.spef", output_dir / "empty.spef"] +def test_copy_rcx_spef_outputs_cleans_partial_publication_when_a_copy_fails(tmp_path, monkeypatch): + data_dir = tmp_path / "RCX_ecc" / "data" + output_dir = tmp_path / "RCX_ecc" / "output" + spef_writer = data_dir / "spef_writer" + spef_writer.mkdir(parents=True) + (spef_writer / "a.spef").write_text("*SPEF\na\n", encoding="utf-8") + (spef_writer / "b.spef").write_text("*SPEF\nb\n", encoding="utf-8") + spef_outputs = [output_dir / "a.spef", output_dir / "b.spef"] + step = EccStep( + name=StepEnum.RCX.value, + data=EccData(dir=data_dir), + output=EccOutput(dir=output_dir, spef=spef_outputs), + ) + workspace = Workspace(directory=tmp_path, logger=FakeLogger()) + real_copy2 = shutil.copy2 + + def fail_second_copy(source, destination, **kwargs): + if Path(destination).name == "b.spef": + raise OSError("disk full") + return real_copy2(source, destination, **kwargs) + + monkeypatch.setattr(ecc_runner.shutil, "copy2", fail_second_copy) + + assert ecc_runner.copy_rcx_spef_outputs(workspace, step) is False + + assert not (output_dir / "a.spef").exists() + assert not (output_dir / "b.spef").exists() + assert step.output.spef is spef_outputs + assert step.output.spef == [output_dir / "a.spef", output_dir / "b.spef"] + + def _make_rcx_step(tmp_path): return EccStep( name=StepEnum.RCX.value, From 9901895d5d38395c5d669f0a6184248950568096 Mon Sep 17 00:00:00 2001 From: Emin Date: Sun, 6 Sep 2026 08:42:35 +0800 Subject: [PATCH 06/10] fix(ecc): clean partial RCX SPEF publication on validation failure too Destination validation ran after the copy loop's failure handler, so a publication whose copies succeeded but whose result was missing or zero-byte returned False without removing the destinations written by the pass. Validation now raises inside the same failure-handling block, and every destination written by this invocation is unlinked whenever publication fails for any reason. Refs #229 --- chipcompiler/tools/ecc/runner.py | 12 +++++------ test/tools/ecc/test_runner.py | 36 ++++++++++++++++++++++++++++++++ 2 files changed, 42 insertions(+), 6 deletions(-) diff --git a/chipcompiler/tools/ecc/runner.py b/chipcompiler/tools/ecc/runner.py index 271678d48..ed8edbc21 100644 --- a/chipcompiler/tools/ecc/runner.py +++ b/chipcompiler/tools/ecc/runner.py @@ -135,19 +135,19 @@ def copy_rcx_spef_outputs(workspace: Workspace, step: EccStep) -> bool: shutil.copy2(source_path, output_path) written.append(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 partial SPEF set left behind by a - # failed copy must not remain visible as current extraction output. + # failed copy or a failed validation must not remain visible as + # current extraction output. for output_path in written: output_path.unlink(missing_ok=True) workspace.logger.error("Failed to publish RCX SPEF artifacts: %s", exc) return False - for output_path in output_paths: - if not (os.path.isfile(output_path) and os.path.getsize(output_path) > 0): - workspace.logger.error("Published RCX SPEF is missing or empty: %s", output_path) - return False - if isinstance(step.output.spef, list): step.output.spef[:] = output_paths return True diff --git a/test/tools/ecc/test_runner.py b/test/tools/ecc/test_runner.py index 53ea45e4b..2adcb717c 100644 --- a/test/tools/ecc/test_runner.py +++ b/test/tools/ecc/test_runner.py @@ -802,6 +802,42 @@ def fail_second_copy(source, destination, **kwargs): assert step.output.spef == [output_dir / "a.spef", output_dir / "b.spef"] +@pytest.mark.parametrize("second_write", ["zero-byte", "skipped"]) +def test_copy_rcx_spef_outputs_cleans_partial_publication_when_validation_fails( + tmp_path, monkeypatch, second_write +): + data_dir = tmp_path / "RCX_ecc" / "data" + output_dir = tmp_path / "RCX_ecc" / "output" + spef_writer = data_dir / "spef_writer" + spef_writer.mkdir(parents=True) + (spef_writer / "a.spef").write_text("*SPEF\na\n", encoding="utf-8") + (spef_writer / "b.spef").write_text("*SPEF\nb\n", encoding="utf-8") + spef_outputs = [output_dir / "a.spef", output_dir / "b.spef"] + step = EccStep( + name=StepEnum.RCX.value, + data=EccData(dir=data_dir), + output=EccOutput(dir=output_dir, spef=spef_outputs), + ) + workspace = Workspace(directory=tmp_path, logger=FakeLogger()) + + def copy_with_invalid_second(source, destination, **_kwargs): + destination = Path(destination) + destination.parent.mkdir(parents=True, exist_ok=True) + if destination.name == "b.spef" and second_write == "zero-byte": + destination.write_text("", encoding="utf-8") + elif destination.name == "a.spef": + destination.write_text("*SPEF\na\n", encoding="utf-8") + + monkeypatch.setattr(ecc_runner.shutil, "copy2", copy_with_invalid_second) + + assert ecc_runner.copy_rcx_spef_outputs(workspace, step) is False + + assert not (output_dir / "a.spef").exists() + assert not (output_dir / "b.spef").exists() + assert step.output.spef is spef_outputs + assert step.output.spef == [output_dir / "a.spef", output_dir / "b.spef"] + + def _make_rcx_step(tmp_path): return EccStep( name=StepEnum.RCX.value, From 72ca0174f7e1b84fef27a991431e589dbbc0eb3e Mon Sep 17 00:00:00 2001 From: Emin Date: Sun, 6 Sep 2026 08:50:56 +0800 Subject: [PATCH 07/10] fix(ecc): clean up destination of a failing RCX SPEF copy copy2 can create or truncate its destination and then raise (for example on ENOSPC), and the destination was only registered for cleanup after copy2 returned, so the failed copy's partial SPEF stayed visible. Destinations are now registered for cleanup before the copy runs, and the copy-failure regression covers partial writes (empty and truncated) followed by OSError(ENOSPC). Refs #229 --- chipcompiler/tools/ecc/runner.py | 4 +++- test/tools/ecc/test_runner.py | 12 +++++++++--- 2 files changed, 12 insertions(+), 4 deletions(-) diff --git a/chipcompiler/tools/ecc/runner.py b/chipcompiler/tools/ecc/runner.py index ed8edbc21..a194d9936 100644 --- a/chipcompiler/tools/ecc/runner.py +++ b/chipcompiler/tools/ecc/runner.py @@ -132,8 +132,10 @@ def copy_rcx_spef_outputs(workspace: Workspace, step: EccStep) -> bool: for output_path in output_paths: source_path = spef_writer_dir / output_path.name output_path.parent.mkdir(parents=True, exist_ok=True) - shutil.copy2(source_path, output_path) + # Register before copying: a copy that truncates its destination + # and then fails must still be cleaned up. written.append(output_path) + shutil.copy2(source_path, output_path) workspace.logger.info("Copied RCX SPEF %s to %s", source_path, output_path) for output_path in output_paths: diff --git a/test/tools/ecc/test_runner.py b/test/tools/ecc/test_runner.py index 2adcb717c..44e227628 100644 --- a/test/tools/ecc/test_runner.py +++ b/test/tools/ecc/test_runner.py @@ -1,3 +1,4 @@ +import errno import json import shutil from pathlib import Path @@ -771,7 +772,10 @@ def test_copy_rcx_spef_outputs_fails_without_partial_copy_when_one_source_is_emp assert step.output.spef == [output_dir / "good.spef", output_dir / "empty.spef"] -def test_copy_rcx_spef_outputs_cleans_partial_publication_when_a_copy_fails(tmp_path, monkeypatch): +@pytest.mark.parametrize("partial_content", ["", "*SPEF\ntrunc"]) +def test_copy_rcx_spef_outputs_cleans_partial_publication_when_a_copy_fails( + tmp_path, monkeypatch, partial_content +): data_dir = tmp_path / "RCX_ecc" / "data" output_dir = tmp_path / "RCX_ecc" / "output" spef_writer = data_dir / "spef_writer" @@ -788,8 +792,10 @@ def test_copy_rcx_spef_outputs_cleans_partial_publication_when_a_copy_fails(tmp_ real_copy2 = shutil.copy2 def fail_second_copy(source, destination, **kwargs): - if Path(destination).name == "b.spef": - raise OSError("disk full") + destination = Path(destination) + if destination.name == "b.spef": + destination.write_text(partial_content, encoding="utf-8") + raise OSError(errno.ENOSPC, "No space left on device") return real_copy2(source, destination, **kwargs) monkeypatch.setattr(ecc_runner.shutil, "copy2", fail_second_copy) From 2660ea9dca34a82390a8cc21c6eed4331693bdd7 Mon Sep 17 00:00:00 2001 From: Emin Date: Sun, 6 Sep 2026 09:36:03 +0800 Subject: [PATCH 08/10] fix(ecc): preserve previous SPEFs across a failed RCX publication Publishing straight onto destinations meant a failed pass could delete or truncate the last known-good SPEF it had already overwritten. Publication now moves any existing destination to a .prev backup before replacing it through a .tmp temp file, and a failed pass restores every backup, removes destinations it created, and drops partial temp copies, leaving the step outputs exactly as before the pass. Refs #229 --- chipcompiler/tools/ecc/runner.py | 34 +++++++++++++++++------- test/tools/ecc/test_runner.py | 45 +++++++++++++++++++++++++++++--- 2 files changed, 66 insertions(+), 13 deletions(-) diff --git a/chipcompiler/tools/ecc/runner.py b/chipcompiler/tools/ecc/runner.py index a194d9936..21f2cb36a 100644 --- a/chipcompiler/tools/ecc/runner.py +++ b/chipcompiler/tools/ecc/runner.py @@ -127,26 +127,40 @@ def copy_rcx_spef_outputs(workspace: Workspace, step: EccStep) -> bool: workspace.logger.error("RCX extraction artifact is missing or empty: %s", source_path) return False - written: list[Path] = [] + 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) - # Register before copying: a copy that truncates its destination - # and then fails must still be cleaned up. - written.append(output_path) - shutil.copy2(source_path, output_path) + 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 partial SPEF set left behind by a - # failed copy or a failed validation must not remain visible as - # current extraction output. - for output_path in written: - output_path.unlink(missing_ok=True) + # 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 diff --git a/test/tools/ecc/test_runner.py b/test/tools/ecc/test_runner.py index 44e227628..b1277d65b 100644 --- a/test/tools/ecc/test_runner.py +++ b/test/tools/ecc/test_runner.py @@ -793,7 +793,7 @@ def test_copy_rcx_spef_outputs_cleans_partial_publication_when_a_copy_fails( def fail_second_copy(source, destination, **kwargs): destination = Path(destination) - if destination.name == "b.spef": + if destination.name == ".b.spef.tmp": destination.write_text(partial_content, encoding="utf-8") raise OSError(errno.ENOSPC, "No space left on device") return real_copy2(source, destination, **kwargs) @@ -829,9 +829,9 @@ def test_copy_rcx_spef_outputs_cleans_partial_publication_when_validation_fails( def copy_with_invalid_second(source, destination, **_kwargs): destination = Path(destination) destination.parent.mkdir(parents=True, exist_ok=True) - if destination.name == "b.spef" and second_write == "zero-byte": + if destination.name == ".b.spef.tmp" and second_write == "zero-byte": destination.write_text("", encoding="utf-8") - elif destination.name == "a.spef": + elif destination.name == ".a.spef.tmp": destination.write_text("*SPEF\na\n", encoding="utf-8") monkeypatch.setattr(ecc_runner.shutil, "copy2", copy_with_invalid_second) @@ -844,6 +844,45 @@ def copy_with_invalid_second(source, destination, **_kwargs): assert step.output.spef == [output_dir / "a.spef", output_dir / "b.spef"] +def test_copy_rcx_spef_outputs_restores_previous_spef_when_a_later_copy_fails( + tmp_path, monkeypatch +): + data_dir = tmp_path / "RCX_ecc" / "data" + output_dir = tmp_path / "RCX_ecc" / "output" + spef_writer = data_dir / "spef_writer" + spef_writer.mkdir(parents=True) + (spef_writer / "a.spef").write_text("*SPEF\nfresh a\n", encoding="utf-8") + (spef_writer / "b.spef").write_text("*SPEF\nfresh b\n", encoding="utf-8") + output_dir.mkdir(parents=True) + previous_a = output_dir / "a.spef" + previous_a.write_text("*SPEF\nprevious a\n", encoding="utf-8") + spef_outputs = [output_dir / "a.spef", output_dir / "b.spef"] + step = EccStep( + name=StepEnum.RCX.value, + data=EccData(dir=data_dir), + output=EccOutput(dir=output_dir, spef=spef_outputs), + ) + workspace = Workspace(directory=tmp_path, logger=FakeLogger()) + real_copy2 = shutil.copy2 + + def fail_second_copy(source, destination, **kwargs): + destination = Path(destination) + if destination.name == ".b.spef.tmp": + destination.write_text("*SPEF\ntrunc", encoding="utf-8") + raise OSError(errno.ENOSPC, "No space left on device") + return real_copy2(source, destination, **kwargs) + + monkeypatch.setattr(ecc_runner.shutil, "copy2", fail_second_copy) + + assert ecc_runner.copy_rcx_spef_outputs(workspace, step) is False + + assert previous_a.read_text(encoding="utf-8") == "*SPEF\nprevious a\n" + assert not (output_dir / "b.spef").exists() + assert list(output_dir.iterdir()) == [previous_a] + assert step.output.spef is spef_outputs + assert step.output.spef == [output_dir / "a.spef", output_dir / "b.spef"] + + def _make_rcx_step(tmp_path): return EccStep( name=StepEnum.RCX.value, From 2aa1fbdcb8216c89343f83dac5439154110ecd97 Mon Sep 17 00:00:00 2001 From: Emin Date: Sun, 6 Sep 2026 09:47:50 +0800 Subject: [PATCH 09/10] refactor(ecc): move RCX artifact handling into rcx_artifacts module Per the repository's decomposition rule, path resolution, stale spef_writer cleanup, and SPEF publication move from the oversized ecc runner into chipcompiler/tools/ecc/rcx_artifacts.py, with their tests in test/tools/ecc/test_rcx_artifacts.py; run_rcx keeps orchestrating extraction. Publication now also removes the .prev backups after the output set passes validation, so a successful rerun does not accumulate a second SPEF set on disk; backup-removal failures are reported without failing the already-published step. A successful-rerun regression asserts fresh content and no temp/backup leftovers. Refs #229 --- chipcompiler/tools/ecc/rcx_artifacts.py | 128 +++++++++++ chipcompiler/tools/ecc/runner.py | 113 +--------- test/tools/ecc/test_rcx_artifacts.py | 270 ++++++++++++++++++++++++ test/tools/ecc/test_runner.py | 221 ------------------- 4 files changed, 405 insertions(+), 327 deletions(-) create mode 100644 chipcompiler/tools/ecc/rcx_artifacts.py create mode 100644 test/tools/ecc/test_rcx_artifacts.py diff --git a/chipcompiler/tools/ecc/rcx_artifacts.py b/chipcompiler/tools/ecc/rcx_artifacts.py new file mode 100644 index 000000000..4069687a0 --- /dev/null +++ b/chipcompiler/tools/ecc/rcx_artifacts.py @@ -0,0 +1,128 @@ +"""Publication policy for RCX extraction artifacts. + +iRCX writes extraction results to ``/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] + if declared_paths: + output_paths = [output_dir / spef_path.name for spef_path in declared_paths] + else: + output_paths = [ + output_dir / spef_path.name for spef_path in 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 diff --git a/chipcompiler/tools/ecc/runner.py b/chipcompiler/tools/ecc/runner.py index 21f2cb36a..74bde795b 100644 --- a/chipcompiler/tools/ecc/runner.py +++ b/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,110 +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 _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_rcx_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] - if declared_paths: - output_paths = [output_dir / spef_path.name for spef_path in declared_paths] - else: - output_paths = [ - output_dir / spef_path.name for spef_path in 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 - - if isinstance(step.output.spef, list): - step.output.spef[:] = output_paths - return True - - 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: @@ -905,9 +806,9 @@ def run_rcx(workspace: Workspace, step: EccStep, ecc_module: ECCToolsModule | No # 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) + data_dir, _ = resolve_rcx_dirs(workspace, step) if data_dir is not None: - _wipe_stale_rcx_spef_artifacts(data_dir) + wipe_stale_spef_artifacts(data_dir) try: if not ecc_module.init_rcx( diff --git a/test/tools/ecc/test_rcx_artifacts.py b/test/tools/ecc/test_rcx_artifacts.py new file mode 100644 index 000000000..10d7dde01 --- /dev/null +++ b/test/tools/ecc/test_rcx_artifacts.py @@ -0,0 +1,270 @@ +import errno +import shutil +from pathlib import Path + +import pytest + +from chipcompiler.data import EccData, EccOutput, EccStep, StepEnum, Workspace +from chipcompiler.tools.ecc import rcx_artifacts + + +class FakeLogger: + def __init__(self): + self.infos = [] + self.warnings = [] + self.errors = [] + + def info(self, message, *args): + self.infos.append((message, args)) + + def warning(self, message, *args): + self.warnings.append((message, args)) + + def error(self, message, *args): + self.errors.append((message, args)) + + +def test_copy_rcx_spef_outputs_publishes_to_step_output_dir(tmp_path): + data_dir = tmp_path / "RCX_ecc" / "data" + output_dir = tmp_path / "RCX_ecc" / "output" + source_path = data_dir / "spef_writer" / "gcd_Cworst_125C.spef" + source_path.parent.mkdir(parents=True) + source_path.write_text("*SPEF\n", encoding="utf-8") + spef_outputs = [data_dir / source_path.name] + step = EccStep( + name=StepEnum.RCX.value, + data=EccData(dir=data_dir), + output=EccOutput(dir=output_dir, spef=spef_outputs), + ) + workspace = Workspace(directory=tmp_path, logger=FakeLogger()) + + assert rcx_artifacts.copy_rcx_spef_outputs(workspace, step) is True + + destination = output_dir / source_path.name + assert destination.read_text(encoding="utf-8") == "*SPEF\n" + assert not (data_dir / source_path.name).exists() + assert step.output.spef is spef_outputs + assert step.output.spef == [destination] + + +def test_copy_rcx_spef_outputs_fails_when_declared_spef_source_is_missing(tmp_path): + data_dir = tmp_path / "RCX_ecc" / "data" + output_dir = tmp_path / "RCX_ecc" / "output" + spef_writer = data_dir / "spef_writer" + spef_writer.mkdir(parents=True) + (spef_writer / "present.spef").write_text("*SPEF\n", encoding="utf-8") + step = EccStep( + name=StepEnum.RCX.value, + data=EccData(dir=data_dir), + output=EccOutput(dir=output_dir, spef=[output_dir / "missing.spef"]), + ) + workspace = Workspace(directory=tmp_path, logger=FakeLogger()) + + assert rcx_artifacts.copy_rcx_spef_outputs(workspace, step) is False + assert not (output_dir / "missing.spef").exists() + assert not (output_dir / "present.spef").exists() + assert step.output.spef == [output_dir / "missing.spef"] + + +def test_copy_rcx_spef_outputs_fails_when_source_spef_is_zero_byte(tmp_path): + data_dir = tmp_path / "RCX_ecc" / "data" + output_dir = tmp_path / "RCX_ecc" / "output" + spef_writer = data_dir / "spef_writer" + spef_writer.mkdir(parents=True) + (spef_writer / "empty.spef").write_text("", encoding="utf-8") + step = EccStep( + name=StepEnum.RCX.value, + data=EccData(dir=data_dir), + output=EccOutput(dir=output_dir, spef=[output_dir / "empty.spef"]), + ) + workspace = Workspace(directory=tmp_path, logger=FakeLogger()) + + assert rcx_artifacts.copy_rcx_spef_outputs(workspace, step) is False + assert not (output_dir / "empty.spef").exists() + assert step.output.spef == [output_dir / "empty.spef"] + + +def test_copy_rcx_spef_outputs_fails_and_keeps_stale_destination_when_source_missing(tmp_path): + data_dir = tmp_path / "RCX_ecc" / "data" + output_dir = tmp_path / "RCX_ecc" / "output" + spef_writer = data_dir / "spef_writer" + spef_writer.mkdir(parents=True) + (spef_writer / "good.spef").write_text("*SPEF\nfresh\n", encoding="utf-8") + stale_destination = output_dir / "missing.spef" + output_dir.mkdir(parents=True) + stale_destination.write_text("*SPEF\nstale\n", encoding="utf-8") + spef_outputs = [output_dir / "good.spef", stale_destination] + step = EccStep( + name=StepEnum.RCX.value, + data=EccData(dir=data_dir), + output=EccOutput(dir=output_dir, spef=spef_outputs), + ) + workspace = Workspace(directory=tmp_path, logger=FakeLogger()) + + assert rcx_artifacts.copy_rcx_spef_outputs(workspace, step) is False + + assert stale_destination.read_text(encoding="utf-8") == "*SPEF\nstale\n" + assert not (output_dir / "good.spef").exists() + assert step.output.spef is spef_outputs + assert step.output.spef == [output_dir / "good.spef", stale_destination] + + +def test_copy_rcx_spef_outputs_fails_without_partial_copy_when_one_source_is_empty(tmp_path): + data_dir = tmp_path / "RCX_ecc" / "data" + output_dir = tmp_path / "RCX_ecc" / "output" + spef_writer = data_dir / "spef_writer" + spef_writer.mkdir(parents=True) + (spef_writer / "good.spef").write_text("*SPEF\n", encoding="utf-8") + (spef_writer / "empty.spef").write_text("", encoding="utf-8") + spef_outputs = [output_dir / "good.spef", output_dir / "empty.spef"] + step = EccStep( + name=StepEnum.RCX.value, + data=EccData(dir=data_dir), + output=EccOutput(dir=output_dir, spef=spef_outputs), + ) + workspace = Workspace(directory=tmp_path, logger=FakeLogger()) + + assert rcx_artifacts.copy_rcx_spef_outputs(workspace, step) is False + + assert not (output_dir / "good.spef").exists() + assert not (output_dir / "empty.spef").exists() + assert step.output.spef is spef_outputs + assert step.output.spef == [output_dir / "good.spef", output_dir / "empty.spef"] + + +@pytest.mark.parametrize("partial_content", ["", "*SPEF\ntrunc"]) +def test_copy_rcx_spef_outputs_cleans_partial_publication_when_a_copy_fails( + tmp_path, monkeypatch, partial_content +): + data_dir = tmp_path / "RCX_ecc" / "data" + output_dir = tmp_path / "RCX_ecc" / "output" + spef_writer = data_dir / "spef_writer" + spef_writer.mkdir(parents=True) + (spef_writer / "a.spef").write_text("*SPEF\na\n", encoding="utf-8") + (spef_writer / "b.spef").write_text("*SPEF\nb\n", encoding="utf-8") + spef_outputs = [output_dir / "a.spef", output_dir / "b.spef"] + step = EccStep( + name=StepEnum.RCX.value, + data=EccData(dir=data_dir), + output=EccOutput(dir=output_dir, spef=spef_outputs), + ) + workspace = Workspace(directory=tmp_path, logger=FakeLogger()) + real_copy2 = shutil.copy2 + + def fail_second_copy(source, destination, **kwargs): + destination = Path(destination) + if destination.name == ".b.spef.tmp": + destination.write_text(partial_content, encoding="utf-8") + raise OSError(errno.ENOSPC, "No space left on device") + return real_copy2(source, destination, **kwargs) + + monkeypatch.setattr(rcx_artifacts.shutil, "copy2", fail_second_copy) + + assert rcx_artifacts.copy_rcx_spef_outputs(workspace, step) is False + + assert not (output_dir / "a.spef").exists() + assert not (output_dir / "b.spef").exists() + assert step.output.spef is spef_outputs + assert step.output.spef == [output_dir / "a.spef", output_dir / "b.spef"] + + +@pytest.mark.parametrize("second_write", ["zero-byte", "skipped"]) +def test_copy_rcx_spef_outputs_cleans_partial_publication_when_validation_fails( + tmp_path, monkeypatch, second_write +): + data_dir = tmp_path / "RCX_ecc" / "data" + output_dir = tmp_path / "RCX_ecc" / "output" + spef_writer = data_dir / "spef_writer" + spef_writer.mkdir(parents=True) + (spef_writer / "a.spef").write_text("*SPEF\na\n", encoding="utf-8") + (spef_writer / "b.spef").write_text("*SPEF\nb\n", encoding="utf-8") + spef_outputs = [output_dir / "a.spef", output_dir / "b.spef"] + step = EccStep( + name=StepEnum.RCX.value, + data=EccData(dir=data_dir), + output=EccOutput(dir=output_dir, spef=spef_outputs), + ) + workspace = Workspace(directory=tmp_path, logger=FakeLogger()) + + def copy_with_invalid_second(source, destination, **_kwargs): + destination = Path(destination) + destination.parent.mkdir(parents=True, exist_ok=True) + if destination.name == ".b.spef.tmp" and second_write == "zero-byte": + destination.write_text("", encoding="utf-8") + elif destination.name == ".a.spef.tmp": + destination.write_text("*SPEF\na\n", encoding="utf-8") + + monkeypatch.setattr(rcx_artifacts.shutil, "copy2", copy_with_invalid_second) + + assert rcx_artifacts.copy_rcx_spef_outputs(workspace, step) is False + + assert not (output_dir / "a.spef").exists() + assert not (output_dir / "b.spef").exists() + assert step.output.spef is spef_outputs + assert step.output.spef == [output_dir / "a.spef", output_dir / "b.spef"] + + +def test_copy_rcx_spef_outputs_restores_previous_spef_when_a_later_copy_fails( + tmp_path, monkeypatch +): + data_dir = tmp_path / "RCX_ecc" / "data" + output_dir = tmp_path / "RCX_ecc" / "output" + spef_writer = data_dir / "spef_writer" + spef_writer.mkdir(parents=True) + (spef_writer / "a.spef").write_text("*SPEF\nfresh a\n", encoding="utf-8") + (spef_writer / "b.spef").write_text("*SPEF\nfresh b\n", encoding="utf-8") + output_dir.mkdir(parents=True) + previous_a = output_dir / "a.spef" + previous_a.write_text("*SPEF\nprevious a\n", encoding="utf-8") + spef_outputs = [output_dir / "a.spef", output_dir / "b.spef"] + step = EccStep( + name=StepEnum.RCX.value, + data=EccData(dir=data_dir), + output=EccOutput(dir=output_dir, spef=spef_outputs), + ) + workspace = Workspace(directory=tmp_path, logger=FakeLogger()) + real_copy2 = shutil.copy2 + + def fail_second_copy(source, destination, **kwargs): + destination = Path(destination) + if destination.name == ".b.spef.tmp": + destination.write_text("*SPEF\ntrunc", encoding="utf-8") + raise OSError(errno.ENOSPC, "No space left on device") + return real_copy2(source, destination, **kwargs) + + monkeypatch.setattr(rcx_artifacts.shutil, "copy2", fail_second_copy) + + assert rcx_artifacts.copy_rcx_spef_outputs(workspace, step) is False + + assert previous_a.read_text(encoding="utf-8") == "*SPEF\nprevious a\n" + assert not (output_dir / "b.spef").exists() + assert list(output_dir.iterdir()) == [previous_a] + assert step.output.spef is spef_outputs + assert step.output.spef == [output_dir / "a.spef", output_dir / "b.spef"] + + +def test_copy_rcx_spef_outputs_removes_backups_after_successful_republish(tmp_path): + data_dir = tmp_path / "RCX_ecc" / "data" + output_dir = tmp_path / "RCX_ecc" / "output" + spef_writer = data_dir / "spef_writer" + spef_writer.mkdir(parents=True) + (spef_writer / "a.spef").write_text("*SPEF\nfresh a\n", encoding="utf-8") + (spef_writer / "b.spef").write_text("*SPEF\nfresh b\n", encoding="utf-8") + output_dir.mkdir(parents=True) + (output_dir / "a.spef").write_text("*SPEF\nprevious a\n", encoding="utf-8") + (output_dir / "b.spef").write_text("*SPEF\nprevious b\n", encoding="utf-8") + spef_outputs = [output_dir / "a.spef", output_dir / "b.spef"] + step = EccStep( + name=StepEnum.RCX.value, + data=EccData(dir=data_dir), + output=EccOutput(dir=output_dir, spef=spef_outputs), + ) + workspace = Workspace(directory=tmp_path, logger=FakeLogger()) + + assert rcx_artifacts.copy_rcx_spef_outputs(workspace, step) is True + + assert (output_dir / "a.spef").read_text(encoding="utf-8") == "*SPEF\nfresh a\n" + assert (output_dir / "b.spef").read_text(encoding="utf-8") == "*SPEF\nfresh b\n" + assert sorted(path.name for path in output_dir.iterdir()) == ["a.spef", "b.spef"] + assert step.output.spef is spef_outputs + assert step.output.spef == [output_dir / "a.spef", output_dir / "b.spef"] diff --git a/test/tools/ecc/test_runner.py b/test/tools/ecc/test_runner.py index b1277d65b..4def6b45b 100644 --- a/test/tools/ecc/test_runner.py +++ b/test/tools/ecc/test_runner.py @@ -1,6 +1,4 @@ -import errno import json -import shutil from pathlib import Path from unittest.mock import Mock @@ -664,225 +662,6 @@ def test_sta_entry_workspace_uses_declared_spef(tmp_path): assert [item["spef_file"] for item in items] == [str(declared_spef)] -def test_copy_rcx_spef_outputs_publishes_to_step_output_dir(tmp_path): - data_dir = tmp_path / "RCX_ecc" / "data" - output_dir = tmp_path / "RCX_ecc" / "output" - source_path = data_dir / "spef_writer" / "gcd_Cworst_125C.spef" - source_path.parent.mkdir(parents=True) - source_path.write_text("*SPEF\n", encoding="utf-8") - spef_outputs = [data_dir / source_path.name] - step = EccStep( - name=StepEnum.RCX.value, - data=EccData(dir=data_dir), - output=EccOutput(dir=output_dir, spef=spef_outputs), - ) - workspace = Workspace(directory=tmp_path, logger=FakeLogger()) - - assert ecc_runner.copy_rcx_spef_outputs(workspace, step) is True - - destination = output_dir / source_path.name - assert destination.read_text(encoding="utf-8") == "*SPEF\n" - assert not (data_dir / source_path.name).exists() - assert step.output.spef is spef_outputs - assert step.output.spef == [destination] - - -def test_copy_rcx_spef_outputs_fails_when_declared_spef_source_is_missing(tmp_path): - data_dir = tmp_path / "RCX_ecc" / "data" - output_dir = tmp_path / "RCX_ecc" / "output" - spef_writer = data_dir / "spef_writer" - spef_writer.mkdir(parents=True) - (spef_writer / "present.spef").write_text("*SPEF\n", encoding="utf-8") - step = EccStep( - name=StepEnum.RCX.value, - data=EccData(dir=data_dir), - output=EccOutput(dir=output_dir, spef=[output_dir / "missing.spef"]), - ) - workspace = Workspace(directory=tmp_path, logger=FakeLogger()) - - assert ecc_runner.copy_rcx_spef_outputs(workspace, step) is False - assert not (output_dir / "missing.spef").exists() - assert not (output_dir / "present.spef").exists() - assert step.output.spef == [output_dir / "missing.spef"] - - -def test_copy_rcx_spef_outputs_fails_when_source_spef_is_zero_byte(tmp_path): - data_dir = tmp_path / "RCX_ecc" / "data" - output_dir = tmp_path / "RCX_ecc" / "output" - spef_writer = data_dir / "spef_writer" - spef_writer.mkdir(parents=True) - (spef_writer / "empty.spef").write_text("", encoding="utf-8") - step = EccStep( - name=StepEnum.RCX.value, - data=EccData(dir=data_dir), - output=EccOutput(dir=output_dir, spef=[output_dir / "empty.spef"]), - ) - workspace = Workspace(directory=tmp_path, logger=FakeLogger()) - - assert ecc_runner.copy_rcx_spef_outputs(workspace, step) is False - assert not (output_dir / "empty.spef").exists() - assert step.output.spef == [output_dir / "empty.spef"] - - -def test_copy_rcx_spef_outputs_fails_and_keeps_stale_destination_when_source_missing(tmp_path): - data_dir = tmp_path / "RCX_ecc" / "data" - output_dir = tmp_path / "RCX_ecc" / "output" - spef_writer = data_dir / "spef_writer" - spef_writer.mkdir(parents=True) - (spef_writer / "good.spef").write_text("*SPEF\nfresh\n", encoding="utf-8") - stale_destination = output_dir / "missing.spef" - output_dir.mkdir(parents=True) - stale_destination.write_text("*SPEF\nstale\n", encoding="utf-8") - spef_outputs = [output_dir / "good.spef", stale_destination] - step = EccStep( - name=StepEnum.RCX.value, - data=EccData(dir=data_dir), - output=EccOutput(dir=output_dir, spef=spef_outputs), - ) - workspace = Workspace(directory=tmp_path, logger=FakeLogger()) - - assert ecc_runner.copy_rcx_spef_outputs(workspace, step) is False - - assert stale_destination.read_text(encoding="utf-8") == "*SPEF\nstale\n" - assert not (output_dir / "good.spef").exists() - assert step.output.spef is spef_outputs - assert step.output.spef == [output_dir / "good.spef", stale_destination] - - -def test_copy_rcx_spef_outputs_fails_without_partial_copy_when_one_source_is_empty(tmp_path): - data_dir = tmp_path / "RCX_ecc" / "data" - output_dir = tmp_path / "RCX_ecc" / "output" - spef_writer = data_dir / "spef_writer" - spef_writer.mkdir(parents=True) - (spef_writer / "good.spef").write_text("*SPEF\n", encoding="utf-8") - (spef_writer / "empty.spef").write_text("", encoding="utf-8") - spef_outputs = [output_dir / "good.spef", output_dir / "empty.spef"] - step = EccStep( - name=StepEnum.RCX.value, - data=EccData(dir=data_dir), - output=EccOutput(dir=output_dir, spef=spef_outputs), - ) - workspace = Workspace(directory=tmp_path, logger=FakeLogger()) - - assert ecc_runner.copy_rcx_spef_outputs(workspace, step) is False - - assert not (output_dir / "good.spef").exists() - assert not (output_dir / "empty.spef").exists() - assert step.output.spef is spef_outputs - assert step.output.spef == [output_dir / "good.spef", output_dir / "empty.spef"] - - -@pytest.mark.parametrize("partial_content", ["", "*SPEF\ntrunc"]) -def test_copy_rcx_spef_outputs_cleans_partial_publication_when_a_copy_fails( - tmp_path, monkeypatch, partial_content -): - data_dir = tmp_path / "RCX_ecc" / "data" - output_dir = tmp_path / "RCX_ecc" / "output" - spef_writer = data_dir / "spef_writer" - spef_writer.mkdir(parents=True) - (spef_writer / "a.spef").write_text("*SPEF\na\n", encoding="utf-8") - (spef_writer / "b.spef").write_text("*SPEF\nb\n", encoding="utf-8") - spef_outputs = [output_dir / "a.spef", output_dir / "b.spef"] - step = EccStep( - name=StepEnum.RCX.value, - data=EccData(dir=data_dir), - output=EccOutput(dir=output_dir, spef=spef_outputs), - ) - workspace = Workspace(directory=tmp_path, logger=FakeLogger()) - real_copy2 = shutil.copy2 - - def fail_second_copy(source, destination, **kwargs): - destination = Path(destination) - if destination.name == ".b.spef.tmp": - destination.write_text(partial_content, encoding="utf-8") - raise OSError(errno.ENOSPC, "No space left on device") - return real_copy2(source, destination, **kwargs) - - monkeypatch.setattr(ecc_runner.shutil, "copy2", fail_second_copy) - - assert ecc_runner.copy_rcx_spef_outputs(workspace, step) is False - - assert not (output_dir / "a.spef").exists() - assert not (output_dir / "b.spef").exists() - assert step.output.spef is spef_outputs - assert step.output.spef == [output_dir / "a.spef", output_dir / "b.spef"] - - -@pytest.mark.parametrize("second_write", ["zero-byte", "skipped"]) -def test_copy_rcx_spef_outputs_cleans_partial_publication_when_validation_fails( - tmp_path, monkeypatch, second_write -): - data_dir = tmp_path / "RCX_ecc" / "data" - output_dir = tmp_path / "RCX_ecc" / "output" - spef_writer = data_dir / "spef_writer" - spef_writer.mkdir(parents=True) - (spef_writer / "a.spef").write_text("*SPEF\na\n", encoding="utf-8") - (spef_writer / "b.spef").write_text("*SPEF\nb\n", encoding="utf-8") - spef_outputs = [output_dir / "a.spef", output_dir / "b.spef"] - step = EccStep( - name=StepEnum.RCX.value, - data=EccData(dir=data_dir), - output=EccOutput(dir=output_dir, spef=spef_outputs), - ) - workspace = Workspace(directory=tmp_path, logger=FakeLogger()) - - def copy_with_invalid_second(source, destination, **_kwargs): - destination = Path(destination) - destination.parent.mkdir(parents=True, exist_ok=True) - if destination.name == ".b.spef.tmp" and second_write == "zero-byte": - destination.write_text("", encoding="utf-8") - elif destination.name == ".a.spef.tmp": - destination.write_text("*SPEF\na\n", encoding="utf-8") - - monkeypatch.setattr(ecc_runner.shutil, "copy2", copy_with_invalid_second) - - assert ecc_runner.copy_rcx_spef_outputs(workspace, step) is False - - assert not (output_dir / "a.spef").exists() - assert not (output_dir / "b.spef").exists() - assert step.output.spef is spef_outputs - assert step.output.spef == [output_dir / "a.spef", output_dir / "b.spef"] - - -def test_copy_rcx_spef_outputs_restores_previous_spef_when_a_later_copy_fails( - tmp_path, monkeypatch -): - data_dir = tmp_path / "RCX_ecc" / "data" - output_dir = tmp_path / "RCX_ecc" / "output" - spef_writer = data_dir / "spef_writer" - spef_writer.mkdir(parents=True) - (spef_writer / "a.spef").write_text("*SPEF\nfresh a\n", encoding="utf-8") - (spef_writer / "b.spef").write_text("*SPEF\nfresh b\n", encoding="utf-8") - output_dir.mkdir(parents=True) - previous_a = output_dir / "a.spef" - previous_a.write_text("*SPEF\nprevious a\n", encoding="utf-8") - spef_outputs = [output_dir / "a.spef", output_dir / "b.spef"] - step = EccStep( - name=StepEnum.RCX.value, - data=EccData(dir=data_dir), - output=EccOutput(dir=output_dir, spef=spef_outputs), - ) - workspace = Workspace(directory=tmp_path, logger=FakeLogger()) - real_copy2 = shutil.copy2 - - def fail_second_copy(source, destination, **kwargs): - destination = Path(destination) - if destination.name == ".b.spef.tmp": - destination.write_text("*SPEF\ntrunc", encoding="utf-8") - raise OSError(errno.ENOSPC, "No space left on device") - return real_copy2(source, destination, **kwargs) - - monkeypatch.setattr(ecc_runner.shutil, "copy2", fail_second_copy) - - assert ecc_runner.copy_rcx_spef_outputs(workspace, step) is False - - assert previous_a.read_text(encoding="utf-8") == "*SPEF\nprevious a\n" - assert not (output_dir / "b.spef").exists() - assert list(output_dir.iterdir()) == [previous_a] - assert step.output.spef is spef_outputs - assert step.output.spef == [output_dir / "a.spef", output_dir / "b.spef"] - - def _make_rcx_step(tmp_path): return EccStep( name=StepEnum.RCX.value, From d5f0542f6144b451a84bc153f078b024bbb01b80 Mon Sep 17 00:00:00 2001 From: Emin Date: Sun, 6 Sep 2026 10:01:16 +0800 Subject: [PATCH 10/10] refactor(ecc): simplify RCX verdict and publication selection Finalize-phase cleanup, functionality-equivalent only: the RCX verdict loop collapses to the guard + all() idiom used elsewhere in the codebase, and the publication selection consolidates the duplicated destination comprehension. Refs #229 --- chipcompiler/engine/flow.py | 8 +++----- chipcompiler/tools/ecc/rcx_artifacts.py | 10 ++++------ 2 files changed, 7 insertions(+), 11 deletions(-) diff --git a/chipcompiler/engine/flow.py b/chipcompiler/engine/flow.py index 02b86e5b7..63ed79919 100644 --- a/chipcompiler/engine/flow.py +++ b/chipcompiler/engine/flow.py @@ -305,11 +305,9 @@ def check_step_result(self, workspace_step: WorkspaceStep): success = True case StepEnum.RCX.value: spef_list = ecc_output.spef if ecc_output else [] - success = bool(spef_list) - for spef in spef_list: - if not (os.path.isfile(spef) and os.path.getsize(spef) > 0): - success = False - break + 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 diff --git a/chipcompiler/tools/ecc/rcx_artifacts.py b/chipcompiler/tools/ecc/rcx_artifacts.py index 4069687a0..c93b03e37 100644 --- a/chipcompiler/tools/ecc/rcx_artifacts.py +++ b/chipcompiler/tools/ecc/rcx_artifacts.py @@ -56,12 +56,10 @@ def copy_rcx_spef_outputs(workspace: Workspace, step: EccStep) -> bool: return False declared_paths = [spef_path for spef_path in step.output.spef if spef_path] - if declared_paths: - output_paths = [output_dir / spef_path.name for spef_path in declared_paths] - else: - output_paths = [ - output_dir / spef_path.name for spef_path in sorted(spef_writer_dir.glob("*.spef")) - ] + 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")