ive harnessed the harness
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197#!/usr/bin/env python3"""Run starter checks without a shell or package installation.
Default: real Python subprocess + shared schema/fixture tests.--rust: also run the authored Rust tests. Missing Cargo is an ERROR, not a skip.--format: explicitly allow rustfmt to edit only the new Rust crate."""from __future__ import annotations
import argparseimport datetimeimport jsonimport osfrom pathlib import Pathimport platformimport shutilimport sqlite3import subprocessimport sysimport tempfileimport timeimport tomllibimport zipfile
ROOT = Path(__file__).resolve().parents[1]
def static_contracts() -> dict: """Only manifest consistency, not dependency resolution or Rust compilation.""" workspace = tomllib.loads((ROOT / "Cargo.toml").read_text()) lock = tomllib.loads((ROOT / "Cargo.lock").read_text()) assert "klbr-runtime" in workspace["workspace"]["members"] packages = lock["package"] entry = next(p for p in packages if p["name"] == "klbr-runtime") for dependency in entry["dependencies"]: parts = dependency.split() matches = [p for p in packages if p["name"] == parts[0] and (len(parts) == 1 or p["version"] == parts[1])] assert len(matches) == 1, f"ambiguous/missing lock entry: {dependency}" crate = tomllib.loads((ROOT / "klbr-runtime/Cargo.toml").read_text()) assert set(crate["dependencies"]) == {d.split()[0] for d in entry["dependencies"]} tomllib.loads((ROOT / "python/klbr-runtime/pyproject.toml").read_text()) return {"status": "passed", "meaning": "manifest/lock-reference consistency only; Cargo resolver not run"}
def main() -> int: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--rust", action="store_true", help="run Cargo tests and lint checks") parser.add_argument("--format", action="store_true", help="run rustfmt in write mode; requires --rust") parser.add_argument("--portable-linker", action="store_true", help="override the existing Linux clang/wild config with cc") parser.add_argument("--report", type=Path, help="write machine-readable evidence and separate step logs") args = parser.parse_args() if sys.version_info < (3, 11): parser.error("Python >=3.11 is required") if args.format and not args.rust: parser.error("--format requires --rust") if os.name != "posix": parser.error("this first slice requires Unix-domain sockets (Linux/macOS)")
environment = os.environ.copy() environment["PYTHONPATH"] = str(ROOT / "python/klbr-runtime/src") environment["PYTHONDONTWRITEBYTECODE"] = "1" environment["KLBR_PYTHON"] = sys.executable if args.portable_linker: # CARGO_ENCODED_RUSTFLAGS takes precedence over .cargo/config.toml. # Empty intentionally clears --ld-path=wild, independently of Nushell. environment["CARGO_ENCODED_RUSTFLAGS"] = "" environment["CARGO_TARGET_X86_64_UNKNOWN_LINUX_GNU_LINKER"] = "cc"
cargo_bin = Path.home() / ".cargo/bin" if cargo_bin.exists() and str(cargo_bin) not in environment.get("PATH", ""): environment["PATH"] = f"{cargo_bin}:{environment.get('PATH', '')}" os.environ["PATH"] = environment["PATH"]
report = { "created_at_utc": datetime.datetime.now(datetime.timezone.utc).isoformat(), "python": sys.version, "platform": platform.platform(), "python_sqlite": sqlite3.sqlite_version, "python_sqlite_scope": "serial in-memory migration tests only; not runtime WAL qualification", "cargo_available": shutil.which("cargo") is not None, "rustc_available": shutil.which("rustc") is not None, "rust_checks_requested": args.rust, "steps": [], } try: report["manifests"] = static_contracts() except (AssertionError, ValueError, StopIteration, KeyError) as error: report["manifests"] = {"status": "failed", "error": str(error)}
def run(name: str, command: list[str]) -> None: print(f"\n== {name} ==", flush=True) started = time.monotonic() try: completed = subprocess.run(command, cwd=ROOT, env=environment, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True, errors="replace", check=False) code, output = completed.returncode, completed.stdout except OSError as error: code, output = 127, str(error) print(output, end="" if output.endswith("\n") else "\n", flush=True) step = {"name": name, "argv": command, "exit_code": code, "elapsed_seconds": round(time.monotonic() - started, 3)} if args.report: args.report.parent.mkdir(parents=True, exist_ok=True) log = args.report.with_name(f"{args.report.stem}-{name}.log") log.write_text(output) step["log"] = log.name report["steps"].append(step)
run("python", [sys.executable, "-B", "-m", "unittest", "discover", "-s", "python/klbr-runtime/tests", "-v"])
def run_wheel_smoke() -> None: print("\n== wheel-packaging-smoke ==", flush=True) started = time.monotonic() with tempfile.TemporaryDirectory(prefix="klbr-wheel-smoke-") as tmp: work = Path(tmp) wheels_dir = work / "wheels" wheels_dir.mkdir() src = work / "source" shutil.copytree(ROOT / "python/klbr-runtime", src, ignore=shutil.ignore_patterns("__pycache__", "build", "*.egg-info"))
uv = shutil.which("uv") pkg_venv_py = Path("/tmp/klbr-pkg-venv/bin/python") if uv: build_cmd = [uv, "build", "--wheel", str(src), "--out-dir", str(wheels_dir)] elif pkg_venv_py.exists(): build_cmd = [str(pkg_venv_py), "-m", "pip", "wheel", "--no-deps", "--no-build-isolation", "--no-index", str(src), "-w", str(wheels_dir)] else: build_cmd = [sys.executable, "-m", "pip", "wheel", "--no-deps", "--no-build-isolation", "--no-index", str(src), "-w", str(wheels_dir)]
build_res = subprocess.run(build_cmd, cwd=work, capture_output=True, text=True, check=False) if build_res.returncode != 0: output = f"Wheel build failed:\n{build_res.stdout}\n{build_res.stderr}" print(output, flush=True) step = {"name": "wheel-packaging-smoke", "argv": build_cmd, "exit_code": build_res.returncode, "elapsed_seconds": round(time.monotonic() - started, 3)} report["steps"].append(step) return
wheels = list(wheels_dir.glob("*.whl")) if not wheels: print("No wheel produced", flush=True) step = {"name": "wheel-packaging-smoke", "argv": build_cmd, "exit_code": 1, "elapsed_seconds": round(time.monotonic() - started, 3)} report["steps"].append(step) return
installed = work / "installed" with zipfile.ZipFile(wheels[0]) as z: z.extractall(installed)
smoke_env = dict(os.environ, PYTHONPATH=str(installed), PYTHONDONTWRITEBYTECODE="1") import_cmd = [sys.executable, "-B", "-c", "import klbr; tpls = klbr.templates.list_templates(); print(f'templates in wheel: {tpls}'); assert len(tpls) == 3"] import_res = subprocess.run(import_cmd, cwd=work, env=smoke_env, capture_output=True, text=True, check=False) output = import_res.stdout + import_res.stderr print(output, end="" if output.endswith("\n") else "\n", flush=True) step = {"name": "wheel-packaging-smoke", "argv": import_cmd, "exit_code": import_res.returncode, "elapsed_seconds": round(time.monotonic() - started, 3)} if args.report: args.report.parent.mkdir(parents=True, exist_ok=True) log = args.report.with_name(f"{args.report.stem}-wheel-smoke.log") log.write_text(output) step["log"] = log.name report["steps"].append(step)
run_wheel_smoke() if args.rust: if not report["cargo_available"] or not report["rustc_available"]: report["steps"].append({"name": "rust", "exit_code": 127, "reason": "Cargo/rustc unavailable; no Rust tests or integration checks ran"}) else: run("rust-version", ["rustc", "--version", "--verbose"]) run("rustfmt", ["cargo", "fmt", "-p", "klbr-runtime"] + ([] if args.format else ["--", "--check"])) run("rust-tests", ["cargo", "test", "--locked", "-p", "klbr-runtime", "--all-targets"]) # Warnings are visible; denying all Clippy style lints before the # first toolchain pass would hide more useful compilation evidence. run("clippy", ["cargo", "clippy", "--locked", "-p", "klbr-runtime", "--all-targets"]) else: report["rust_status"] = "not run; pass --rust on a machine with Cargo/rustc" passed = report["manifests"]["status"] == "passed" and all(s["exit_code"] == 0 for s in report["steps"]) report["requested_checks_passed"] = passed if args.report: args.report.parent.mkdir(parents=True, exist_ok=True) args.report.write_text(json.dumps(report, indent=2) + "\n") print(f"\nEvidence: {args.report}") return 0 if passed else 1
if __name__ == "__main__": raise SystemExit(main())