stack/tests/pfs/test_runlog.py

"""pfs.runlog — append-only resumable ledger."""

from __future__ import annotations

import json

from conf import ROOT
from pfs.runlog import RunLog, input_hash, ledger_path, run_units


def _units(calls, outs=({"v": 1}, {"v": 2}, {"v": 3}), hashes=("h1", "h2", "h3")):
    def mk(i):
        def fn():
            calls.append(i)
            return outs[i]

        return fn

    return [(f"u{i}", hashes[i], mk(i)) for i in range(3)]


def test_runlog_resume_skips_logged_units(tmp_path):
    log = RunLog(tmp_path / "l.jsonl")
    calls: list[int] = []
    seen = []
    r1 = run_units(
        _units(calls),
        log,
        model="m",
        version="v1",
        progress=lambda u, s: seen.append(s),
    )
    assert (
        r1 == {"done": 3, "skipped": 0, "failed": 0, "failed_pending": 0}
        and seen == ["done"] * 3
    )
    calls.clear()
    r2 = run_units(_units(calls), log, model="m", version="v1")
    assert (
        r2 == {"done": 0, "skipped": 3, "failed": 0, "failed_pending": 0}
        and calls == []
    )


def test_runlog_new_hash_reruns(tmp_path):
    log = RunLog(tmp_path / "l.jsonl")
    run_units(_units([]), log, model="m", version="v1")
    calls: list[int] = []
    r = run_units(_units(calls, hashes=("h1", "X", "h3")), log, model="m", version="v1")
    assert r["done"] == 1 and r["skipped"] == 2 and calls == [1]
    calls.clear()
    r = run_units(_units(calls), log, model="m", version="v2")
    assert r["done"] == 3
    assert log.get("u1", "X", "m", "v1").output == {"v": 2}
    assert log.get("u1", "nope", "m", "v1") is None


def test_failed_units_retry_only_when_asked(tmp_path):
    log = RunLog(tmp_path / "l.jsonl")
    calls: list[int] = []
    outs = ({"v": 1}, None, {"v": 3})
    seen = []
    r = run_units(
        _units(calls, outs),
        log,
        model="m",
        version="v",
        progress=lambda u, s: seen.append((u, s)),
    )
    assert (
        r == {"done": 2, "skipped": 0, "failed": 1, "failed_pending": 0}
        and ("u1", "failed") in seen
    )
    calls.clear()
    r = run_units(_units(calls, outs), log, model="m", version="v")
    assert (
        r == {"done": 0, "skipped": 2, "failed": 0, "failed_pending": 1} and calls == []
    )
    r = run_units(
        _units(calls, ({"v": 1}, {"v": 2}, {"v": 3})),
        log,
        model="m",
        version="v",
        retry_failed=True,
    )
    assert r == {
        "done": 1,
        "skipped": 2,
        "failed": 0,
        "failed_pending": 0,
    } and calls == [1]
    assert log.get("u1", "h2", "m", "v").output == {"v": 2}
    assert log.outputs("m", "v") == {"u0": {"v": 1}, "u1": {"v": 2}, "u2": {"v": 3}}


def test_outputs_latest_non_null_and_scoped(tmp_path):
    log = RunLog(tmp_path / "l.jsonl")
    log.put("a", "h", "m", "v", {"x": 1})
    log.put("a", "h", "m", "v", None)
    log.put("b", "h", "other", "v", {"x": 2})
    log.put("c", "h", "m", "v", {"x": 3})
    log.put("c", "h", "m", "v", {"x": 4})
    assert log.outputs("m", "v") == {"a": {"x": 1}, "c": {"x": 4}}


def test_limit_caps_new_calls(tmp_path):
    log = RunLog(tmp_path / "l.jsonl")
    calls: list[int] = []
    r = run_units(_units(calls), log, model="m", version="v", limit=2)
    assert r["done"] == 2 and calls == [0, 1]
    calls.clear()
    r = run_units(_units(calls), log, model="m", version="v", limit=2)
    assert r == {
        "done": 1,
        "skipped": 2,
        "failed": 0,
        "failed_pending": 0,
    } and calls == [2]


def test_put_appends_json_lines_and_blank_lines_ignored(tmp_path):
    p = tmp_path / "sub" / "l.jsonl"
    log = RunLog(p)
    assert log.outputs("m", "v") == {}
    e = log.put("a", "h", "m", "v", {"k": 1})
    with p.open("a") as f:
        f.write("\n")
    log.put("b", "h", "m", "v", None)
    lines = p.read_text().splitlines()
    assert json.loads(lines[0])["unit"] == "a" and e.ts
    assert log.get("a", "h", "m", "v") == e


def test_input_hash_stable_and_part_sensitive():
    assert input_hash("a", "b") == input_hash("a", "b")
    assert len(input_hash("a")) == 16
    assert input_hash("ab", "c") != input_hash("a", "bc")


def test_ledger_path_under_data():
    assert ledger_path("origin", "HOME-VISITS") == (
        ROOT / "data/pfs/ledger" / "origin-HOME-VISITS.jsonl"
    )


def test_truncated_and_malformed_lines_are_skipped_and_put_recovers(tmp_path, caplog):
    p = tmp_path / "l.jsonl"
    RunLog(p).put("a", "h", "m", "v", {"k": 1})
    with p.open("a") as f:
        f.write('\n{"unit": "x", "missing": 1}\n')  # blank + valid JSON, wrong keys
        f.write('{"unit": "b", "input_ha')  # cut mid-write, no newline
    log = RunLog(p)
    with caplog.at_level("WARNING"):
        assert log.outputs("m", "v") == {"a": {"k": 1}}
    assert caplog.text.count("skipping unreadable") == 2
    log.put("c", "h", "m", "v", {"k": 3})
    fresh = RunLog(p)
    assert fresh.outputs("m", "v") == {"a": {"k": 1}, "c": {"k": 3}}
    assert fresh.get("c", "h", "m", "v").output == {"k": 3}