"""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())