| """Orchestrate batch macro expansion verification across libraries.""" |
|
|
| from __future__ import annotations |
|
|
| import json |
| import time |
| from pathlib import Path |
|
|
| from cl_macros.ext.library_index import LibraryIndex, LibraryInfo |
| from cl_macros.verif.sbcl_bridge import SBCLVerifier |
|
|
| VERIFIED_DIR = Path("data/verified") |
|
|
|
|
| class BatchVerifier: |
| def __init__(self): |
| self.verifier = SBCLVerifier() |
| self.index = LibraryIndex() |
|
|
| def verify_library(self, lib_name: str, max_calls: int = 100) -> dict: |
| """Verify macros from one library. Returns summary.""" |
| lib = self.index.get(lib_name) |
| if not lib: |
| return {"library": lib_name, "status": "unknown-library"} |
|
|
| |
| ext_path = Path("data/extractions") / f"{lib_name}_extractions.jsonl" |
| cs_path = Path("data/extractions") / f"{lib_name}_call_sites.jsonl" |
|
|
| if not ext_path.exists(): |
| return {"library": lib_name, "status": "no-extractions"} |
|
|
| extras = [] |
| with open(ext_path) as f: |
| for line in f: |
| extras.append(json.loads(line.strip())) |
|
|
| call_sites = [] |
| if cs_path.exists(): |
| with open(cs_path) as f: |
| for line in f: |
| call_sites.append(json.loads(line.strip())) |
|
|
| system_name = lib.systems[0] if lib.systems else lib_name |
|
|
| |
| seen = set() |
| calls_to_verify = [] |
| for cs in call_sites: |
| key = (cs["macro_name"], cs["call_form"]) |
| if key not in seen: |
| seen.add(key) |
| calls_to_verify.append(cs) |
| if len(calls_to_verify) >= max_calls: |
| break |
|
|
| print(f" Verifying {len(calls_to_verify)} unique calls (from {len(call_sites)} total)") |
| start = time.time() |
| results = self.verifier.verify_library( |
| lib_name, system_name, calls_to_verify |
| ) |
| elapsed = time.time() - start |
| print(f" Completed in {elapsed:.1f}s") |
|
|
| |
| output_dir = VERIFIED_DIR |
| output_dir.mkdir(parents=True, exist_ok=True) |
| output_path = output_dir / f"{lib_name}_verified.jsonl" |
| with open(output_path, "w") as f: |
| for r in results: |
| f.write(json.dumps(r) + "\n") |
|
|
| verified = sum(1 for r in results if r["status"] == "verified") |
| failed = sum(1 for r in results if r["status"] != "verified") |
|
|
| return { |
| "library": lib_name, |
| "macros_extracted": len(extras), |
| "calls_verified": len(results), |
| "verified_count": verified, |
| "failed_count": failed, |
| "elapsed": elapsed, |
| } |
|
|
| def verify_all(self, tier: str = "tier1", max_calls_per_lib: int = 100) -> dict: |
| """Verify all libraries in a tier.""" |
| libs = self.index.list_libraries(tier) |
| summaries = [] |
| for lib in libs: |
| print(f"\n--- {lib.name} ---") |
| try: |
| summary = self.verify_library(lib.name, max_calls_per_lib) |
| summaries.append(summary) |
| print(f" Verified: {summary.get('verified_count', 0)}, " |
| f"Failed: {summary.get('failed_count', 0)}") |
| except Exception as e: |
| print(f" FAILED: {e}") |
| summaries.append({"library": lib.name, "status": "error", "error": str(e)}) |
|
|
| |
| manifest = { |
| "tier": tier, |
| "libraries": len(libs), |
| "summaries": summaries, |
| "total_verified": sum(s.get("verified_count", 0) for s in summaries), |
| "total_failed": sum(s.get("failed_count", 0) for s in summaries), |
| } |
| manifest_path = VERIFIED_DIR / "verification_manifest.json" |
| with open(manifest_path, "w") as f: |
| json.dump(manifest, f, indent=2) |
|
|
| print(f"\nManifest: {manifest_path}") |
| return manifest |
|
|