File size: 7,472 Bytes
e64c5d2 | 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 151 152 153 154 155 156 157 158 | """Action-level checks; scripted proposers here are fixtures, not learned results."""
import copy
import pytest
from lightnest.iterative_patch import (ByteEdit, PatchDecision, ValidationDecision,
apply_edits, replay_audit, run_iterative_patch)
def test_repeated_proposal_sees_the_modified_program_and_can_stop():
seen = []
def proposer(request, current, step, history):
seen.append(current)
if b"a" not in current:
return PatchDecision(stop=True)
offset = current.index(b"a")
return PatchDecision((ByteEdit(offset, b"a", b"b"),))
one = run_iterative_patch("replace all a", b"aca", proposer, max_steps=1)
assert one.final == b"bca" and one.stop_reason == "step_budget"
seen.clear()
many = run_iterative_patch("replace all a", b"aca", proposer, max_steps=4)
assert seen == [b"aca", b"bca", b"bcb"]
assert many.final == b"bcb" and many.accepted_steps == 2
assert many.stop_reason == "predictor_stop" and not many.complete
assert replay_audit(b"aca", many.audit()) == b"bcb"
def test_stale_second_span_rolls_back_the_entire_transaction():
calls = []
def proposer(request, current, step, history):
calls.append(current)
if not history:
return PatchDecision((ByteEdit(0, b"a", b"x"), ByteEdit(2, b"z", b"y")))
assert history[-1].status == "rejected_error"
return PatchDecision(stop=True)
result = run_iterative_patch("change", b"abc", proposer)
assert calls == [b"abc", b"abc"]
assert result.final == b"abc" and result.accepted_steps == 0
assert "stale" in result.history[0].error
assert replay_audit(b"abc", result.audit()) == b"abc"
def test_validator_rejection_is_visible_and_preserves_parent():
seen = []
def proposer(request, current, step, history):
seen.append(current)
if step == 0:
return PatchDecision((ByteEdit(0, b"a", b"!"),))
if step == 1:
assert history[-1].validation.reason == "compiler failed"
return PatchDecision((ByteEdit(0, b"a", b"b"),))
return PatchDecision(stop=True)
def validator(request, parent, candidate, decision):
return ValidationDecision(candidate != b"!", reason="compiler failed" if candidate == b"!" else "compiled")
result = run_iterative_patch("change", b"a", proposer, validator=validator)
assert seen == [b"a", b"a", b"b"] and result.final == b"b"
assert [t.status for t in result.history] == ["rejected", "accepted", "stopped"]
assert replay_audit(b"a", result.audit()) == b"b"
def test_variable_length_spans_and_current_offsets():
def proposer(request, current, step, history):
if step == 0:
return PatchDecision((ByteEdit(1, b"", b"XY"), ByteEdit(2, b"c", b"")))
if step == 1:
assert current == b"aXYb"
return PatchDecision((ByteEdit(3, b"b", b"B"),))
return PatchDecision(stop=True)
result = run_iterative_patch("change", b"abc", proposer, max_output_bytes=8)
assert result.final == b"aXYB"
assert replay_audit(b"abc", result.audit()) == result.final
@pytest.mark.parametrize("edits", [
(ByteEdit(0, b"ab", b"x"), ByteEdit(1, b"b", b"y")),
(ByteEdit(0, b"", b"x"), ByteEdit(0, b"a", b"y")),
(ByteEdit(-1, b"", b"x"),),
(ByteEdit(4, b"", b"x"),),
(ByteEdit(0, b"a", b"a"),),
])
def test_invalid_spans_are_atomic(edits):
with pytest.raises(ValueError):
apply_edits(b"abc", edits, max_output_bytes=16)
def test_intermediate_acceptance_does_not_claim_completion():
result = run_iterative_patch("two edits", b"aa",
lambda *_: PatchDecision((ByteEdit(0, b"a", b"b"),)),
validator=lambda *_: ValidationDecision(True, complete=False), max_steps=1)
assert result.final == b"ba" and not result.complete
def test_validator_completion_stops_before_another_proposal():
result = run_iterative_patch("edit", b"a", lambda *_: PatchDecision((ByteEdit(0, b"a", b"b"),)),
validator=lambda *_: ValidationDecision(True, complete=True, score=1.0))
assert result.complete and result.stop_reason == "validator_complete"
assert len(result.history) == 1 and result.final == b"b"
def test_budget_and_cycle_reject_without_consuming_accepted_state():
decisions = [PatchDecision((ByteEdit(0, b"a", b"b"),)),
PatchDecision((ByteEdit(0, b"b", b"a"),)),
PatchDecision((ByteEdit(1, b"a", b"b"),))]
result = run_iterative_patch("edit", b"aa", lambda r, c, s, h: decisions[s],
max_steps=3, max_total_changed_bytes=1)
assert result.final == b"ba" and result.accepted_steps == 1
assert all(t.after_sha256 == result.history[0].after_sha256 for t in result.history)
assert replay_audit(b"aa", result.audit()) == b"ba"
cycle = run_iterative_patch("edit", b"aa", lambda r, c, s, h: decisions[s], max_steps=2)
assert "revisited" in cycle.history[1].error and cycle.final == b"ba"
def test_child_inherits_accepted_bytes_history_and_audit_identity():
parent = run_iterative_patch("edit", b"aa", lambda *_: PatchDecision((ByteEdit(0, b"a", b"b"),)), max_steps=1)
def child_proposer(request, current, step, history):
assert step == 1 and len(history) == 1 and history[0].accepted
assert current == b"ba"
return PatchDecision((ByteEdit(1, b"a", b"b"),))
child = run_iterative_patch("edit", parent.final, child_proposer, max_steps=1, parent=parent)
assert child.final == b"bb" and child.parent_episode_sha256 == parent.audit_sha256
assert replay_audit(parent.final, child.audit()) == child.final
with pytest.raises(ValueError, match="exact accepted parent"):
run_iterative_patch("edit", b"aa", child_proposer, parent=parent)
modified = copy.deepcopy(child.audit())
modified["transitions"][0]["decision"]["edits"][0]["replacement_hex"] = b"x".hex()
with pytest.raises(ValueError, match="candidate hash"):
replay_audit(parent.final, modified)
def test_invalid_metadata_and_validator_errors_do_not_change_bytes():
bad = run_iterative_patch("edit", b"a", lambda *_: PatchDecision(stop=True, confidence=float("nan")))
assert bad.stop_reason == "predictor_error" and bad.final == b"a"
bad_span = run_iterative_patch("edit", b"a", lambda *_: PatchDecision((ByteEdit(0, "a", b"b"),)))
assert bad_span.stop_reason == "predictor_error" and bad_span.audit()["final_bytes"] == 1
proposal = lambda *_: PatchDecision((ByteEdit(0, b"a", b"b"),))
for validator in (lambda *_: ValidationDecision(False, complete=True),
lambda *_: ValidationDecision(True, score=float("nan")),
lambda *_: (_ for _ in ()).throw(RuntimeError("compiler unavailable"))):
result = run_iterative_patch("edit", b"a", proposal, validator=validator, max_steps=1)
assert result.final == b"a" and result.history[0].status == "rejected_error"
def test_output_size_bound_is_checked_before_constructing_candidate():
result = run_iterative_patch("grow", b"a", lambda *_: PatchDecision((ByteEdit(1, b"", b"123"),)),
max_steps=1, max_output_bytes=2)
assert result.final == b"a" and result.history[0].candidate_sha256 is None
assert replay_audit(b"a", result.audit()) == b"a"
|