diff --git a/.github/workflows/gcd-undo-verify.yml b/.github/workflows/gcd-undo-verify.yml new file mode 100644 index 0000000..c1ec068 --- /dev/null +++ b/.github/workflows/gcd-undo-verify.yml @@ -0,0 +1,98 @@ +name: GCD packaged undo + +on: + workflow_dispatch: + push: + paths: + - '.github/workflows/gcd-undo-verify.yml' + - 'scripts/**' + - 'setup/**' + - 'examples/backend/gcd/**' + - 'toolchain.json' + - 'tools/**' + - 'tests/**' + pull_request: + paths: + - '.github/workflows/gcd-undo-verify.yml' + - 'scripts/**' + - 'setup/**' + - 'examples/backend/gcd/**' + - 'toolchain.json' + - 'tools/**' + - 'tests/**' + +permissions: + contents: read + +jobs: + undo-replay: + runs-on: ubuntu-24.04 + timeout-minutes: 45 + defaults: + run: + shell: bash + env: + RUN: runs/gcd-undo + PYTHONPATH: '' + NAJAEDA_SRC: '' + steps: + - uses: actions/checkout@v4 + - uses: actions/setup-python@v5 + with: + python-version: '3.13' + - name: Offline checks + run: python -m unittest discover -s tests -v + - name: Install Nix (reuse an existing installation) + uses: cachix/install-nix-action@v31 + with: + install_url: https://releases.nixos.org/nix/nix-2.35.1/install + enable_kvm: false + extra_nix_config: | + experimental-features = nix-command flakes + max-jobs = 0 + builders = + - name: Obtain cached OpenROAD (fail before downloading dependencies on a cache miss) + run: | + mkdir -p .cache/gcd-tools + df -h . /nix | tee .cache/gcd-tools/disk-before.txt + openroad=$(python -c 'import json; print(json.load(open("toolchain.json"))["openroad"]["installable"])') + cache=$(python -c 'import json; print(json.load(open("toolchain.json"))["openroad"]["cache"])') + bash tools/install-cached-package.sh openroad "$openroad" "$cache" .cache/gcd-tools + echo "$PWD/.cache/gcd-tools/openroad/bin" >> "$GITHUB_PATH" + - name: Install native Python wheels and pinned pure-Python Kepler MCP package + run: | + mkdir -p .cache/agent-config-codex .cache/agent-config-claude + python setup/mcp.py configure --client codex --project .cache/agent-config-codex \ + --venv .cache/gcd-python --apply | tee .cache/gcd-tools/setup-codex.log + python setup/mcp.py configure --client claude-code --project .cache/agent-config-claude \ + --venv .cache/gcd-python --apply | tee .cache/gcd-tools/setup-claude.log + .cache/gcd-python/bin/python -m pip freeze > .cache/gcd-tools/python-packages.txt + echo "$PWD/.cache/gcd-python/bin" >> "$GITHUB_PATH" + - name: Agent MCP discovery and existing live-session regressions + run: | + python setup/mcp.py check --venv .cache/gcd-python + python scripts/agent_mcp_regression.py --work-dir runs/agent-mcp + python scripts/live_session_regression.py --work-dir runs/live-session + python scripts/live_inspection_regression.py --work-dir runs/live-inspection + - name: Version history retention, failures, undo and continued editing + run: python scripts/versioned_session_regression.py --jupyter --work-dir runs/versioned-session + - name: GCD baseline, edit and undo with Scope, SEC and three fresh OpenROAD runs + run: | + openroad -version | tee .cache/gcd-tools/openroad-version.txt + python scripts/gcd_undo_regression.py --work-dir "$RUN" + - name: Preserve baseline, edited and restored observations and reports + if: always() + uses: actions/upload-artifact@v4 + with: + name: gcd-packaged-undo + path: | + runs/gcd-undo/ + runs/versioned-session/ + runs/live-inspection/ + runs/agent-mcp/ + runs/live-session/ + .cache/gcd-tools/*.json + .cache/gcd-tools/*.txt + .cache/gcd-tools/*.log + if-no-files-found: warn + retention-days: 14 diff --git a/SKILL.md b/SKILL.md index 3ccf141..336aa35 100644 --- a/SKILL.md +++ b/SKILL.md @@ -44,6 +44,9 @@ or comparison afterward, not hints for independent discovery. copies never replace either live design. If a design is later exported for another tool, verify that exported representation separately. Preserve the structured outcome, logs and actual output coverage. + For exploration with rollback and external tools, select + [versioned sessions](tools/session-history.md): numbered netlist checkpoints, + ten recent edits by default, explicit undo, and a protected measured best. 6. For backend tasks, rerun [OpenROAD](tools/openroad/SKILL.md) with the same physical setup. Compare timing, area, estimated power, hold and routing checks. diff --git a/flow/backend/SKILL.md b/flow/backend/SKILL.md index e891ec3..ae35586 100644 --- a/flow/backend/SKILL.md +++ b/flow/backend/SKILL.md @@ -5,6 +5,12 @@ description: Improve a synthesized design using OpenROAD physical reports, Naja- # Backend Improvement +For iterative optimization with undo, use +[session history](../../tools/session-history.md). Keep measurements tied to the +exact numbered netlist and unchanged physical setup. Scope defaults to the active +checkpoint; after undo it must inspect the restored revision. Keep the measured +best independently of rolling history, according to the stated objective. + Read the [parent contract](../../SKILL.md). Start from mapped Verilog, Liberty, LEF/technology data and an SDC; RTL synthesis is not implicit in this flow. diff --git a/scripts/gcd_undo_regression.py b/scripts/gcd_undo_regression.py new file mode 100644 index 0000000..3512f4a --- /dev/null +++ b/scripts/gcd_undo_regression.py @@ -0,0 +1,175 @@ +"""Packaged GCD baseline -> edit -> undo, with real Scope, SEC and OpenROAD.""" + +import argparse +import ast +import asyncio +import json +from pathlib import Path +import sys + +ROOT = Path(__file__).resolve().parents[1] +sys.path.insert(0, str(ROOT)) + +from scripts import gcd_reference_regression as reference +from tools.live_session import _McpClient, SEC +from tools.scope_checkpoints import ScopeCheckpoints +from tools.versioned_session import VersionedDesignSession + + +class ScopeClient(_McpClient): + async def _serve(self): + from mcp import ClientSession, StdioServerParameters + from mcp.client.stdio import stdio_client + + params = StdioServerParameters(command=sys.executable, args=["-m", "naja_scope.server"], + env=reference.clean_env(), cwd=str(ROOT)) + with (self.directory / "scope-server.log").open("w") as log: + async with stdio_client(params, errlog=log) as streams: + async with ClientSession(*streams) as session: + await asyncio.wait_for(session.initialize(), 30) + schemas = await session.list_tools() + reference.save(self.directory / "scope-tools.json", schemas.model_dump(mode="json")) + self.ready.set_result({tool.name for tool in schemas.tools}) + index = 0 + while True: + request = await asyncio.to_thread(self.requests.get) + if request is None: + return + name, arguments, future = request + try: + response = await session.call_tool(name, arguments) + index += 1 + reference.save(self.directory / f"scope-{index:04d}.json", { + "tool": name, "arguments": arguments, + "response": response.model_dump(mode="json")}) + future.set_result(reference.scope_payload(response)) + except Exception as error: + future.set_exception(error) + + +def observations(scope, revision=None): + result = {} + for name, pattern in (("original_gates", "_21[5-9]_"), ("replacement_gates", "ppa_cla_*")): + response = scope.query("find", {"pattern": pattern, "kind": "instance", "limit": 200}, + revision=revision)["result"] + if response.get("has_more") or response.get("truncated"): + raise ValueError("Scope query did not cover the complete replacement boundary") + result[name] = sorted(response["matches"], key=lambda item: item["path"]) + # An unchanged downstream net exposes the changed driver, not just cell counts. + result["boundary"] = scope.query("get_drivers", {"path": "gcd._057_", "limit": 200}, + revision=revision)["result"] + if result["boundary"].get("truncated"): + raise ValueError("Truncated Scope connectivity evidence") + return result + + +def require_restored(baseline, restored): + # Identical bytes, package and setup: demand the same results, not improvement. + for key in ("setup_ns", "hold_ns", "tns_ns", "area_um2", "routing_drc", "power_report"): + if baseline[key] != restored[key]: + raise ValueError(f"Restored OpenROAD result differs from baseline: {key}") + + +def live_cells(session): + # Check actual live objects too: selecting an old file alone is not undo. + with session._inspection_access(): + return sorted((cell.getName(), cell.getModel().getName()) + for cell in session._candidate.getInstances()) + + +def run(work, fixture, timeout): + metadata = reference.prepare(work, fixture) + liberty = fixture / "test/sky130hd/sky130hd_tt.lib" + script_tree = ast.parse((reference.REFERENCE / "edit.py").read_text()) + script = ast.unparse(ast.Module(body=[n for n in script_tree.body if isinstance(n, ast.FunctionDef)], + type_ignores=[])) + objective = {"weights": {"setup_ns": -1}, + "bounds": {"hold_ns": {"min": 0}, "routing_drc": {"max": 0}}} + session = VersionedDesignSession(work / "inputs/input.v", [liberty], sessions_root=work, + objective=objective, timeout=timeout) + client = None + try: + client = ScopeClient(work, timeout) + scope = ScopeCheckpoints(session, client.call) + golden_hash = session.status()["golden_sha256"] + baseline_cells = live_cells(session) + original_attachment = session.mcp_attachment() + context = {"openroad": metadata["openroad_version"], + "sdc": reference.digest(work / "inputs/constraints.sdc"), + "flow": reference.digest(reference.PLATFORM / "run.tcl"), + "fixture": metadata["fixture_manifest_sha256"]} + + def physical(label, measure=True): + print(f"OpenROAD: {label}", flush=True) + directory = work / label + with session.use_checkpoint() as checkpoint: + SEC.require_full(checkpoint["export_proof"], reference.EXPECTED_OUTPUTS) + env = reference.clean_env() + env.update(GCD_RUN_DIR=str(directory), GCD_TEST_DIR=str(fixture / "test"), + GCD_INPUT=checkpoint["verilog_file"], GCD_SDC=str(work / "inputs/constraints.sdc")) + reference.run_command(directory, [metadata["tools"]["openroad"], "-no_init", "-exit", "-metrics", + str(directory / "metrics.json"), str(reference.PLATFORM / "run.tcl")], + timeout=timeout, env=env) + summary = reference.physical_summary(directory) + reference.save(directory / "summary.json", dict(summary, revision=checkpoint["revision"], + input_sha256=checkpoint["files"]["design.v"])) + if measure: + session.record_measurement({k: v for k, v in summary.items() if k != "power_report"}, + context=context, evidence=[directory / "summary.json", + directory / "metrics.json", directory / "reports/power.rpt"]) + return summary + + print("Naja-Scope: baseline revision zero", flush=True) + before = observations(scope) + assert len(before["original_gates"]) == 5 and not before["replacement_gates"] + baseline = physical("baseline") + print("NajaEDA: reference edit, live SEC, checkpoint SEC", flush=True) + SEC.require_full(session.apply_edit(script), reference.EXPECTED_OUTPUTS) + assert live_cells(session) != baseline_cells + edited = session.checkpoint() + after = observations(scope) + assert not after["original_gates"] and len(after["replacement_gates"]) == 31 + assert before["boundary"] != after["boundary"] + assert observations(scope, revision=0) == before # Explicit historical selection. + assert observations(scope) == after # Default returns to current. + candidate = physical("candidate") + gain = reference.compare(baseline, candidate) + assert session.history.best["revision"] == edited["revision"] + best_hash = reference.digest(session.directory / "best/design.v") + print("Undo: restore previous netlist, SEC, then delete discarded revision", flush=True) + undo = session.undo() + SEC.require_full(undo["proof"], reference.EXPECTED_OUTPUTS) + assert undo["restored_revision"] == 0 + assert not Path(edited["directory"]).exists() + assert session.status()["golden_sha256"] == golden_hash + assert live_cells(session) == baseline_cells + assert session.mcp_attachment()["session_id"] != original_attachment["session_id"] + assert reference.digest(session.directory / "best/design.v") == best_hash + restored_observations = observations(scope) + assert restored_observations == before + restored = physical("restored", measure=False) + require_restored(baseline, restored) + assert session.checkpoint()["files"]["design.v"] == reference.digest(work / "inputs/input.v") + reference.validate_inputs(work, metadata) + reference.save(work / "scope-comparison.json", {"baseline": before, "candidate": after, + "restored": restored_observations}) + reference.save(work / "comparison.json", dict(gain, restored=restored, undo=undo, + status="passed", scope_restored=True, physical_restored=True, + live_candidate_restored=True, best_preserved=True, discarded_netlist_deleted=True)) + print(f"PASS: Scope and OpenROAD restored; edit gained {gain['setup_gain_ns']:.9f} ns; " + "18/18 outputs proved after edit and undo", flush=True) + finally: + if client is not None: + client.close() + session.close() + + +if __name__ == "__main__": + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--work-dir", type=Path, required=True) + parser.add_argument("--fixture-dir", type=Path, default=ROOT / ".cache/gcd-fixture-v2") + parser.add_argument("--timeout-seconds", type=int, default=600) + args = parser.parse_args() + if args.timeout_seconds <= 0: + parser.error("--timeout-seconds must be positive") + run(args.work_dir.resolve(), args.fixture_dir.resolve(), args.timeout_seconds) diff --git a/scripts/versioned_session_regression.py b/scripts/versioned_session_regression.py new file mode 100644 index 0000000..757d9da --- /dev/null +++ b/scripts/versioned_session_regression.py @@ -0,0 +1,158 @@ +"""Real packaged-tool history guards, independent of the physical GCD run.""" + +import argparse +import json +from pathlib import Path +import sys +import tempfile +import time +from unittest.mock import patch + +ROOT = Path(__file__).resolve().parents[1] +sys.path.insert(0, str(ROOT)) + +from scripts.live_session_regression import LIBERTY, FIRST, SECOND, DIFFERENT +from scripts.gcd_undo_regression import ScopeClient +from tools.scope_checkpoints import ScopeCheckpoints +from tools.versioned_session import VersionedDesignSession +from tools.live_session import SEC + + +def run(work): + work.mkdir(parents=True, exist_ok=False) + source, library = work / "input.v", work / "cells.lib" + source.write_text("module top(input a, output y); BUF g(.A(a), .Y(y)); endmodule\n") + library.write_text(LIBERTY) + with VersionedDesignSession(source, [library], sessions_root=work, retention=2) as session: + client = ScopeClient(work, 120) + scope = ScopeCheckpoints(session, client.call) + try: + def gates(): + response = scope.query("find", {"pattern": "*", "kind": "instance", "limit": 200}) + return sorted(item["path"] for item in response["result"]["matches"]) + + initial = gates() + original = session.mcp_attachment() + for script in (FIRST, SECOND): + assert session.apply_edit(script)["proved_outputs"] == 1 + assert gates() == ["top.g", "top.h1", "top.h2"] + checkpoint_before_error = session.checkpoint() + with patch.object(SEC, "run_sec", side_effect=RuntimeError("simulated export check failure")): + try: + session.apply_edit('def edit(top):\n top.create_net("unsaved")\n') + except RuntimeError as error: + assert "simulated export" in str(error) + else: + raise AssertionError("Failed checkpoint was published") + assert session.status()["netlist_revision"] is None + assert session.history.active == checkpoint_before_error["revision"] + assert session.undo()["restored_revision"] == 2 + try: + session.apply_edit(DIFFERENT) + except ValueError as error: + assert "counterexample" in str(error) + else: + raise AssertionError("Counterexample accepted") + try: + session.checkpoint() + except RuntimeError: + pass + else: + raise AssertionError("Stale checkpoint claimed as current") + # A failed edit restores the last saved version, without discarding it. + assert session.undo()["restored_revision"] == 2 + assert gates() == ["top.g", "top.h1", "top.h2"] + before_failed_undo = session.status() + with patch.object(SEC, "run_sec", side_effect=RuntimeError("simulated restore proof failure")): + try: + session.undo() + except RuntimeError: + pass + else: + raise AssertionError("Undo ignored its failed file check") + assert session.status() == before_failed_undo + assert Path(session.checkpoint()["verilog_file"]).exists() + assert session.undo()["restored_revision"] == 1 + assert gates() == ["top.g", "top.h"] + assert session.undo()["restored_revision"] == 0 + assert gates() == initial + assert session.mcp_attachment()["session_id"] != original["session_id"] + stale = session._client.call("verify_session", { + "session_id": session.mcp_attachment()["session_id"], + "design1": original["design1"], "design2": original["design2"], "verification": "sec"}) + assert stale["status"] == "error" # Native IDs may match; binding identity cannot. + assert session.apply_edit(FIRST)["revision"] == 5 # Never reuse discarded IDs. + assert gates() == ["top.g", "top.h"] + for number in range(3): + session.apply_edit(f'def edit(top):\n top.create_net("unused{number}")\n') + assert sorted(session.history.records) == [0, 7, 8] + session.configure_history(retention=1) + assert sorted(session.history.records) == [0, 8] + artifact = session.checkpoint() + Path(artifact["verilog_file"]).write_text("corrupt export") + before = session.status() + try: + session.checkpoint() + except ValueError: + pass + else: + raise AssertionError("Modified checkpoint accepted") + assert session.status() == before # Disk tampering never mutates live golden/candidate. + (work / "result.json").write_text(json.dumps({"status": "passed", "retention": True, + "undo": True, "scope_refresh": True, "counterexample_rejected": True, + "failed_edit_recovered": True, "stale_native_reference_rejected": True, + "failed_checkpoint_unpublished": True, "failed_undo_nondestructive": True, + "post_undo_edit": True, "modified_checkpoint_rejected": True}, indent=2) + "\n") + print("PASS: real SEC, Scope, undo, retention, continued edits and stale-ID rejection", flush=True) + finally: + client.close() + + +def run_notebook(work): + from jupyter_client import KernelManager + from scripts.gcd_reference_regression import clean_env + + work.mkdir(parents=True, exist_ok=False) + with tempfile.TemporaryDirectory(prefix="22b-history-kernel-", dir="/tmp") as private: + manager = KernelManager(transport="ipc", connection_file=str(Path(private) / "connection.json")) + manager.kernel_spec.argv = [sys.executable, "-m", "ipykernel_launcher", "-f", "{connection_file}"] + manager.start_kernel(cwd=str(ROOT), env=clean_env()) + client = manager.blocking_client() + client.start_channels() + try: + client.wait_for_ready(timeout=30) + code = ("from pathlib import Path\nfrom scripts.versioned_session_regression import run\n" + f"run(Path({str(work / 'native')!r}))\n") + (work / "cell.py").write_text(code) + request = client.execute(code, store_history=False, allow_stdin=False) + deadline = time.monotonic() + 600 + error = None + with (work / "kernel.log").open("w") as log: + while True: + message = client.get_iopub_msg(timeout=max(0.1, deadline - time.monotonic())) + if message.get("parent_header", {}).get("msg_id") != request: + continue + kind, content = message["msg_type"], message["content"] + if kind == "stream": + log.write(content["text"]) + elif kind == "error": + error = content["ename"] + ": " + content["evalue"] + log.write(error + "\n") + elif kind == "status" and content["execution_state"] == "idle": + break + if error: + raise RuntimeError(error) + if json.loads((work / "native/result.json").read_text())["status"] != "passed": + raise RuntimeError("Missing successful notebook history result") + print("PASS: versioned history, Scope and SEC inside a real Jupyter kernel", flush=True) + finally: + client.stop_channels() + manager.shutdown_kernel(now=True) + + +if __name__ == "__main__": + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--work-dir", required=True, type=Path) + parser.add_argument("--jupyter", action="store_true") + args = parser.parse_args() + (run_notebook if args.jupyter else run)(args.work_dir.resolve()) diff --git a/tests/test_session_history.py b/tests/test_session_history.py new file mode 100644 index 0000000..0c95adc --- /dev/null +++ b/tests/test_session_history.py @@ -0,0 +1,180 @@ +"""Offline storage/selection guards, not physical or formal verification.""" + +from contextlib import contextmanager +import json +from pathlib import Path +import tempfile +import unittest + +from tools.session_history import RevisionHistory +from tools.scope_checkpoints import ScopeCheckpoints +from scripts.gcd_undo_regression import require_restored + + +class HistoryTests(unittest.TestCase): + def setUp(self): + self.temp = tempfile.TemporaryDirectory() + self.addCleanup(self.temp.cleanup) + self.root = Path(self.temp.name) + self.history = RevisionHistory(self.root) + + def publish(self, revision, status="proved"): + directory = Path(tempfile.mkdtemp(dir=self.root)) + (directory / "design.v").write_text(f"// revision {revision}\nmodule top(); endmodule\n") + return self.history.publish(directory, revision, {"top": "top", "export_proof": {"status": status}}) + + def test_default_ten_recent_edits_plus_preserved_baseline(self): + for i in range(13): + self.publish(i) + self.assertEqual(sorted(self.history.records), [0, *range(3, 13)]) + self.assertEqual(self.history.active, 12) + self.assertEqual(json.loads((self.root / "config.json").read_text())["retention"], 10) + + def test_configurable_retention_and_invalid_values(self): + for i in range(4): + self.publish(i) + for bad in (0, -1, True, 1.5, "10"): + with self.assertRaises(ValueError): + self.history.configure(bad) + self.history.configure(1) + self.assertEqual(sorted(self.history.records), [0, 3]) + + def test_undo_deletes_latest_and_does_not_reuse_numbers(self): + self.publish(0) + latest = self.publish(1) + target = self.history.undo_target() + self.history.finish_undo(target["revision"]) + self.assertEqual(self.history.active, 0) + self.assertFalse(Path(latest["directory"]).exists()) + with self.assertRaises(ValueError): + self.publish(1) + self.publish(2) + self.assertEqual(sorted(self.history.records), [0, 2]) + self.assertEqual(self.history.get()["parent"], 0) + + def test_empty_history_or_baseline_cannot_undo(self): + self.publish(0) + with self.assertRaises(ValueError): + self.history.undo_target() + with self.assertRaises(ValueError): + self.history.get(99) + + def test_inflight_inputs_survive_pruning_and_block_deletion(self): + self.publish(0) + self.publish(1) + self.history.configure(1) + with self.history.acquire(1): + with self.assertRaises(RuntimeError): + self.history.undo_target() + self.publish(2) + self.assertIn(1, self.history.records) + self.assertNotIn(1, self.history.records) + + def test_modified_export_or_manifest_rejected(self): + record = self.publish(0) + path = Path(record["verilog_file"]) + original = path.read_text() + path.write_text("corrupt") + with self.assertRaises(ValueError): + self.history.get() + path.write_text(original) + (Path(record["directory"]) / "manifest.json").write_text("{}") + with self.assertRaises(ValueError): + self.history.get() + + def test_bad_or_empty_snapshot_not_published(self): + directory = Path(tempfile.mkdtemp(dir=self.root)) + with self.assertRaises(ValueError): + self.history.publish(directory, 0, {}) + self.assertFalse(self.history.records) + self.assertIsNone(self.history.active) + + def test_best_survives_undo_and_retention_and_requires_comparable_evidence(self): + self.history.objective = {"weights": {"slack": -1}, "bounds": {"area": {"max": 20}}} + proof = self.root / "report.txt" + proof.write_text("synthetic unit-test evidence, not a real timing result") + context = {"constraints": "fixture"} + self.publish(0) + self.publish(1) + result = self.history.measure(1, {"slack": 1, "area": 10}, context, [proof]) + self.assertTrue(result["promoted"]) + best = (self.root / "best/design.v").read_bytes() + self.history.finish_undo(0) + self.publish(2) + self.assertFalse(self.history.measure(2, {"slack": 0, "area": 10}, context, [proof])["promoted"]) + self.publish(3) + self.assertFalse(self.history.measure(3, {"slack": 2, "area": 30}, context, [proof])["promoted"]) + self.history.configure(1) + self.assertEqual((self.root / "best/design.v").read_bytes(), best) + self.publish(4) + with self.assertRaises(ValueError): + self.history.measure(4, {"slack": 2, "area": 10}, {"constraints": "changed"}, [proof]) + + def test_warning_preserved_not_promoted_as_full_proof(self): + self.history.objective = {"weights": {"slack": -1}} + self.publish(0, "warning") + source = self.root / "report.txt" + source.write_text("fixture") + result = self.history.measure(0, {"slack": 3}, {"setup": "fixture"}, [source]) + self.assertFalse(result["promoted"]) + self.assertEqual(self.history.get()["export_proof"]["status"], "warning") + + def test_missing_nan_metrics_or_evidence_rejected(self): + self.publish(0) + for metrics, context, evidence in (({"slack": float("nan")}, {"setup": 1}, [__file__]), + ({"slack": 1}, {}, [__file__]), + ({"slack": 1}, {"setup": 1}, [])): + with self.assertRaises(ValueError): + self.history.measure(0, metrics, context, evidence) + + +class ScopeSelectionTests(unittest.TestCase): + def test_current_historical_undo_and_reuse(self): + current = [0] + calls, loaded = [], [] + + class Session: + libraries = [] + + @contextmanager + def use_checkpoint(self, revision=None): + key = current[0] if revision is None else revision + yield {"revision": key, "verilog_file": f"{key}.v", "top": "top", + "files": {"design.v": str(key)}, "export_proof": {"status": "proved"}} + + def call(tool, arguments): + calls.append(tool) + if tool == "load_verilog": + loaded[:] = arguments["files"] + if tool == "reset_universe": + loaded.clear() + return {"loaded": bool(loaded), "top": {"name": "top"}, "loaded_files": loaded[:], "answer": loaded[:]} + + scope = ScopeCheckpoints(Session(), call) + self.assertEqual(scope.query("get_stats")["revision"], 0) + scope.query("get_stats") + self.assertEqual(calls.count("load_verilog"), 1) + current[0] = 1 + self.assertEqual(scope.query("get_stats")["revision"], 1) + self.assertEqual(scope.query("get_stats", revision=0)["revision"], 0) + self.assertEqual(scope.query("get_stats")["revision"], 1) + current[0] = 0 + self.assertEqual(scope.query("get_stats")["result"]["answer"], ["0.v"]) + with self.assertRaises(ValueError): + scope.query("query_python") + + def test_restored_physical_metrics_must_match_all_fields(self): + baseline = {k: 1 for k in ("setup_ns", "hold_ns", "tns_ns", "area_um2", "routing_drc", "power_report")} + require_restored(baseline, dict(baseline)) + for key in baseline: + with self.assertRaises(ValueError): + require_restored(baseline, dict(baseline, **{key: 2})) + + def test_new_workflow_is_distinct_and_exercises_undo(self): + root = Path(__file__).resolve().parents[1] + old = (root / ".github/workflows/gcd-reference-verify.yml").read_text() + new = (root / ".github/workflows/gcd-undo-verify.yml").read_text() + self.assertNotIn("gcd_undo_regression.py", old) + self.assertIn("gcd_undo_regression.py", new) + self.assertIn("versioned_session_regression.py", new) + self.assertIn("if: always()", new) diff --git a/tools/live-session.md b/tools/live-session.md index b8b5401..35027bd 100644 --- a/tools/live-session.md +++ b/tools/live-session.md @@ -1,5 +1,9 @@ # Persistent Python/Jupyter Sessions +For automatic numbered Verilog checkpoints, configurable retention, undo and a +protected best result, use [versioned sessions](session-history.md). That opt-in +mode builds on this helper; the no-export mode described below remains available. + Use this mode for cumulative NajaEDA edits with automatic SEC after each edit. One dedicated kernel holds two designs: immutable golden and mutable candidate. The candidate is never replaced by a reload between iterations. Both designs diff --git a/tools/naja-scope/SKILL.md b/tools/naja-scope/SKILL.md index faf42ad..531e15d 100644 --- a/tools/naja-scope/SKILL.md +++ b/tools/naja-scope/SKILL.md @@ -5,6 +5,10 @@ description: Inspect design connectivity through Naja-Scope MCP to establish dri # Inspect Before Rewiring +For [versioned sessions](../session-history.md), resolve the current numbered +checkpoint (or a requested historical revision) before querying. Reload Scope +when that selection changes, including after undo. Reuse an unchanged loaded copy. + Use the [package guide](install.md). Discover the installed typed-tool schemas, then load Liberty and Verilog with `load_liberty` and `load_verilog`. Confirm the top and loaded design with `status` before querying. diff --git a/tools/najaeda/SKILL.md b/tools/najaeda/SKILL.md index 02f5323..a273946 100644 --- a/tools/najaeda/SKILL.md +++ b/tools/najaeda/SKILL.md @@ -5,6 +5,10 @@ description: Inspect and edit structural hardware connectivity with NajaEDA, pre # Structural Editing +Use [versioned sessions](../session-history.md) for automatic netlist checkpoints, +retention and undo. Golden stays live and immutable. Undo restores only candidate; +obtain fresh Kepler attachment references afterward, then continue editing. + Use the [package guide](install.md). For incremental edits in one Python/Jupyter kernel, read [persistent sessions](../live-session.md). Supply only `edit(top)` and pure helpers to `session.apply_edit(script)`; do not import, reset, load or diff --git a/tools/scope_checkpoints.py b/tools/scope_checkpoints.py new file mode 100644 index 0000000..ec1ef40 --- /dev/null +++ b/tools/scope_checkpoints.py @@ -0,0 +1,37 @@ +"""Select numbered checkpoints for a separate, file-based Naja-Scope MCP.""" + + +READ_ONLY = frozenset({"status", "resolve", "find", "get_hierarchy", "get_drivers", + "get_loads", "trace_cone", "get_stats", "get_module_card"}) + + +class ScopeCheckpoints: + """call(tool, arguments) is supplied by the caller's existing MCP client.""" + + def __init__(self, session, call): + self.session = session + self.call = call + self.loaded = None + + def query(self, tool, arguments=None, *, revision=None): + if tool not in READ_ONLY: + raise ValueError("Checkpoint queries must use typed read-only Scope tools") + with self.session.use_checkpoint(revision) as checkpoint: + path = checkpoint["verilog_file"] + identity = (path, checkpoint["files"]["design.v"]) + status = self.call("status", {}) + if (self.loaded != identity or not status.get("loaded") + or path not in status.get("loaded_files", [])): + self.loaded = None + self.call("reset_universe", {}) # The Scope process only. + if self.session.libraries: + self.call("load_liberty", {"files": [str(p) for p in self.session.libraries]}) + self.call("load_verilog", {"files": [path]}) + status = self.call("status", {}) + if (not status.get("loaded") or status.get("top", {}).get("name") != checkpoint["top"] + or path not in status.get("loaded_files", [])): + raise RuntimeError("Scope did not load the selected checkpoint") + self.loaded = identity + result = self.call(tool, arguments or {}) + return {"revision": checkpoint["revision"], "design_sha256": identity[1], + "result": result, "export_proof": checkpoint["export_proof"]} diff --git a/tools/session-history.md b/tools/session-history.md new file mode 100644 index 0000000..4099d7b --- /dev/null +++ b/tools/session-history.md @@ -0,0 +1,134 @@ +# Numbered Netlist History And Undo + +Use `VersionedDesignSession` when exploration needs saved netlists, rollback, +Scope inspection and physical measurements. It extends the existing +[live session](live-session.md); the original no-export mode is unchanged. +All implementation is in 22b, using the existing packaged tools. + +```python +from tools.versioned_session import VersionedDesignSession + +session = VersionedDesignSession( + reference="original.v", liberty_files=["cells.lib"], + sessions_root="runs", retention=10, + objective={ + "weights": {"setup_ns": -1}, # Minimize score: maximize slack. + "bounds": {"hold_ns": {"min": 0}, "routing_drc": {"max": 0}}, + }, +) +``` + +The default directory is `runs/session_/`. An explicit, +unused `work_dir` is also supported. The date identifies the session, not the +netlist revision. Numbered `versions/revision-0000/`, `revision-0001/`, etc. +contain `design.v`, a manifest, edit script where applicable, and export proof. +`inputs/` preserves original Verilog and Liberty files. Keep constraints and +physical setup immutable too; record their hashes in measurement context. + +Initialization proves the baseline and saves revision zero. Every validated +`apply_edit(script)` runs mandatory live SEC, exports the result, and runs +separate file-based SEC before publishing its checkpoint. Counterexamples and +tool errors are hard failures; partial proof remains explicitly unproven. +Incomplete checkpoint writes are never selected as current. Failed attempts +retain diagnostics, not accepted versions. If live editing succeeds but +checkpointing fails, current inspection is blocked until repair or undo. + +## Select And Inspect + +```python +current = session.checkpoint() # Latest retained, active netlist. +baseline = session.checkpoint(0) # Explicit historical selection. +session.configure_history(retention=20) +``` + +Retain ten recent edited checkpoints by default, plus the permanent baseline. +Numbers use the existing edit-attempt counter: failures can leave gaps, and undo +never reuses a number. `session.status()["revision"]` is the attempt counter; +`session.status()["netlist_revision"]` is the active saved netlist. Manifests +and file hashes are checked before reuse. + +Naja-Scope runs in a separate process. With an agent's direct MCP tools, resolve +`session.checkpoint()` (or an explicit revision), then load its Verilog and +`session.libraries` through Scope's reset/load tools if the selected copy changed. +Never reset the editing kernel. Confirm Scope's status and record the revision +with answers; current-design questions must not reuse historical data. + +For an existing synchronous MCP client, `ScopeCheckpoints(session, call)` in +[scope_checkpoints.py](scope_checkpoints.py) automates selection and refresh. +`call(tool, arguments)` returns the parsed payload and must raise on MCP errors. +`query(tool, arguments, revision=None)` defaults to current, supports historical +versions and reuses an unchanged loaded copy. It does not register an MCP with +the agent or replace the agent's own client configuration. Do not let another +writer share that Scope server during queries. + +Use `with session.use_checkpoint() as checkpoint:` around external runs. This +pins their input, prevents mid-run undo/pruning, and checks hashes afterward. +The API is for trusted local callers, not a filesystem security sandbox. + +## Undo + +```python +result = session.undo() +attachment = session.mcp_attachment() # External Kepler clients reattach here. +``` + +Undo validates the previous saved file, restores only the candidate in the same +Python kernel, and reruns SEC against unchanged golden. Only after success does +it delete the discarded latest checkpoint. Undo after an unsaved failed edit +restores the latest good checkpoint without deleting it. If intermediate versions +were pruned, undo selects the preceding retained version; baseline is permanent. + +Naja can reuse native IDs on reload. Undo expires the old Kepler attachment and +creates a new binding identity; external MCP clients must reattach and obtain +fresh references. The Python session remains alive, and golden is not reloaded. +Old proof reports remain historical. The attempt counter never decreases. +If native restoration fails, the session is marked invalid and no checkpoint is +deleted; retain the files and start a new session rather than using that candidate. + +## Measurements And Best + +```python +session.record_measurement( + {"setup_ns": 0.04, "hold_ns": 0.02, "routing_drc": 0}, + context={"constraints_sha256": "recorded-hash", "tool_version": "recorded-version"}, + evidence=["physical-run/summary.json", "physical-run/power.rpt"], +) +``` + +These values illustrate the API, not actual measurements. Supply parsed tool +results and evidence, never model estimates. The helper checks numeric validity, +required metrics, consistent context, proof policy and objective bounds; it does +not establish that arbitrary caller-provided metrics are true. Include libraries, +corners, flow, seed/thread settings and constraints in context. Each external run +needs a fresh output directory and an identified checkpoint. + +The objective minimizes the weighted sum subject to bounds. Choose units, +normalization and weights explicitly for combined goals. No objective means no +automatic best selection. Equal/worse or infeasible results do not replace best. +Promotion requires full exported SEC proof by default; `allow_unproven: True` +explicitly permits warning-labelled results without claiming full equivalence. +The general non-blocking warning policy is unchanged. + +`best/` holds an independent copy of the netlist, proof and measurements, not a +symlink into rolling history. Undo and pruning cannot delete it. Changing the +objective or physical setup requires a new comparison/session. Baseline and best +are protected independently of the rolling retention limit. + +## Validation + +The new [GCD packaged undo workflow](../.github/workflows/gcd-undo-verify.yml) +uses the same packaged tools and reference rewrite as the existing GCD workflow, +which is unchanged. It checks original/edited/restored Scope connectivity, full +SEC coverage, timing improvement, discarded-netlist deletion and best preservation. +It runs OpenROAD three times and requires restored setup, hold, TNS, area, power +report and DRC results to match this run's baseline exactly. + +```sh +python -m unittest discover -s tests -v +python scripts/versioned_session_regression.py --jupyter --work-dir runs/history-check +python scripts/gcd_undo_regression.py --work-dir runs/gcd-undo-check +``` + +Offline tests do not prove equivalence or physical results. The real history +regression also tests failed-edit recovery, Scope refresh, continued editing, +retention and rejection of stale native references after undo. diff --git a/tools/session_history.py b/tools/session_history.py new file mode 100644 index 0000000..ad38951 --- /dev/null +++ b/tools/session_history.py @@ -0,0 +1,189 @@ +"""Numbered, integrity-checked checkpoints; no native tools required here.""" + +from contextlib import contextmanager +import hashlib +import json +import math +from pathlib import Path +import shutil +import tempfile + + +def digest(path): + return hashlib.sha256(Path(path).read_bytes()).hexdigest() + + +def write_json(path, value): + path = Path(path) + data = json.dumps(value, indent=2, allow_nan=False) + "\n" + with tempfile.NamedTemporaryFile(mode="w", dir=path.parent, delete=False) as stream: + temporary = Path(stream.name) + stream.write(data) + try: + temporary.replace(path) + finally: + temporary.unlink(missing_ok=True) + + +class RevisionHistory: + def __init__(self, directory, retention=10, objective=None): + self.directory = Path(directory) + self.versions = self.directory / "versions" + self.versions.mkdir() + self.records = {} + self.active = None + self.high_water = -1 + self.leases = {} + self.best = None + self.context = None + self.objective = json.loads(json.dumps(objective)) if objective is not None else None + if self.objective is not None: + weights = self.objective.get("weights", {}) + if not weights or not all(self._number(x) for x in weights.values()): + raise ValueError("Objective requires finite numeric weights (lower score wins)") + for bounds in self.objective.get("bounds", {}).values(): + if not set(bounds) <= {"min", "max"} or not all(self._number(x) for x in bounds.values()): + raise ValueError("Objective bounds must be finite min/max values") + self.configure(retention) + + @staticmethod + def _number(value): + return type(value) in (int, float) and math.isfinite(value) + + def configure(self, retention): + if type(retention) is not int or retention < 1: + raise ValueError("Retention must be a positive integer") + self.retention = retention + write_json(self.directory / "config.json", {"retention": retention, "objective": self.objective}) + self.prune() + + def _pointer(self): + write_json(self.directory / "current.json", {"revision": self.active, + "high_water": self.high_water}) + + def publish(self, staging, revision, metadata): + if type(revision) is not int or revision <= self.high_water: + raise ValueError("Revision numbers must increase and cannot be reused") + staging = Path(staging) + if not (staging / "design.v").is_file() or not (staging / "design.v").stat().st_size: + raise ValueError("Missing checkpoint netlist") + record = dict(metadata, revision=revision, parent=self.active) + record["files"] = {str(p.relative_to(staging)): digest(p) + for p in staging.rglob("*") if p.is_file()} + write_json(staging / "manifest.json", record) + destination = self.versions / f"revision-{revision:04d}" + staging.rename(destination) + self.records[revision] = {"directory": destination, + "manifest_hash": digest(destination / "manifest.json")} + self.active = self.high_water = revision + self._pointer() + self.prune() + return self.get(revision) + + def get(self, revision=None): + revision = self.active if revision is None else revision + if type(revision) is not int or revision not in self.records: + raise ValueError("Revision is not retained in this session") + entry = self.records[revision] + directory = entry["directory"] + if digest(directory / "manifest.json") != entry["manifest_hash"]: + raise ValueError("Checkpoint manifest was modified") + record = json.loads((directory / "manifest.json").read_text()) + for name, expected in record["files"].items(): + path = directory / name + if path.is_symlink() or not path.resolve().is_relative_to(directory.resolve()) or digest(path) != expected: + raise ValueError("Checkpoint file was modified: " + name) + return dict(record, directory=str(directory), verilog_file=str(directory / "design.v")) + + @contextmanager + def acquire(self, revision=None): + record = self.get(revision) + key = record["revision"] + self.leases[key] = self.leases.get(key, 0) + 1 + try: + yield record + self.get(key) + finally: + self.leases[key] -= 1 + self.prune() + + def prune(self): + # Baseline is permanent; active tool inputs are never deleted mid-run. + recent = sorted(key for key in self.records if key != 0) + keep = set(recent[-self.retention:]) | {0, self.active} + for key in list(self.records): + if key not in keep and not self.leases.get(key): + shutil.rmtree(self.records.pop(key)["directory"]) + + def undo_target(self, discard=True): + if discard: + candidates = [key for key in self.records if key < self.active] + if not candidates: + raise ValueError("No previous checkpoint to restore") + if self.leases.get(self.active): + raise RuntimeError("Current revision is in use by another tool") + return self.get(max(candidates)) + return self.get() + + def finish_undo(self, target): + self.get(target) + discarded = [key for key in self.records if key > target] + if any(self.leases.get(key) for key in discarded): + raise RuntimeError("A discarded revision is still in use") + self.active = target + self._pointer() + for key in discarded: + shutil.rmtree(self.records.pop(key)["directory"]) + # high_water deliberately does not decrease after undo. + + def measure(self, revision, metrics, context, evidence): + record = self.get(revision) + if not metrics or not all(self._number(x) for x in metrics.values()): + raise ValueError("Measurements must be finite numeric values") + if not context or not evidence: + raise ValueError("Measurements require setup identity and evidence files") + if self.context is not None and context != self.context: + raise ValueError("Measurement setup differs from earlier results") + objective = self.objective + if objective: + required = set(objective["weights"]) | set(objective.get("bounds", {})) + if not required <= metrics.keys(): + raise ValueError("Missing objective metrics") + directory = Path(record["directory"]) / "measurements" + directory.mkdir(exist_ok=False) + for index, source in enumerate(evidence): + source = Path(source) + shutil.copyfile(source, directory / f"{index:03d}-{source.name}") + result = {"revision": revision, "design_sha256": record["files"]["design.v"], + "metrics": metrics, "context": context, "promoted": False} + self.context = json.loads(json.dumps(context)) + if objective: + score = sum(metrics[key] * weight for key, weight in objective["weights"].items()) + if not math.isfinite(score): + raise ValueError("Objective score is not finite") + eligible = record["export_proof"]["status"] == "proved" or ( + objective.get("allow_unproven", False) and record["export_proof"]["status"] == "warning") + for key, bounds in objective.get("bounds", {}).items(): + eligible &= (metrics[key] >= bounds.get("min", -math.inf) + and metrics[key] <= bounds.get("max", math.inf)) + result.update(score=score, eligible=bool(eligible)) + result["promoted"] = bool(eligible and (self.best is None or score < self.best["score"])) + write_json(directory / "result.json", result) + if result["promoted"]: + # A real copy, not a link into the rolling history. + temporary = Path(tempfile.mkdtemp(prefix=".best-", dir=self.directory)) + shutil.copytree(record["directory"], temporary, dirs_exist_ok=True) + best = self.directory / "best" + previous = self.directory / ".previous-best" + if best.exists(): + best.rename(previous) + try: + temporary.rename(best) + except BaseException: + if previous.exists(): + previous.rename(best) + raise + if previous.exists(): + shutil.rmtree(previous) + self.best = dict(result) + return result diff --git a/tools/versioned_session.py b/tools/versioned_session.py new file mode 100644 index 0000000..646e314 --- /dev/null +++ b/tools/versioned_session.py @@ -0,0 +1,212 @@ +"""Opt-in live editing with numbered Verilog checkpoints and checked undo.""" + +from contextlib import contextmanager +from concurrent.futures import ThreadPoolExecutor +from datetime import datetime, timezone +import json +from pathlib import Path +import shutil +import tempfile +import threading + +from tools.live_session import LiveDesignSession, SEC, _fingerprint +from tools.session_history import RevisionHistory, digest, write_json + + +class VersionedDesignSession(LiveDesignSession): + def _record(self): + record = super()._record() + if hasattr(self, "history"): + current = getattr(self, "_checkpoint_live_hash", None) == self._candidate_hash + record.update(netlist_revision=self.history.active if current else None, + last_saved_revision=self.history.active, retention=self.history.retention, + best_revision=self.history.best["revision"] if self.history.best else None) + write_json(self.directory / "status.json", record) + return record + + def __init__(self, reference, liberty_files, work_dir=None, *, retention=10, + objective=None, sessions_root="runs", timeout=600): + self._history_lock = threading.RLock() + if work_dir is None: + root = Path(sessions_root) + root.mkdir(parents=True, exist_ok=True) + stamp = datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%S.%fZ") + work_dir = root / ("session_" + stamp) + super().__init__(reference, liberty_files, work_dir, timeout=timeout) + try: + self.history = RevisionHistory(self.directory, retention, objective) + self.inputs = self.directory / "inputs" + self.inputs.mkdir() + self.reference_file = self.inputs / "golden.v" + shutil.copyfile(reference, self.reference_file) + self.reference_file.chmod(0o400) + self.libraries = [] + for i, source in enumerate(self._liberty_paths): + destination = self.inputs / f"cells-{i:03}.lib" + shutil.copyfile(source, destination) + destination.chmod(0o400) + self.libraries.append(destination) + self._input_hashes = {str(p): digest(p) for p in [self.reference_file, *self.libraries]} + if digest(self.reference_file) != self._source_hashes[str(Path(reference).resolve())]: + raise ValueError("Reference changed while starting the session") + if any(digest(dst) != self._source_hashes[str(src)] + for src, dst in zip(self._liberty_paths, self.libraries)): + raise ValueError("Library changed while starting the session") + self._undo_count = 0 + self._checkpoint_live_hash = None + self._checkpoint_epoch = 0 + super().verify() + self._checkpoint(initial=True) + except BaseException: + super().close() + raise + + def _check_inputs(self): + if any(digest(path) != sha for path, sha in self._input_hashes.items()): + raise ValueError("Session baseline or libraries were modified") + + def _file_sec(self, directory, candidate): + # Notebook cells already have an event loop. The file-based MCP client + # owns another loop; keep it off the kernel's thread, like the live client. + with ThreadPoolExecutor(max_workers=1) as executor: + return executor.submit(SEC.run_sec, directory, self.reference_file, candidate, + self.libraries, timeout=self.timeout).result() + + def _checkpoint(self, initial=False): + self._check_inputs() + staging = Path(tempfile.mkdtemp(prefix=".checkpoint-", dir=self.directory)) + try: + with self._inspection_access(): + before = self._candidate_hash + if initial: + shutil.copyfile(self.reference_file, staging / "design.v") + else: + self._candidate.dumpVerilog(str(staging), "design.v") + shutil.copyfile(self.directory / f"revision-{self.revision:04d}/edit.py", staging / "edit.py") + live_proof = json.loads(json.dumps(self.proof)) + proof = self._file_sec(staging / "export-proof", staging / "design.v") + self._check_inputs() + with self._inspection_access(): + if before != self._candidate_hash: + raise RuntimeError("Candidate changed during checkpoint verification") + record = self.history.publish(staging, self.revision, { + "top": self._candidate.getName(), "live_proof": live_proof, + "export_proof": proof, "candidate_reference": dict(self._candidate_ref), + "created_at": datetime.now(timezone.utc).isoformat(), + }) + self._checkpoint_live_hash = before + self._checkpoint_epoch += 1 + self._record() + return record + except BaseException as error: + if staging.exists(): + # Do not publish a failed/partial export as a selectable revision. + write_json(staging / "error.json", {"error": str(error)}) + raise + + def apply_edit(self, script): + with self._history_lock: + self._check_inputs() + result = super().apply_edit(script) + self._checkpoint() + return result + + def checkpoint(self, revision=None): + """Return an explicit historical version or the current saved design.""" + with self._history_lock: + with self._inspection_access(): + self._check_inputs() + if revision is None and self._checkpoint_live_hash != self._candidate_hash: + raise RuntimeError("Live candidate has unsaved changes; undo or repair it before current inspection") + return self.history.get(revision) + + @contextmanager + def use_checkpoint(self, revision=None): + """Pin the exact tool input for the duration of an external run/query.""" + with self._history_lock: + record = self.checkpoint(revision) + with self.history.acquire(record["revision"]) as leased: + yield leased + + def configure_history(self, *, retention): + with self._history_lock: + self.history.configure(retention) + + def record_measurement(self, metrics, *, context, evidence, revision=None): + with self._history_lock: + record = self.checkpoint(revision) + return self.history.measure(record["revision"], metrics, context, evidence) + + def undo(self): + """Restore first; only then delete the discarded netlist checkpoint.""" + from kepler_formal_mcp.session_bridge import SessionBridge + + with self._history_lock: + self._check_inputs() + self._undo_count += 1 + output = self.directory / f"undo-{self._undo_count:04d}" + output.mkdir() + with self._inspection_access(): + target = self.history.undo_target( + discard=self._checkpoint_live_hash == self._candidate_hash) + # Revalidate the actual persisted file before replacing any live objects. + file_proof = self._file_sec(output / "file-proof", target["verilog_file"]) + with self._operation: + self._check() + old_bridge = self._bridge + if self._pending or not old_bridge.lock.acquire(blocking=False): + raise RuntimeError("Native work is running; undo is blocked") + try: + self.history.get(target["revision"]) + self._check_inputs() + self.state, self.proof = "restoring", None + self._record() + # Naja can reuse native IDs on reload. Expire the old binding + # BEFORE destroying objects; old agent references must fail. + detached = self._client.call("close_session", {"session_id": self._session_id}) + if detached.get("status") != "success": + raise RuntimeError("Kepler could not detach before undo") + old_bridge.close() + db = self._candidate.getDB() + for library in list(db.getLibraries()): + if not library.isPrimitives(): + for design in list(library.getSNLDesigns()): + design.destroy() + db.loadVerilog([target["verilog_file"]]) + self._candidate = db.getTopDesign() + if self._candidate is None: + raise RuntimeError("Restored file did not produce a top design") + self._universe.setTopDesign(self._candidate) + self._candidate_hash = _fingerprint(self._candidate) + self._bridge = SessionBridge(output_dir=output / "formal").start() + self._golden_ref = self._bridge.design_reference(self._golden) + self._candidate_ref = self._bridge.design_reference(self._candidate) + self._check() + except BaseException: + self.state, self.proof = "invalid", None + self._record() + raise + finally: + old_bridge.lock.release() + attached = self._client.call("attach_session", { + "connection_file": str(self._bridge.connection_file)}) + write_json(output / "attachment-result.json", attached) + if attached.get("status") != "success" or attached.get("session_id") != self._bridge.session_id: + self.state, self.proof = "invalid", None + self._record() + raise RuntimeError("Kepler could not reattach after undo") + self._session_id = attached["session_id"] + proof = self._verify() + self.history.finish_undo(target["revision"]) + self._checkpoint_live_hash = self._candidate_hash + self._checkpoint_epoch += 1 + self._record() + result = {"restored_revision": target["revision"], "proof": proof, + "export_proof": file_proof, "attempt_counter": self.revision, + "reattach_required": True} + write_json(output / "result.json", result) + return result + + def close(self): + with self._history_lock: + super().close()