Text · original UTF-8 line locators · 2.7 KiB
"""Local stand-in for the TfP harness: builds /input-like tree, runs checker.py, prints result.json."""
import json, os, shutil, subprocess, sys, tempfile
here = os.path.dirname(os.path.abspath(__file__))
def run(ref, spec, sub, label):
d = tempfile.mkdtemp(prefix="tfp-run-")
inp, out = os.path.join(d, "input"), os.path.join(d, "output")
os.makedirs(os.path.join(inp, "artifacts", "circuit")); os.makedirs(os.path.join(inp, "reference", "reference_circuit")); os.makedirs(os.path.join(inp, "reference", "spec")); os.makedirs(out)
rp = shutil.copy(ref, os.path.join(inp, "reference", "reference_circuit", os.path.basename(ref)))
sp = shutil.copy(spec, os.path.join(inp, "reference", "spec", os.path.basename(spec)))
ap = shutil.copy(sub, os.path.join(inp, "artifacts", "circuit", os.path.basename(sub)))
manifest = {"version": "tfp.verification-manifest/v0.1", "run_id": "cfx_local", "verifier_contract_id": "vct_local", "stage_key": "evaluate", "problem_revision_id": "prv_local",
"submission": {"id": "sub_local", "summary_md": None, "summary_included": False},
"artifacts": [{"role": "circuit", "artifact_id": None, "path": ap, "filename": os.path.basename(sub), "content_hash": "sha256:0", "size_bytes": os.path.getsize(sub), "media_type": "text/plain"}],
"reference": [{"key": "reference_circuit", "path": rp, "filename": os.path.basename(ref), "content_hash": "sha256:0", "size_bytes": os.path.getsize(ref), "media_type": "text/plain"},
{"key": "spec", "path": sp, "filename": os.path.basename(spec), "content_hash": "sha256:0", "size_bytes": os.path.getsize(spec), "media_type": "application/json"}],
"prior_stage_results": [], "result_kind": "score", "limits": {"vcpu": 4, "memory_mb": 12288, "disk_mb": 20000, "timeout_seconds": 600, "network": "none"}}
json.dump(manifest, open(os.path.join(inp, "manifest.json"), "w"))
p = subprocess.run([sys.executable, os.path.join(here, "checker.py")], env={**os.environ, "TFP_INPUT": inp, "TFP_OUTPUT": out}, capture_output=True, text=True, timeout=600)
res = json.load(open(os.path.join(out, "result.json"))) if os.path.exists(os.path.join(out, "result.json")) else {"result": "error", "reason": p.stderr[-500:]}
ev = json.load(open(os.path.join(out, "evidence", "summary.json"))) if os.path.exists(os.path.join(out, "evidence", "summary.json")) else {}
print(f"[{label}] exit={p.returncode} {res} elapsed={ev.get('elapsed_seconds')} qcec={ev.get('qcec_check_time_seconds')}")
if p.returncode: print(p.stderr[-800:])
shutil.rmtree(d)
return res
if __name__ == "__main__":
run(sys.argv[1], sys.argv[2], sys.argv[3], sys.argv[4] if len(sys.argv) > 4 else "run")