| """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" |
|
|
