"""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) # kill mid-append: no trailing newline 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]