Focused tests of v2.2.16, cut from the two test files inside marianos-scraper-2.2.16.zip. The whole files are in the
package; their SHA-256 values are in 10-SOURCE-MANIFEST-2.2.16.txt. Helpers used here (session, route, fixture_products,
Probe, CFG, SITE, COUPON, FakeApi, LocalStorage) are defined at the top of each file.

===== scraper/tests/test_oxylabs_probe.py, lines 831-1047 (unchanged text) =====
# --- v2.2.16: durable shared ledger in Blob (lease + If-Match + read-back), reconciliation ----------------------

from marianos_scraper import allowance as allowance_mod  # noqa: E402
from marianos_scraper.allowance import BlobStore, FileStore, LedgerError  # noqa: E402

PRIOR = {"ceiling": 15, "executions": [
    {"folder": "probe/oxylabs-probe-20261027T030743Z/", "oxylabs_jobs_submitted": 1},
    {"folder": "probe/oxylabs-probe-20261030T033021Z/", "package": "v2.2.14", "oxylabs_jobs_submitted": 3}],
    "total_jobs_submitted": 4}


class HttpErr(Exception):
    def __init__(self, status):
        super().__init__(f"HTTP {status}")
        self.status_code = status


class FakeBlob:
    """The calls BlobStore makes on azure.storage.blob.BlobClient, with Azure's lease and If-Match rules."""
    url = "https://acct.blob.core.windows.net/scraper-test/probe/cumulative-usage.json"

    def __init__(self, data=None):
        self.body = None if data is None else allowance_mod.encode(data)
        self.etag = 0
        self.lease = None
        self.uploads = 0
        self.store_differently = None       # callable(body) -> bytes: storage keeps other bytes than were sent
        self.read_back_differently = None   # callable(body) -> bytes: an unleased read returns other bytes

    class _Lease:
        def __init__(self, blob, lid):
            self.blob, self.id = blob, lid

        def release(self):
            if self.blob.lease == self.id:
                self.blob.lease = None

    class _Download:
        def __init__(self, body, etag):
            self._b = body
            self.properties = type("P", (), {"etag": f'"0x{etag}"'})()

        def readall(self):
            return self._b

    def acquire_lease(self, lease_duration):
        if self.body is None:
            raise HttpErr(404)
        if self.lease is not None:
            raise HttpErr(409)
        self.lease = f"lease-{self.etag}-{self.uploads}"
        return self._Lease(self, self.lease)

    def download_blob(self, lease=None):
        if self.body is None:
            raise HttpErr(404)
        body = self.body
        if lease is None and self.read_back_differently:
            body = self.read_back_differently(body)
        return self._Download(body, self.etag)

    def upload_blob(self, body, *, overwrite, lease, etag, match_condition):
        if self.lease is not None and (lease is None or lease.id != self.lease):
            raise HttpErr(412)                     # LeaseIdMissing / LeaseIdMismatch
        if etag != f'"0x{self.etag}"':
            raise HttpErr(412)                     # ConditionNotMet
        self.body = self.store_differently(body) if self.store_differently else body
        self.etag += 1
        self.uploads += 1
        return {"etag": f'"0x{self.etag}"'}

    def data(self):
        return json.loads(self.body.decode("utf-8"))


def blob_store(blob, **kw):
    return BlobStore(blob, lock_wait=kw.pop("lock_wait", 4), sleep=lambda s: None, **kw)


def run_ledger_probe(tmp_path, name, ledger, products, max_requests=14):
    s, http = session(lambda p: route(p, products=products), max_requests=max_requests)
    code = Probe(CFG, CFG.location("marianos-vernon-hills"), s, [products[0]["id"]], tmp_path / name,
                 transport="oxylabs", ledger=ledger, ledger_ceiling=15).run()
    return code, s, http


def test_blob_ledger_reserves_before_dispatch_and_records_the_final_count(tmp_path):
    blob = FakeBlob(PRIOR)
    products = fixture_products()
    s, http = session(lambda p: route(p, products=products), max_requests=14)
    posts_at_upload = []
    orig = blob.upload_blob

    def watch(body, **kw):
        posts_at_upload.append(len(http.posts))
        return orig(body, **kw)
    blob.upload_blob = watch
    code = Probe(CFG, CFG.location("marianos-vernon-hills"), s, [products[0]["id"]], tmp_path / "e1",
                 transport="oxylabs", ledger=blob_store(blob)).run()
    assert code in (0, 1) and s.max_requests == 11            # 15 - 4 counted, though 14 was asked
    assert posts_at_upload == [0, len(http.posts)]             # reservation before any POST, final after the last
    led = blob.data()
    e = led["executions"][-1]
    assert e["status"] == "final" and e["reserved_jobs"] == 11 and e["oxylabs_jobs_submitted"] == len(http.posts)
    assert led["executions"][:2] == PRIOR["executions"]        # earlier executions untouched
    assert led["total_jobs_counted"] == 4 + len(http.posts) and blob.lease is None
    report = json.loads((tmp_path / "e1" / "report.json").read_text())
    assert report["allowance"]["final_update"] == "verified" and report["allowance"]["reservation_id"] == e["reservation_id"]
    assert (tmp_path / "e1" / "cumulative-usage.json").read_bytes() == blob.body   # evidence copy = the verified blob


def test_interruption_after_dispatch_before_final_update_is_counted_in_full_by_a_fresh_session(tmp_path, monkeypatch):
    """A run reserves 5, dispatches jobs, and dies before its final ledger update and before the evidence upload.
    A fresh session (new process, folder and Oxylabs session) counts the full reservation and resets nothing."""
    blob = FakeBlob(PRIOR)
    products = fixture_products()
    s, http = session(lambda p: route(p, products=products), max_requests=5)
    p = Probe(CFG, CFG.location("marianos-vernon-hills"), s, [products[0]["id"]], tmp_path / "killed",
              transport="oxylabs", ledger=blob_store(blob))
    monkeypatch.setattr(p, "_finish", lambda code, lock_timeout=None: None)   # the process is killed here
    p.run()
    assert len(http.posts) >= 2 and not (tmp_path / "killed" / "report.json").exists()
    killed = blob.data()["executions"][-1]
    assert killed["status"] == "reserved" and killed["reserved_jobs"] == 5 and killed["oxylabs_jobs_submitted"] is None
    # the fresh session
    code2, s2, http2 = run_ledger_probe(tmp_path, "fresh", blob_store(blob), products)
    led = blob.data()
    assert led["executions"][2]["status"] == "unfinished"     # reconciled, kept at its full reservation
    assert led["executions"][2]["oxylabs_jobs_submitted"] is None and led["executions"][2]["reserved_jobs"] == 5
    assert s2.max_requests == 15 - 4 - 5                       # 6 left: the jobs it cannot see are not guessed
    report = json.loads((tmp_path / "fresh" / "report.json").read_text())
    assert report["allowance"]["reconciled_unfinished"] == ["killed"] and report["allowance"]["counted_before"] == 9
    assert led["total_jobs_counted"] == 9 + len(http2.posts)
    # when nothing is left, a further session reconciles, records that, and sends nothing
    blob2 = FakeBlob({"ceiling": 15, "executions": PRIOR["executions"] + [
        {"folder": "k", "reservation_id": "r", "status": "reserved", "reserved_jobs": 11, "oxylabs_jobs_submitted": None}]})
    code3, _, http3 = run_ledger_probe(tmp_path, "after", blob_store(blob2), products)
    assert code3 == 2 and not http3.posts and blob2.data()["executions"][-1]["status"] == "unfinished"


def test_a_held_lease_stops_the_probe_before_anything_is_sent(tmp_path):
    blob = FakeBlob(PRIOR)
    blob.lease = "held-by-another-execution"
    code, s, http = run_ledger_probe(tmp_path, "e", blob_store(blob), fixture_products())
    assert code == 2 and not http.posts and blob.uploads == 0
    report = json.loads((tmp_path / "e" / "report.json").read_text())
    assert "409" in report["stop_reason"] and "nothing sent" in report["stop_reason"]


def test_a_stale_copy_cannot_overwrite_a_newer_ledger():
    blob = FakeBlob(PRIOR)
    store = blob_store(blob)
    store.lock()
    body, etag = store.read()
    blob.etag += 1                                     # changed after our read
    with pytest.raises(LedgerError, match="ETag"):
        store.write(b"{}", etag)
    store.unlock()
    assert blob.data() == PRIOR


def test_a_write_without_the_lease_is_refused():
    blob = FakeBlob(PRIOR)
    a = blob_store(blob)
    a.lock()
    b = blob_store(blob)                               # never took the lease
    with pytest.raises(LedgerError):
        b.write(b"{}", f'"0x{blob.etag}"')
    a.unlock()
    assert blob.data() == PRIOR


def test_read_back_mismatch_on_reservation_stops_before_dispatch(tmp_path):
    blob = FakeBlob(PRIOR)
    blob.read_back_differently = lambda b: b[:-2]
    code, s, http = run_ledger_probe(tmp_path, "e", blob_store(blob), fixture_products())
    assert code == 2 and not http.posts
    report = json.loads((tmp_path / "e" / "report.json").read_text())
    assert "LedgerMismatch" in report["stop_reason"] and "unresolved" in report["stop_reason"]


def test_read_back_mismatch_on_final_update_stops_and_stays_unresolved(tmp_path):
    blob = FakeBlob(PRIOR)
    products = fixture_products()
    s, http = session(lambda p: route(p, products=products), max_requests=14)
    p = Probe(CFG, CFG.location("marianos-vernon-hills"), s, [products[0]["id"]], tmp_path / "e",
              transport="oxylabs", ledger=blob_store(blob))
    orig = p.allowance.finalize

    def corrupt_then_finalize(*a, **kw):
        blob.store_differently = lambda b: b.replace(b'"final"', b'"fin4l"')   # storage keeps other bytes
        return orig(*a, **kw)
    p.allowance.finalize = corrupt_then_finalize
    code = p.run()
    assert code == 2 and http.posts                            # it had dispatched; the run is still reported stopped
    report = json.loads((tmp_path / "e" / "report.json").read_text())
    assert report["exit_code"] == 2 and report["transport"] == "stopped"
    assert report["allowance"]["final_update"].startswith("UNRESOLVED: LedgerMismatch")
    assert not (tmp_path / "e" / "cumulative-usage.json").exists()   # no unverified copy shown as evidence


def test_missing_or_malformed_ledger_fails_closed(tmp_path):
    code, _, http = run_ledger_probe(tmp_path, "missing", blob_store(FakeBlob(None)), fixture_products())
    assert code == 2 and not http.posts
    assert "not found" in json.loads((tmp_path / "missing" / "report.json").read_text())["stop_reason"]
    bad = FakeBlob({"ceiling": 15, "executions": [{"folder": "x"}]})
    code, _, http = run_ledger_probe(tmp_path, "bad", blob_store(bad), fixture_products())
    assert code == 2 and not http.posts and bad.uploads == 0


def test_file_ledger_uses_an_exclusive_lock(tmp_path):
    path = tmp_path / "cumulative-usage.json"
    path.write_bytes(allowance_mod.encode(PRIOR))
    (tmp_path / "cumulative-usage.json.lock").write_text("other execution")
    code, _, http = run_ledger_probe(tmp_path, "e", FileStore(path, lock_wait=0, sleep=lambda s: None), fixture_products())
    assert code == 2 and not http.posts and json.loads(path.read_text()) == PRIOR


===== scraper/tests/test_scraper.py, lines 365-505 (unchanged text) =====
# --- v2.2.16: full-run retention across repeat runs ----------------------------------------------------------

class ChangingStore:
    """The whole website for SiteRunner.run(), offline: every taxonomy root has one leaf of five products. Between
    runs the test changes `unavailable` (listed, not sellable) and `absent` (no longer listed at all)."""
    STORE = {"locationId": "53100503", "facilityId": "15096", "brand": "MARIANOS", "storeNumber": "00503",
             "vanityName": "Mariano's Lakeshore East",
             "locale": {"address": {"name": "Lakeshore East", "addressLines": ["333 E Benton Pl"], "cityTown": "CHICAGO",
                                    "stateProvince": "IL", "postalCode": "60601"}}}

    def __init__(self):
        import types
        self.unavailable, self.absent = set(), set()
        self.calls, self.max_requests, self.max_seconds = [], None, None
        self.stats = types.SimpleNamespace(requests=0, retries=0, errors=0, total_seconds=0.0)

    @staticmethod
    def leaf(root):
        return root if len(root) >= 5 else root + "001"

    @staticmethod
    def upcs_of(leaf):
        return [f"{leaf}{i:02d}".rjust(13, "0") for i in range(1, 6)]

    def listed(self, leaf):
        return [u for u in self.upcs_of(leaf) if u not in self.absent]

    def open(self):
        pass

    def close(self):
        pass

    def store(self, zipcode, location_id):
        return self.STORE

    def set_store(self, *a, **kw):
        pass

    def check_context(self, products, path):
        pass

    def search(self, location_id, taxonomy, offset=0, size=100, brand="", facets=False, brand_facet=False):
        if facets:
            if len(taxonomy) >= 5:
                n = len(self.listed(taxonomy)) if taxonomy in self._leaves() else 0
                return [], n, [{"id": taxonomy, "name": taxonomy, "count": n, "children": []}]
            leaf = self.leaf(taxonomy)
            n = len(self.listed(leaf))
            return [], n, [{"id": taxonomy, "name": taxonomy, "count": n,
                            "children": [{"id": leaf, "name": f"Aisle {leaf}", "count": n, "children": []}]}]
        ids = self.listed(taxonomy)
        return [{"upc": u} for u in ids[offset:offset + size]], len(ids), {}

    def _leaves(self):
        import json as _j
        tax = _j.loads((Path(CFG.config_dir) / "site_taxonomy.json").read_text(encoding="utf-8"))["categories"]
        return {r for spec in tax.values() for r in spec.get("roots", []) if len(r) >= 5}

    def products(self, upcs, verified=True):
        import copy
        out = []
        for u in upcs:
            p = copy.deepcopy(SITE["regular"])
            p["id"] = u
            p["item"] = {**p.get("item", {}), "upc": u, "description": f"Roundy's Product {u}"}
            p["price"]["storePrices"]["sourceLocationId"] = "53100503"
            if u in self.unavailable:
                for s in p.get("fulfillmentSummaries") or []:
                    s.setdefault("availability", {})["sellable"] = False
            out.append(p)
        return out

    def all_coupons(self, kind):
        return ([COUPON] if kind == "standard" else []), {}

    def coupons_for(self, upcs):
        return []


def test_unavailable_and_later_absent_products_survive_repeat_runs_with_their_real_last_observation(tmp_path, monkeypatch):
    """Three full runs of SiteRunner.run() on one store (LocalStorage, no network):
    run 1 (Oct 01) everything listed; run 2 (Oct 02) GONE is no longer listed and OOS is listed but not sellable;
    run 3 (Oct 03) GONE is still absent and OOS is absent too.
    Every product stays in every published output; a carried record keeps the crawl_date of its last observation
    (never today's), availability.jsonl tells fresh from carried, and nothing is invented."""
    from datetime import datetime as _dt
    from marianos_scraper import site_runner as sr
    from marianos_scraper.site_runner import SiteRunner
    for k, val in {"min_records_per_category": 1, "min_ppm_share": 0.0, "min_total_records": 10}.items():
        monkeypatch.setitem(CFG.settings["validation"], k, val)
    CFG.storage_root = str(tmp_path)
    storage = LocalStorage(str(tmp_path))
    site = ChangingStore()
    leaf = site.leaf("02")                                    # Dairy
    GONE, OOS = site.upcs_of(leaf)[0], site.upcs_of(leaf)[1]

    day = {"v": "2026-10-01"}

    class FixedDay(_dt):
        @classmethod
        def now(cls, tz=None):
            return _dt.fromisoformat(day["v"] + "T12:00:00+00:00")
    monkeypatch.setattr(sr, "datetime", FixedDay)

    def run(d):
        day["v"] = d
        runner = SiteRunner(CFG, CFG.location("marianos-lakeshore-east"), site=site, api=FakeApi(), storage=storage)
        report = runner.run()
        assert report["status"] == "published", [c for c in report["validation"]["checks"] if not c["passed"]]
        out = f"runs/marianos-lakeshore-east/{report['run_id']}/"
        products = {r["upc"]: r for r in json.loads(storage.read_text(out + "products.json"))}
        avail = {a["upc"]: a for a in map(json.loads, storage.read_text(out + "availability.jsonl").splitlines())}
        latest = {r["upc"] for r in json.loads(storage.read_text("latest/marianos-lakeshore-east/products.json"))}
        assert set(products) == latest
        return report["run_id"], products, avail

    run1, p1, a1 = run("2026-10-01")
    assert {GONE, OOS} <= set(p1) and a1[GONE]["observation"] == a1[OOS]["observation"] == "fresh"
    total = len(p1)

    site.absent, site.unavailable = {GONE}, {OOS}
    run2, p2, a2 = run("2026-10-02")
    assert len(p2) == total                                         # nothing dropped
    assert p2[GONE]["crawl_date"] == "2026-10-01"                   # carried: its real last observation
    assert a2[GONE] == {**a2[GONE], "observation": "carried_forward", "last_observed_run": run1, "crawl_date": "2026-10-01"}
    assert p2[OOS]["crawl_date"] == "2026-10-02" and a2[OOS]["observation"] == "fresh"   # listed: a fresh record
    assert a2[OOS]["fulfillment_sellable"] == [] and a2[OOS]["last_observed_run"] == run2

    site.absent = {GONE, OOS}
    run3, p3, a3 = run("2026-10-03")
    assert len(p3) == total
    assert p3[GONE]["crawl_date"] == "2026-10-01" and a3[GONE]["last_observed_run"] == run1   # not reset by run 2
    assert p3[OOS]["crawl_date"] == "2026-10-02" and a3[OOS]["last_observed_run"] == run2
    assert a3[GONE]["observation"] == a3[OOS]["observation"] == "carried_forward"
    fresh = [u for u, a in a3.items() if a["observation"] == "fresh"]
    assert len(fresh) == total - 2 and all(p3[u]["crawl_date"] == "2026-10-03" for u in fresh)
    assert p3[GONE] == p1[GONE] and p3[OOS] == p2[OOS]              # the record itself is the last observation, unchanged
    changes = json.loads(storage.read_text(f"runs/marianos-lakeshore-east/{run3}/availability_changes.json"))
    assert {c["upc"] for c in changes if c["kept_as"].startswith("carried_forward")} == {GONE, OOS}