| """U10-1: resuming a Tracker after a torn tail must not glue records.""" |
| import pytest |
|
|
| from align.utils import tracking |
| from align.utils.tracking import Tracker, read_history |
|
|
|
|
| def _torn_then_resume(tmp_path, tail): |
| t = Tracker(str(tmp_path), "sft") |
| t.log(1, loss=0.5) |
| t.log(2, loss=0.6) |
| t.finish() |
| with open(t.history_path, "a") as fh: |
| fh.write(tail) |
| t2 = Tracker(str(tmp_path), "sft") |
| t2.log(3, loss=0.8) |
| t2.log(4, loss=0.9) |
| t2.finish() |
|
|
|
|
| def test_resume_after_torn_tail_keeps_history_readable(tmp_path): |
| _torn_then_resume(tmp_path, '{"step": 3, "stage": "sft", "lo') |
| assert [r["step"] for r in read_history(str(tmp_path))] == [1, 2, 3, 4] |
|
|
|
|
| def test_resume_after_complete_record_missing_newline_keeps_it(tmp_path): |
| _torn_then_resume(tmp_path, '{"step": 9, "stage": "sft", "loss": 0.1}') |
| assert [r["step"] for r in read_history(str(tmp_path))] == [1, 2, 9, 3, 4] |
|
|
|
|
| def test_resume_when_whole_file_is_one_torn_line(tmp_path): |
| (tmp_path / "history.jsonl").write_text('{"step": 1, "lo') |
| t = Tracker(str(tmp_path), "sft") |
| t.log(2, loss=0.1) |
| t.finish() |
| assert [r["step"] for r in read_history(str(tmp_path))] == [2] |
|
|
