"""Smoke verification probe for candidate remote execution bundles. Checks: 1. Python interpreter and ABI compatibility 2. Target platform matches expected platform (never reuse Mac environment on Linux) 3. klbr_runtime importability and protocol readiness 4. Optional capabilities (e.g. dill): explicitly reports availability or unavailability 5. Working directory accessibility """ import argparse import json import os import platform import sys def main() -> int: parser = argparse.ArgumentParser(description="Smoke verification for klbr candidate bundle") parser.add_argument("--bundle-dir", required=False, default=".") parser.add_argument("--expected-platform", required=False, default=None) parser.add_argument("--workspace", required=False, default=None) parser.add_argument("--require-skills", nargs="*", default=[]) args = parser.parse_args() # Add bundle python directories to sys.path if present candidates = [ args.bundle_dir, os.path.join(args.bundle_dir, "python"), os.path.join(args.bundle_dir, "python", "klbr-runtime", "src"), os.path.join(args.bundle_dir, "src"), ] for c in candidates: if os.path.isdir(c) and c not in sys.path: sys.path.insert(0, c) actual_platform = platform.system() if args.expected_platform and args.expected_platform.lower() != actual_platform.lower(): sys.stderr.write( f"Platform mismatch: expected {args.expected_platform}, actual {actual_platform}\n" ) return 2 # Workspace check if args.workspace: if not os.path.isdir(args.workspace): sys.stderr.write(f"Workspace not found or not a directory: {args.workspace}\n") return 3 # Smoke import klbr_runtime try: import klbr_runtime # noqa: F401 except ImportError as e: sys.stderr.write(f"Failed to import klbr_runtime: {e}\n") return 4 capabilities = ["klbr_runtime", "runner"] unavailable = [] # Check dill try: import dill # noqa: F401 capabilities.append("dill") except ImportError: unavailable.append("dill") # Check required skills if any for skill in args.require_skills: skill_path = os.path.join(args.bundle_dir, "skills", skill) if os.path.exists(skill_path): capabilities.append(f"skill:{skill}") else: sys.stderr.write(f"Missing required skill: {skill} at {skill_path}\n") return 5 report = { "status": "ok", "platform": actual_platform, "machine": platform.machine(), "python_version": platform.python_version(), "python_path": sys.executable, "capabilities": capabilities, "unavailable_capabilities": unavailable, } print(json.dumps(report)) return 0 if __name__ == "__main__": sys.exit(main())