run_unit.py

#!/usr/bin/env python3
# SPDX-License-Identifier: GPL-2.0
"""Charon-pass harness: assemble a scratch crate for one unit against the shared
`kernel` stub, run `charon cargo --preset=aeneas`, capture outcome + first error.

Usage: run_unit.py <unit> <root_rel> <src_dir_under_linux> [extra_crate=...]
  <unit>       result key
  <root_rel>   crate-root .rs relative to the copied src tree (becomes lib root)
  <src_dir>    dir (or single file) under ~/linux to copy into scratch/<unit>/src

Emits one CSV line (no header) to stdout:
  unit,outcome,first_error,failure_category,minutes_spent,has_unions
"""
import os, sys, shutil, subprocess, time, glob, re, csv

HOME = os.path.expanduser("~")
LINUX = os.path.join(HOME, "linux")
ROOT = os.path.join(HOME, "kernel-rust-coverage-map")
SCRATCH = os.path.join(ROOT, "scratch")
STUB = os.path.join(ROOT, "stubs", "kernel")
CHARON = os.path.join(HOME, "charon", "bin", "charon")
CAP_SECS = 15 * 60

def categorize(err):
    e = err.lower()
    if "module!" in err or "bitfield!" in err or "declare_drm_ioctls" in err or "kunit" in err.lower():
        return "macro expansion"
    if re.search(r"cannot find macro|expected macro", e):
        return "macro expansion"
    if re.search(r"proc-macro|procedural macro|custom derive|derive macro", e):
        return "proc-macro"
    if re.search(r"unresolved import|cannot find (type|function|value|trait|module|struct|enum)|no method named|no function|no associated item|has no member|is not a member|unknown|no variant", e):
        return "missing kernel API"
    if re.search(r"extern|ffi|c-variadic|foreign", e):
        return "FFI boundary"
    if re.search(r"feature|nightly|#!\[feature|unstable", e):
        return "unsupported language feature"
    if "charon" in e and ("panic" in e or "unsupported" in e or "unimplemented" in e):
        return "unsupported language feature"
    return "other"

def first_error_line(out):
    for line in out.splitlines():
        s = line.strip()
        if s.startswith("error[") or s.startswith("error:") or s.startswith("error "):
            return s[:200]
    for line in out.splitlines():
        if "error" in line.lower():
            return line.strip()[:200]
    return ""

def main():
    unit = sys.argv[1]
    root_rel = sys.argv[2]
    src = sys.argv[3]
    extra_deps = [a for a in sys.argv[4:]]
    crate_name = re.sub(r"[^a-zA-Z0-9_]", "_", unit)
    # honor #![crate_name = "..."] in the root source so cargo's --crate-name matches
    _root_abs = os.path.join(LINUX, src) if os.path.isfile(os.path.join(LINUX, src)) else os.path.join(LINUX, src, root_rel)
    if os.path.isfile(_root_abs):
        m = re.search(r'#!\[crate_name\s*=\s*"([^"]+)"\]', open(_root_abs, errors="replace").read())
        if m:
            crate_name = m.group(1)
    dst = os.path.join(SCRATCH, unit)
    if os.path.exists(dst):
        shutil.rmtree(dst)
    os.makedirs(os.path.join(dst, "src"))

    # Shared preprocessing (NOT per-target): strip kernel proc-macro attributes
    # that are pure codegen wrappers, so extraction sees the underlying logic.
    STRIP_ATTRS = re.compile(r'^\s*#\[(export|vtable|pinned_drop)\b[^\]]*\]\s*$', re.M)
    def preprocess(text):
        return STRIP_ATTRS.sub('', text)

    def copy_pp(srcf, tgtf):
        os.makedirs(os.path.dirname(tgtf), exist_ok=True)
        open(tgtf, "w").write(preprocess(open(srcf, encoding="utf-8", errors="replace").read()))

    src_abs = os.path.join(LINUX, src)
    if os.path.isfile(src_abs):
        copy_pp(src_abs, os.path.join(dst, "src", os.path.basename(src_abs)))
        root_rel = os.path.basename(src_abs)
    else:
        for f in glob.glob(os.path.join(src_abs, "**", "*.rs"), recursive=True):
            rel = os.path.relpath(f, src_abs)
            copy_pp(f, os.path.join(dst, "src", rel))

    deps = f'kernel = {{ path = "{STUB}" }}\n'
    for d in extra_deps:
        name, path = d.split("=", 1)
        deps += f'{name} = {{ path = "{path}" }}\n'

    cargo = f"""[package]
name = "{crate_name}"
version = "0.1.0"
edition = "2021"

[lib]
path = "src/{root_rel}"

[dependencies]
{deps}"""
    open(os.path.join(dst, "Cargo.toml"), "w").write(cargo)

    # has_unions: from inventory.csv
    has_unions = "no"
    inv = os.path.join(ROOT, "inventory.csv")
    if os.path.exists(inv):
        for r in csv.DictReader(open(inv)):
            if r["unit"] == unit:
                has_unions = "yes" if r["union_touch"] == "yes" else "no"
                break

    env = dict(os.environ)
    env["PATH"] = os.path.join(HOME, "charon", "bin") + ":" + env["PATH"]
    t0 = time.time()
    try:
        p = subprocess.run([CHARON, "cargo", "--preset=aeneas"], cwd=dst, env=env,
                           capture_output=True, text=True, timeout=CAP_SECS)
        out = p.stdout + "\n" + p.stderr
        rc = p.returncode
    except subprocess.TimeoutExpired as e:
        mins = round((time.time() - t0) / 60, 1)
        print(f'{unit},TIMEBOX,"hit {CAP_SECS//60}min cap",timebox,{mins},{has_unions}')
        return
    mins = round((time.time() - t0) / 60, 1)

    llbc = glob.glob(os.path.join(dst, "*.llbc")) + glob.glob(os.path.join(dst, "llbc", "*.llbc"))
    if rc == 0 and llbc:
        print(f"{unit},CLEAN,,,{mins},{has_unions}")
    else:
        err = first_error_line(out).replace('"', "'")
        cat = categorize(out)
        print(f'{unit},FAIL,"{err}",{cat},{mins},{has_unions}')
    # keep log
    open(os.path.join(dst, "charon.log"), "w").write(out)

if __name__ == "__main__":
    main()