Spaces:
Running
Running
File size: 7,067 Bytes
7dcbd04 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 | import pytest
from mouh.jev.audit import AuditLog
from mouh.jev.types import (
JevMessageType as T, JevProofObligation, TaskPayload, VERIFICATION_THRESHOLD,
)
from mouh.verification import (
CallableGenerator, CorrectionStore, NullGenerator, TestTimeVerificationLoop, parse_files,
)
BUGGY = "def add(a, b):\n return a - b\n"
FIXED = "def add(a, b):\n return a + b\n"
TESTS = "from calc import add\n\ndef test_add():\n assert add(2, 3) == 5\n\ndef test_zero():\n assert add(0, 0) == 0\n"
def make_task(code=None, tests=TESTS, proofs=()):
return TaskPayload(instruction="écrire add(a, b)", candidate_files={"calc.py": code} if code else {},
test_files={"test_calc.py": tests} if tests else {}, proofs=tuple(proofs))
def types_of(outcome):
return [m.type for m in outcome.messages]
def test_threshold_constant():
assert VERIFICATION_THRESHOLD == 0.85
def test_certified_on_first_attempt_emits_certificate(tmp_path):
audit = AuditLog(tmp_path / "audit.jsonl")
loop = TestTimeVerificationLoop(audit=audit)
out = loop.run(make_task(FIXED))
assert out.status == "certified" and out.attempts == 1
assert out.final.certificate.score >= 0.85 and "tests" in out.final.certificate.evidence
assert T.FINAL_ANSWER in types_of(out) and out.abort is None
assert audit.verify() == (True, None)
# chaque saut a été journalisé
kinds = [r["event"]["kind"] for r in audit.records()]
assert kinds.count("accepted") == len(out.messages)
def test_correction_loop_repairs_bug_and_remembers_fix(tmp_path):
calls = []
def gen(req):
calls.append(req)
assert req.message_type is T.CORRECTION_REQUEST and req.attempt == 2
assert req.experts[0] == 5 # expert n°5 (refactoring) prioritaire
assert any(d.error_type.value == "test_failure" for d in req.diagnostics)
return {"calc.py": FIXED}
store = CorrectionStore(tmp_path / "fixes.jsonl")
loop = TestTimeVerificationLoop(CallableGenerator(gen), corrections=store)
out = loop.run(make_task(BUGGY))
assert out.status == "certified" and out.attempts == 2 and len(calls) == 1
assert [a["certified"] for a in out.attempts_log] == [False, True]
assert out.attempts_log[0]["score"] < 1.0
assert T.CORRECTION_REQUEST in types_of(out)
assert len(store) == 1
# la correction mémorisée devient un indice et une donnée d'affinage
again = TestTimeVerificationLoop(CallableGenerator(lambda r: {"calc.py": FIXED} if r.hints else pytest.fail("pas d'indice")),
corrections=CorrectionStore(tmp_path / "fixes.jsonl"))
assert again.run(make_task(BUGGY)).status == "certified"
n = store.export_sft(tmp_path / "sft.jsonl")
assert n == 1 and "<diagnostics>" in (tmp_path / "sft.jsonl").read_text()
def test_exhausted_budget_emits_no_final_text():
loop = TestTimeVerificationLoop(CallableGenerator(lambda r: {"calc.py": BUGGY + f"# v{r.attempt}\n"}), max_attempts=3)
out = loop.run(make_task(BUGGY))
assert out.status == "exhausted" and out.final is None and out.attempts == 3
assert T.FINAL_ANSWER not in types_of(out) and out.abort.reason.startswith("budget de tentatives épuisé")
def test_no_progress_detected():
loop = TestTimeVerificationLoop(CallableGenerator(lambda r: {"calc.py": BUGGY}), max_attempts=5)
out = loop.run(make_task(BUGGY))
assert out.status == "aborted" and "aucun progrès" in out.abort.reason and out.attempts == 1
def test_null_generator_only_verifies_given_code():
out = TestTimeVerificationLoop(NullGenerator()).run(make_task(BUGGY))
assert out.status == "aborted" and out.final is None
def test_no_oracle_means_nothing_is_certified():
out = TestTimeVerificationLoop().run(make_task(FIXED, tests=""))
assert out.status == "aborted" and "aucun oracle" in out.abort.reason and out.final is None
def test_syntax_error_then_repaired():
seq = iter([{"calc.py": FIXED}])
out = TestTimeVerificationLoop(CallableGenerator(lambda r: next(seq))).run(make_task("def add(a, b:\n return\n"))
assert out.status == "certified" and out.attempts == 2
assert out.attempts_log[0]["components"]["static_analysis"] == 0.0
def test_policy_violation_blocks_certification():
evil = "import subprocess\ndef add(a, b):\n return a + b\n"
out = TestTimeVerificationLoop(NullGenerator()).run(make_task(evil))
assert out.status == "aborted" and out.final is None
assert out.attempts_log[0]["components"]["static_analysis"] == 0.0
def test_formal_proof_obligations_participate():
good = JevProofObligation(name="monotone", declarations={"x": "Int"}, assumptions=([">", "x", 0],),
goal=[">", ["+", "x", 1], 1])
bad = JevProofObligation(name="faux", declarations={"x": "Int"}, assumptions=([">", "x", 0],), goal=[">", "x", 5])
ok = TestTimeVerificationLoop().run(make_task(FIXED, proofs=[good]))
assert ok.status == "certified" and "proofs" in ok.final.certificate.evidence
ko = TestTimeVerificationLoop().run(make_task(FIXED, proofs=[good, bad]))
assert ko.final is None and ko.attempts_log[0]["components"]["proofs"] == 0.5
def test_proofs_alone_can_certify_without_tests():
p = JevProofObligation(name="p", declarations={"a": "Int"}, goal=["=", ["+", "a", 0], "a"])
out = TestTimeVerificationLoop().run(TaskPayload(instruction="x", candidate_files={"m.py": "x = 1\n"}, proofs=(p,)))
assert out.status == "certified"
def test_generator_crash_is_contained_and_reported():
def boom(req):
raise RuntimeError("générateur en panne")
out = TestTimeVerificationLoop(CallableGenerator(boom)).run(make_task(BUGGY))
assert out.final is None and out.status == "aborted"
assert out.abort.diagnostics and out.abort.diagnostics[0].error_type.value == "handler_failure"
def test_generator_proposes_first_attempt_when_no_candidate():
out = TestTimeVerificationLoop(CallableGenerator(lambda r: {"calc.py": FIXED})).run(make_task(None))
assert out.status == "certified" and out.attempts == 1
code_msgs = [m for m in out.messages if m.type is T.CODE_PROPOSAL]
assert code_msgs[0].payload.origin == "model"
def test_all_hops_are_schema_validated_and_routed_through_bus():
out = TestTimeVerificationLoop().run(make_task(FIXED))
seq = types_of(out)
for needed in (T.TASK_SUBMISSION, T.ROUTE_REQUEST, T.ROUTE_DECISION, T.CODE_PROPOSAL, T.TOOL_CALL,
T.TOOL_RESULT, T.VERIFICATION_RESULT, T.FINAL_ANSWER):
assert needed in seq, needed
assert seq.index(T.TOOL_CALL) < seq.index(T.TOOL_RESULT) < seq.index(T.VERIFICATION_RESULT) < seq.index(T.FINAL_ANSWER)
def test_parse_files_formats():
txt = "### FILE: a.py\n```python\nx = 1\n```\n### END\n### FILE: b/c.py\ny = 2\n### END\n"
assert parse_files(txt) == {"a.py": "x = 1\n", "b/c.py": "y = 2\n"}
assert parse_files("```python\nz = 3\n```") == {"solution.py": "z = 3\n"}
assert parse_files(" ") == {}
|