150 lines
5.3 KiB
Python
150 lines
5.3 KiB
Python
#!/usr/bin/env python3
|
|
"""null lane gate for the agentloop artifact -- one command, exit code IS the verdict.
|
|
|
|
python3 check.py
|
|
|
|
0 every leg passed
|
|
1 a graded fault (a leg reported one)
|
|
2 COULD NOT GRADE (python3 missing, the package will not import, ...)
|
|
|
|
Legs:
|
|
1. the package imports and exposes the documented API;
|
|
2. the unittest suite passes (chain integrity, tamper detection, argument
|
|
validation, loop budgets, determinism);
|
|
3. an INDEPENDENT tamper probe: build a transcript, mutate one payload byte on
|
|
disk, and require that verify_file names record 2 with reason hash_mismatch.
|
|
This leg does not reuse the suite's helpers -- it re-derives the expectation
|
|
from the file's own bytes, so a bug in the test fixtures cannot hide here.
|
|
|
|
This file resolves its own directory, so the copy in a clone grades that clone.
|
|
"""
|
|
|
|
import os
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
|
|
HERE = os.path.dirname(os.path.abspath(__file__))
|
|
PY = sys.executable or "python3"
|
|
|
|
|
|
def leg_import():
|
|
probe = os.path.join(HERE, "_probe_import.py")
|
|
with open(probe, "w", encoding="utf-8") as fh:
|
|
fh.write(
|
|
"import sys; sys.path.insert(0, %r)\n"
|
|
"import agentloop\n"
|
|
"missing = [n for n in ('Transcript','verify_file','Tool','ToolRegistry',"
|
|
"'AgentLoop','ModelReply','StopReason','ToolCall') if not hasattr(agentloop, n)]\n"
|
|
"sys.exit(3 if missing else 0)\n" % (os.path.join(HERE, ".."),)
|
|
)
|
|
try:
|
|
rc = subprocess.call([PY, probe])
|
|
finally:
|
|
os.unlink(probe)
|
|
if rc == 3:
|
|
return 1, "import leg: documented API is missing"
|
|
if rc != 0:
|
|
return 2, "import leg: probe exited %d" % rc
|
|
return 0, "import leg: agentloop imports, documented API present"
|
|
|
|
|
|
def leg_suite():
|
|
env = dict(os.environ)
|
|
env["PYTHONPATH"] = os.path.join(HERE, "..") + os.pathsep + env.get("PYTHONPATH", "")
|
|
proc = subprocess.Popen(
|
|
[PY, "-m", "unittest", "discover", "-s", "tests", "-t", HERE, "-q"],
|
|
cwd=HERE, env=env, stdout=subprocess.PIPE, stderr=subprocess.STDOUT,
|
|
)
|
|
out = proc.communicate()[0].decode("utf-8", "replace")
|
|
tail = [l for l in out.strip().splitlines() if l.strip()][-6:]
|
|
if proc.returncode != 0:
|
|
return 1, "suite leg: FAILED\n" + "\n".join(tail)
|
|
ran = [l for l in out.splitlines() if l.startswith("Ran ")]
|
|
return 0, "suite leg: PASS (%s)" % (ran[0] if ran else "no summary line")
|
|
|
|
|
|
def leg_tamper():
|
|
"""Independent probe: mutate a transcript on disk, require the right verdict."""
|
|
sys.path.insert(0, os.path.join(HERE, ".."))
|
|
try:
|
|
from agentloop import Transcript, canonical_json
|
|
from agentloop.transcript import KIND_ASSISTANT, KIND_FINAL, KIND_TOOL, KIND_USER
|
|
except Exception as exc:
|
|
return 2, "tamper leg: could not import agentloop (%s)" % exc
|
|
|
|
tr = Transcript()
|
|
tr.append(KIND_USER, {"text": "add 2 and 3"})
|
|
tr.append(KIND_ASSISTANT, {"text": "calling add", "tool_calls": []})
|
|
tr.append(KIND_TOOL, {"name": "add", "ok": True, "value": 5})
|
|
tr.append(KIND_FINAL, {"stop_reason": "no_tool_calls", "steps": 1, "tokens": 4})
|
|
|
|
fd, path = tempfile.mkstemp(suffix=".jsonl")
|
|
os.close(fd)
|
|
try:
|
|
with open(path, "wb") as fh:
|
|
fh.write(tr.bytes())
|
|
|
|
# Read the file's own bytes, flip one character of record 2's payload,
|
|
# and write it back. Nothing here uses the suite's helpers.
|
|
with open(path, "rb") as fh:
|
|
raw = fh.read()
|
|
target = raw.index(b'"value":5')
|
|
flipped = raw[:target] + b'"value":9' + raw[target + len(b'"value":5'):]
|
|
self_check = flipped != raw
|
|
with open(path, "wb") as fh:
|
|
fh.write(flipped)
|
|
|
|
# Re-verify from disk, using only verify_file.
|
|
from agentloop import verify_file
|
|
ok, violations, count = verify_file(path)
|
|
finally:
|
|
os.unlink(path)
|
|
|
|
if not self_check:
|
|
return 2, "tamper leg: the mutation did not change the file"
|
|
if ok:
|
|
return 1, "tamper leg: a mutated transcript VERIFIED -- the chain is decorative"
|
|
if not violations:
|
|
return 1, "tamper leg: failed but named no violation"
|
|
v = violations[0]
|
|
if v.get("seq") != 2 or v.get("reason") != "hash_mismatch":
|
|
return 1, "tamper leg: wrong verdict -- got %r" % (v,)
|
|
if count != 4:
|
|
return 1, "tamper leg: expected 4 records, counted %d" % count
|
|
return 0, "tamper leg: mutation caught at seq 2 (hash_mismatch), 4 records intact"
|
|
|
|
|
|
LEGS = (leg_import, leg_suite, leg_tamper)
|
|
|
|
|
|
def main():
|
|
if not PY:
|
|
print("COULD NOT GRADE: no python3 interpreter")
|
|
return 2
|
|
|
|
rc = 0
|
|
print("agentloop gate -- %s" % HERE)
|
|
for leg in LEGS:
|
|
try:
|
|
code, message = leg()
|
|
except Exception as exc:
|
|
code, message = 2, "%s leg: raised %s: %s" % (leg.__name__, type(exc).__name__, exc)
|
|
print(" [%s] %s" % ("ok" if code == 0 else ("FAIL" if code == 1 else "N/A"), message))
|
|
if code == 1:
|
|
rc = 1
|
|
elif code == 2 and rc == 0:
|
|
rc = 2
|
|
|
|
if rc == 0:
|
|
print("VERDICT: every leg passed")
|
|
elif rc == 1:
|
|
print("VERDICT: graded fault")
|
|
else:
|
|
print("VERDICT: could not grade")
|
|
return rc
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|