"""Small, capped, reviewable test of one transport against one store: store check -> a handful of products -> their
coupons -> the same mapping the full run uses. Everything it sends and receives is written to one evidence folder.

  python -m marianos_scraper probe --location marianos-vernon-hills --transport oxylabs \\
      --upcs 0003940001880,0001600040803,0073820310311,0004157305440,0002100006058 \\
      --roles ppm-volume,ppm-digital,regular,unavailable,unavailable \\
      --max-requests 15 --max-seconds 1200 --out evidence/run-20261026T1500Z

Stop conditions, all enforced in this process (nothing depends on the operator watching):
  * request cap        : every dispatched attempt counts, checked again immediately before dispatch (retries,
                         timeouts, lost pages included)
  * runtime cap        : a monotonic deadline; no wait or request runs past it
  * overall stop       : a watchdog armed from start to the end of evidence writing (cleanup included). At deadline
                         + 30 s it gives the evidence writer at most 20 s in its own thread, then ends the process
                         whatever that writer does (latest end: deadline + 50 s). infra/run_probe.ps1 adds an
                         independent outer kill of the whole process tree (default deadline + 120 s)
  * first 401/403      : SiteUnavailable -> stop, exit 4 (same rule as the full run)
  * transport failures : N consecutive Oxylabs-side failures -> stop, exit 2 (OxylabsSession)
  * store mismatch     : locator answer not the configured store/facility -> stop before any product request, exit 2
  * context lost       : a priced product whose sourceLocationId is not the configured store, or any product whose
                         echoed fulfilment context (laf: modality type, storeId, facilityId) is not the one sent
                         -> stop before the coupon request, exit 2

Cumulative allowance (--ledger blob:<url> or FILE, see allowance.py): before any request the probe takes the ledger's
exclusive lease, reconciles any unfinished reservation (kept at its full count), reserves what is left of the ceiling
(default 15) as this run's cap, writes it with If-Match on the ETag and reads it back. Only a verified reservation lets
the probe send anything. At the end (also after a stop or the watchdog) it records the jobs actually dispatched, on a
fresh read matched by reservation id, verified the same way; if that cannot be verified the run ends with exit 2 and
the reservation stays counted. A new folder, session, process or machine can never reset the total.

Session replacement (Oxylabs): a failed session is never reused. When the transport replaced its session after the
store was verified, the store is verified again on the new session (counted) and the products are asked again; only
an answer obtained on a session whose store check passed can qualify.

Transport completion and acceptance are separate (report.json "transport" and "acceptance"):
  exit 0 = transport completed AND every acceptance check passed
  exit 1 = transport completed, but at least one acceptance check failed or is inconclusive (read report.json)
  exit 2 / 4 = transport stopped (see above); acceptance is then not evaluated
Acceptance checks per UPC role (--roles): store/fulfilment context of the source answer, products returned, PICKUP
sellable for products expected in stock, displayed PPM copied exactly, volume offer, digital coupon, unavailable
product retained. A criterion the site does not serve today (stock or offer changed) is "inconclusive", never "pass".

Evidence folder:
  report.json            what ran, when (UTC and US Central), caps, requests used, transport result, acceptance checks
  requests.jsonl         one line per request: time, path, HTTP status, bytes, duration, job id (Oxylabs)
  oxylabs_jobs.jsonl     Oxylabs only: the exact payload sent (headers included, no credentials) and the job result
  raw/NN-<step>.json|txt every attempt's response body exactly as returned (failures, every coupon page included)
  raw/NN-<step>.oxylabs.json  Oxylabs only: the exact Realtime answer (envelope) of that attempt
  raw/index.jsonl        every attempt: seq, path, status, error, files and their SHA-256
  trace.jsonl            per UPC: raw file + index -> sourceLocationId, fulfilment, price status, coupons -> record
  products.csv           the 22-field records, same mapper and field order as a full run (unavailable products kept)
  summary.csv            per UPC: role, returned?, store id, echoed context, fulfilment, prices, PPM, deal terms
  manifest.sha256        SHA-256 of every file above
"""
from __future__ import annotations

import csv
import hashlib
import json
import os
import platform
import re
import socket
import threading
import time
from datetime import datetime, timedelta, timezone
from pathlib import Path
from urllib.request import Request, urlopen
from zoneinfo import ZoneInfo, ZoneInfoNotFoundError

from . import __version__
from .kroger_api import SourceBlocked, SourceError
from .log import log
from .mapper import FIELD_ORDER, POINTS_OFFER, SiteMapper, fix_text, usd
from .runner import code_fingerprint
from .allowance import Allowance, LedgerError, encode, jobs_counted, open_store
from .site_source import RequestCapReached, SiteSession, SiteUnavailable, echoed_context

try:
    CENTRAL = ZoneInfo("America/Chicago")
except ZoneInfoNotFoundError:   # Windows Python without the tzdata package (e.g. the Azure VM): use the US rule below
    CENTRAL = None

ROLES = ("ppm-volume", "ppm-digital", "regular", "unavailable", "any")
WATCHDOG_GRACE_SECONDS = 30          # after the request deadline, before the watchdog acts
WATCHDOG_WRITE_SECONDS = 20          # the most the watchdog waits for the evidence writer before ending the process


def _central(dt: datetime) -> datetime:
    """US Central time. Without a tz database: CDT from the 2nd Sunday of March 02:00 CST to the 1st Sunday of
    November 02:00 CDT, CST otherwise."""
    if CENTRAL is not None:
        return dt.astimezone(CENTRAL)
    start = datetime(dt.year, 3, 8, 8, tzinfo=timezone.utc)
    start += timedelta(days=(6 - start.weekday()) % 7)
    end = datetime(dt.year, 11, 1, 7, tzinfo=timezone.utc)
    end += timedelta(days=(6 - end.weekday()) % 7)
    if start <= dt < end:
        return dt.astimezone(timezone(timedelta(hours=-5), "CDT"))
    return dt.astimezone(timezone(timedelta(hours=-6), "CST"))

SUMMARY_FIELDS = ["upc", "role", "returned", "title", "size", "sourceLocationId", "store_match", "echoed_modality",
                  "echoed_store_id", "echoed_facility_id", "price_status",
                  "fulfillment_listed", "fulfillment_sellable", "stock_level", "regular_price", "promo_price", "ppm",
                  "sale_price", "volume_deals_description", "volume_deals_price", "digital_deals_description",
                  "digital_deals_price", "coupons_returned", "kept_in_output", "exclusion_reason", "raw_ref", "coupon_refs"]


def _stamp(ts: float | None = None) -> dict:
    dt = datetime.fromtimestamp(ts or time.time(), tz=timezone.utc)
    return {"utc": dt.strftime("%Y-%m-%dT%H:%M:%SZ"), "central": _central(dt).strftime("%Y-%m-%d %H:%M:%S %Z")}


def _kind(path: str) -> str:
    """Evidence file name part for one request path."""
    if "/stores/" in path:
        return "store-locator"
    if "/product/v2/products" in path:
        return "products" if "filter.verified=true" in path else "products-unverified"
    if "/savings-coupons/" in path:
        m = re.search(r"page\.offset=(\d+)", path)
        return "coupons" + (f"-offset{m.group(1)}" if m else "")
    return "request"


def _body_file(att: dict) -> str:
    body = att.get("body")
    if body is None:
        return ""
    ext = "json" if str(body).lstrip()[:1] in ("{", "[") else "txt"
    return f"raw/{att['seq']:02d}-{_kind(att['path'])}.{ext}"


def _azure_instance() -> dict:
    """Azure Instance Metadata Service (link-local, no site request): which VM ran the probe, and its public IP."""
    url = "http://169.254.169.254/metadata/instance?api-version=2021-02-01"
    try:
        with urlopen(Request(url, headers={"Metadata": "true"}), timeout=3) as r:
            data = json.loads(r.read().decode("utf-8"))
    except Exception as exc:
        return {"available": False, "error": f"{type(exc).__name__}: {str(exc)[:120]}"}
    c = data.get("compute") or {}
    ips = [a.get("publicIpAddress") for i in (data.get("network") or {}).get("interface") or []
           for a in ((i.get("ipv4") or {}).get("ipAddress") or []) if a.get("publicIpAddress")]
    return {"available": True, "vmId": c.get("vmId"), "name": c.get("name"), "location": c.get("location"),
            "resourceGroupName": c.get("resourceGroupName"), "vmSize": c.get("vmSize"), "osType": c.get("osType"),
            "publicIpAddresses": ips}


_ledger_used = jobs_counted


def _is_clip_coupon(c: dict) -> bool:
    """Same filter as SiteMapper.digital_text: a standard coupon that can be added to the card, not a points offer."""
    text = fix_text(c.get("shortDescription") or c.get("displayDescription") or "")
    return (str(c.get("type", "")).upper() == "STANDARD" and c.get("canBeAddedToCard", True)
            and not POINTS_OFFER.search(text))


def _coupon_terms(c: dict) -> list[tuple[str, str]]:
    """Every term of one clip coupon as the source states it, each of which must appear in the record."""
    terms = [("clipping", "Digital Coupon (clip to card)")]
    text = fix_text(c.get("shortDescription") or c.get("displayDescription") or "")
    if text:
        terms.append(("title", text))
    req = fix_text(c.get("requirementDescription") or "")
    if req:
        terms.append(("requirement (quantity, eligibility, limits, validity)", req))
    exp = (c.get("expirationDate") or "")[:10]
    if exp:
        terms.append(("expiry", f"Expires {exp}."))
    return terms


def _volume_terms(p: dict) -> dict:
    """Every volume-offer term the source states (wording, validity, order types, multi-buy and its price)."""
    terms: dict = {}
    sp = (p.get("price") or {}).get("storePrices") or {}
    promo = sp.get("promo") or {}
    nfor = int(promo.get("nfor") or 1)
    if usd(promo.get("price")) and nfor > 1:
        terms[f"{nfor} for ${usd(promo.get('nforPrice'))}"] = True
        ends = str((promo.get("expirationDate") or {}).get("value") or "")[:10]
        if ends and not ends.startswith("9999"):
            terms[f"Ends {ends}."] = True
        terms["__price__"] = usd(promo.get("nforPrice"))
    for o in p.get("offers") or []:
        text = fix_text(o.get("defaultDescription", ""))
        if not text:
            continue
        terms[text] = True
        start, end = str(o.get("start") or "")[:10], str(o.get("end") or "")[:10]
        if start and end:
            terms[f"Valid {start} to {end}"] = True
        elif end:
            terms[f"Ends {end}"] = True
        modes = [str(m).replace("_", " ").lower() for m in o.get("modalities") or []]
        if modes:
            terms["Order types: " + ", ".join(modes)] = True
    return terms


def _unavailable_proof(p: dict) -> str:
    """The source's own statement that the product is unavailable at the store, or '' when it states nothing."""
    pickup = [s for s in p.get("fulfillmentSummaries") or [] if s.get("type") == "PICKUP"]
    # only an explicit false counts: a missing or null "sellable" says nothing about availability
    if pickup and all((s.get("availability") or {}).get("sellable") is False for s in pickup):
        return "PICKUP fulfilment summary with sellable = false"
    levels = [l.get("stockLevel") for l in ((p.get("inventory") or {}).get("locations") or []) if l.get("stockLevel")]
    levels += [s.get("stockLevel") for s in p.get("inventorySummaries") or [] if s.get("stockLevel")]
    if "TEMPORARILY_OUT_OF_STOCK" in levels:
        return "stock level TEMPORARILY_OUT_OF_STOCK"
    return ""


class Probe:
    def __init__(self, cfg, location, site: SiteSession, upcs: list[str], out: Path, *, transport: str,
                 egress_ip: str = "", roles: list[str] | None = None, azure_imds: bool = False,
                 ledger=None, ledger_ceiling: int = 15):
        self.cfg, self.loc, self.site, self.upcs = cfg, location, site, [u.strip() for u in upcs if u.strip()]
        roles = [r.strip() for r in (roles or []) if r.strip()]
        if roles and len(roles) != len(self.upcs):
            raise ValueError(f"--roles has {len(roles)} entries for {len(self.upcs)} UPCs")
        bad = [r for r in roles if r not in ROLES]
        if bad:
            raise ValueError(f"unknown roles {bad}; use {', '.join(ROLES)}")
        self.roles = dict(zip(self.upcs, roles)) if roles else {u: "any" for u in self.upcs}
        self.out = out
        self.transport = transport
        self.egress_ip = egress_ip
        self.azure_imds = azure_imds
        if ledger is None or ledger == "":
            self.allowance = None
        else:
            store = open_store(str(ledger)) if isinstance(ledger, (str, Path)) else ledger
            self.allowance = Allowance(store, ceiling=ledger_ceiling)
        self._store_generation = -1      # session generation on which the store was last verified
        self.mapper = SiteMapper(cfg.settings)
        self.site.capture = True          # keep every attempt with its exact answer for raw/
        self.products: list[dict] = []
        self.coupons: list[dict] = []
        self.origin: dict[str, str] = {}        # product id -> raw file + index it came from
        self.coupon_origin: dict[str, str] = {}  # coupon id -> raw file + index
        self._lock = threading.Lock()
        self._finished = False
        self._exit_override: int | None = None   # set when the ledger's final update cannot be verified
        self.report: dict = {"version": __version__, "code_fingerprint": code_fingerprint(), "transport_name": transport,
                             "location": location.key, "location_id": location.location_id,
                             "facility_id_expected": location.facility_id, "zipcode": location.zipcode,
                             "upcs": self.upcs, "roles": self.roles,
                             "caps": {"max_requests": site.max_requests, "max_seconds": site.max_seconds},
                             "host": socket.gethostname(), "platform": platform.platform(), "egress_ip": egress_ip,
                             "started": _stamp(site.started), "steps": [], "transport": "", "stop_reason": "",
                             "acceptance": {"overall": "not evaluated", "checks": []}, "exit_code": None,
                             "exit_codes": {"0": "transport completed and every acceptance check passed",
                                            "1": "transport completed; at least one acceptance check failed or is inconclusive",
                                            "2": "stopped: cap, Oxylabs error, store or fulfilment mismatch",
                                            "4": "stopped: the site refused (first 401/403 or challenge)"}}

    # --- files --------------------------------------------------------------------------------------------
    def _write(self, name: str, text: str) -> None:
        path = self.out / name
        path.parent.mkdir(parents=True, exist_ok=True)
        with open(path, "w", encoding="utf-8", newline="") as fh:   # exact text: no newline translation
            fh.write(text)

    def _step(self, name: str, **fields) -> None:
        self.report["steps"].append({"step": name, **_stamp(), "requests_so_far": self.site.stats.requests, **fields})
        log("info", "probe_step", step=name, **fields)

    def _last_ok(self, since: int) -> list[dict]:
        """Successful attempts made since attempt index `since`."""
        return [a for a in self.site.attempts[since:] if a.get("status") == 200 and a.get("body") is not None]

    # --- the probe -------------------------------------------------------------------------------------------
    def run(self) -> int:
        self.out.mkdir(parents=True, exist_ok=True)
        watchdog = None
        if self.site.max_seconds is not None:
            # the overall stop: if anything still runs 30 s after the deadline, write the evidence and end the process
            left = max(0.0, self.site.time_left() or 0.0)
            watchdog = threading.Timer(left + WATCHDOG_GRACE_SECONDS, self._watchdog)
            watchdog.daemon = True
            watchdog.start()
        if self.azure_imds:
            self.report["azure_instance"] = _azure_instance()
        code = 2
        if self.allowance and not self._reserve_allowance():
            self.report["transport"] = "stopped"
            self._finish(2)
            if watchdog:
                watchdog.cancel()
            return 2
        try:
            self.site.open()
            location = self._check_store()
            self._collect()
            self._map(location)
            self.report["transport"] = "completed"
            self.report["stop_reason"] = "completed"
            code = 0 if self.report["acceptance"]["overall"] == "pass" else 1
        except SiteUnavailable as exc:
            self.report["stop_reason"] = f"blocked: {exc}"
            code = 4
        except RequestCapReached as exc:
            self.report["stop_reason"] = f"cap: {exc}"
            code = 2
        except SourceBlocked as exc:
            self.report["stop_reason"] = f"blocked: {exc}"
            code = 4
        except SourceError as exc:
            self.report["stop_reason"] = f"error: {exc}"
            code = 2
        finally:
            # the watchdog stays armed through cleanup and evidence writing: a stall here is still ended by it
            try:
                self.site.close()
            except Exception:
                pass
            if code != 0 and code != 1:
                self.report["transport"] = "stopped"
            self._finish(code)
            if watchdog:
                watchdog.cancel()
            if self._exit_override is not None:
                code = self._exit_override
        return code

    def _reserve_allowance(self) -> bool:
        """Reserve this run's cap in the shared ledger, durably and verified, before any request."""
        try:
            info = self.allowance.reserve(self.out.name, f"v{__version__}", self.site.max_requests)
        except LedgerError as exc:
            self.report["allowance"] = {"ledger": self.allowance.store.describe(), "error": str(exc)}
            self.report["stop_reason"] = f"allowance: ledger {type(exc).__name__}: {exc}; nothing sent"
            return False
        self.report["allowance"] = info
        if info["cap_this_run"] <= 0:
            self.report["stop_reason"] = (f"allowance: {info['counted_before']} of {info['ceiling']} jobs already "
                                          f"counted; nothing sent")
            return False
        self.site.max_requests = info["cap_this_run"]
        self.report["caps"]["max_requests"] = info["cap_this_run"]
        return True

    def _on_verified_session(self, ask):
        """Run one request group so that its answer comes from a session whose store check passed. If the transport
        replaced its session in the meantime, verify the store on the new session (counted) and ask once more."""
        for _ in range(2):
            since = len(self.site.attempts)
            got = ask()
            if self.site.session_generation == self._store_generation:
                return got, since
            self._step("session_replaced", generation=self.site.session_generation,
                       note="the answer came on a session without a store check: store verified again, request repeated")
            self._check_store()
        raise SourceError("the Oxylabs session was replaced again: store context not re-established on a stable session")

    def _watchdog(self) -> None:
        """Ends the process. The evidence is written by a separate thread given WATCHDOG_WRITE_SECONDS; if the main
        thread is stuck (in cleanup, or inside the evidence writer holding its lock), the process ends anyway."""
        self.report["transport"] = "stopped"
        self.report["stop_reason"] = (f"watchdog: still running {WATCHDOG_GRACE_SECONDS}s after the {self.site.max_seconds}s "
                                      f"deadline; evidence writer given {WATCHDOG_WRITE_SECONDS}s, then the process ended")
        writer = threading.Thread(target=self._finish, args=(2,), kwargs={"lock_timeout": WATCHDOG_WRITE_SECONDS}, daemon=True)
        writer.start()
        writer.join(WATCHDOG_WRITE_SECONDS)
        os._exit(2)

    def _check_store(self) -> dict:
        since = len(self.site.attempts)
        s = self.site.store(self.loc.zipcode, self.loc.location_id)
        ok = self._last_ok(since)
        addr = (s.get("locale") or {}).get("address") or {}
        problems = []
        if str(s.get("brand", "")).upper() != self.loc.expected_chain.upper():
            problems.append(f"brand {s.get('brand')} != {self.loc.expected_chain}")
        if str(addr.get("postalCode", ""))[:5] != self.loc.zipcode:
            problems.append(f"zip {addr.get('postalCode')} != {self.loc.zipcode}")
        if self.loc.facility_id and str(s.get("facilityId")) != self.loc.facility_id:
            problems.append(f"facility {s.get('facilityId')} != {self.loc.facility_id}")
        line = ", ".join(x for x in [" ".join(addr.get("addressLines", [])), str(addr.get("cityTown", "")).title(),
                                     f"{addr.get('stateProvince', '')} {addr.get('postalCode', '')}".strip()] if x)
        self.report["store"] = {"locationId": s.get("locationId"), "facilityId": s.get("facilityId"), "brand": s.get("brand"),
                                "storeNumber": s.get("storeNumber"), "name": addr.get("name") or s.get("vanityName"),
                                "address": line, "problems": problems,
                                "verified_from": _body_file(ok[-1]) if ok else ""}
        self._step("store_check", ok=not problems, store=self.report["store"])
        if problems:
            raise SourceError(f"store check failed, no product request sent: {problems}")
        self.site.set_store(self.loc.location_id, str(s.get("facilityId")), {
            "addressLines": addr.get("addressLines", []), "cityTown": addr.get("cityTown", ""), "countryCode": "US",
            "postalCode": addr.get("postalCode", ""), "stateProvince": addr.get("stateProvince", "")})
        self.report["laf_header"] = json.loads(self.site.laf)
        self._store_generation = self.site.session_generation
        self.report.setdefault("store_checks", []).append({"session_generation": self._store_generation,
                                                           "verified_from": self.report["store"]["verified_from"], **_stamp()})
        return {"zipcode": str(addr.get("postalCode", ""))[:5], "store_name": self.loc.store_name, "store_location": line,
                "store_logo": self.loc.store_logo, "location_id": self.loc.location_id}

    def _collect(self) -> None:
        products, since = self._on_verified_session(lambda: self.site.products(self.upcs))
        ok = self._last_ok(since)
        name = _body_file(ok[-1]) if ok else ""
        self.origin = {p.get("id"): f"{name}#data.products[{i}]" for i, p in enumerate(products)}
        returned = {p.get("id") for p in products}
        missing = [u for u in self.upcs if u not in returned]
        if missing:   # same second ask as the full run: without the 'verified' filter
            more, since = self._on_verified_session(lambda: self.site.products(missing, verified=False))
            ok = self._last_ok(since)
            name = _body_file(ok[-1]) if ok else ""
            for i, p in enumerate(more):
                if p.get("id") in missing and p.get("id") not in self.origin:
                    self.origin[p.get("id")] = f"{name}#data.products[{i}]"
                    products.append(p)
        self.products = products
        # store and fulfilment context must hold in every product response; the first one that does not ends the probe
        wanted = self.site.sent_context()
        for p in products:
            src = str(((p.get("price") or {}).get("storePrices") or {}).get("sourceLocationId", ""))
            if src and src != self.loc.location_id:
                self._step("products", returned=len(products), context_lost_on=p.get("id"), sourceLocationId=src)
                raise SourceError(f"store context lost: product {p.get('id')} priced for {src}, expected {self.loc.location_id}")
            echoed = echoed_context(p)
            if echoed != wanted:
                self._step("products", returned=len(products), context_lost_on=p.get("id"), echoed=list(echoed), sent=list(wanted))
                raise SourceError(f"fulfilment context lost: product {p.get('id')} answered for {echoed}, sent {wanted}")
        self.report["context_verified"] = {"products": len(products), "modality": wanted[0], "store_id": wanted[1],
                                           "facility_id": wanted[2], "from": "laf echoed in each product response"}
        self._step("products", returned=len(products), missing=[u for u in self.upcs if u not in {p.get('id') for p in products}])
        if products:
            self.coupons, since = self._on_verified_session(lambda: self.site.coupons_for(self.upcs))
            for att in self._last_ok(since):            # every coupon page, each linked to its own raw file
                try:
                    page = json.loads(att["body"]).get("data", {}).get("coupons") or []
                except ValueError:
                    continue
                for i, c in enumerate(page):
                    self.coupon_origin.setdefault(str(c.get("id")), f"{_body_file(att)}#data.coupons[{i}]")
        self._step("coupons", returned=len(self.coupons))

    # --- mapping and acceptance ----------------------------------------------------------------------------
    def _map(self, location: dict) -> None:
        by_upc: dict[str, list[dict]] = {}
        for c in self.coupons:
            for u in c.get("upcs") or []:
                by_upc.setdefault(u, []).append(c)
        crawl = datetime.now(timezone.utc).strftime("%Y-%m-%d")   # same crawl_date rule as the full run
        by_id = {p.get("id"): p for p in self.products}
        records, summary, trace, rows = [], [], [], {}
        for upc in self.upcs:
            role = self.roles.get(upc, "any")
            if upc not in by_id:
                row = {"upc": upc, "role": role, "returned": "no",
                       "kept_in_output": "identity-only in a full run (UPC kept, nothing invented)", "raw_ref": "not listed by the site"}
                summary.append(row)
                rows[upc] = (row, None, None, None)
                trace.append({"upc": upc, "role": role, "returned": False})
                continue
            p = by_id[upc]
            coupons = by_upc.get(upc, [])
            rec, reason, meta = self.mapper.map_site(p, coupons, location=location, category="probe",
                                                     sub_category="", crawl_date=crawl)
            item, sp = p.get("item") or {}, (p.get("price") or {}).get("storePrices") or {}
            regular, promo = sp.get("regular") or {}, sp.get("promo") or {}
            echoed = echoed_context(p)
            row = {"upc": upc, "role": role, "returned": "yes", "title": fix_text(item.get("description", "")),
                   "size": fix_text(item.get("customerFacingSize", "")), "sourceLocationId": meta["source_location"],
                   "store_match": "yes" if meta["source_location"] == self.loc.location_id else ("no" if meta["source_location"] else "no price"),
                   "echoed_modality": echoed[0], "echoed_store_id": echoed[1], "echoed_facility_id": echoed[2],
                   "price_status": meta.get("price_status", ""),
                   "fulfillment_listed": "|".join(meta["fulfillment_listed"]), "fulfillment_sellable": "|".join(meta["fulfillment"]),
                   "stock_level": meta.get("stock_level", ""), "regular_price": usd(regular.get("price")),
                   "promo_price": usd(promo.get("price")),
                   "ppm": rec["ppm"] if rec else fix_text((promo or regular).get("equivalizedUnitPriceString", "")),
                   "sale_price": rec["sale_price"] if rec else "",
                   "volume_deals_description": rec["volume_deals_description"] if rec else "",
                   "volume_deals_price": rec["volume_deals_price"] if rec else "",
                   "digital_deals_description": rec["digital_deals_description"] if rec else "",
                   "digital_deals_price": rec["digital_deals_price"] if rec else "",
                   "coupons_returned": len(coupons),
                   "kept_in_output": "yes" if rec else "no", "exclusion_reason": reason, "raw_ref": self.origin.get(upc, ""),
                   "coupon_refs": "|".join(self.coupon_origin.get(str(c.get("id")), "") for c in coupons)}
            summary.append(row)
            rows[upc] = (row, p, rec, meta)
            trace.append({"upc": upc, "role": role, "returned": True, "raw_ref": row["raw_ref"],
                          "coupon_refs": [self.coupon_origin.get(str(c.get("id")), "") for c in coupons],
                          "echoed_context": list(echoed), "meta": meta, "record": rec, "excluded": reason})
            if rec:
                records.append(rec)
        with (self.out / "products.csv").open("w", newline="", encoding="utf-8") as fh:
            w = csv.DictWriter(fh, fieldnames=FIELD_ORDER)
            w.writeheader()
            w.writerows(records)
        with (self.out / "summary.csv").open("w", newline="", encoding="utf-8") as fh:
            w = csv.DictWriter(fh, fieldnames=SUMMARY_FIELDS, extrasaction="ignore")
            w.writeheader()
            for row in summary:
                w.writerow({k: row.get(k, "") for k in SUMMARY_FIELDS})
        self._write("trace.jsonl", "".join(json.dumps(t, ensure_ascii=False) + "\n" for t in trace))
        self.report["results"] = {
            "upcs": len(self.upcs), "returned": sum(1 for s in summary if s["returned"] == "yes"),
            "store_price": sum(1 for s in summary if s.get("price_status") == "store_price"),
            "with_ppm": sum(1 for s in summary if s.get("ppm")),
            "with_volume_terms": sum(1 for s in summary if s.get("volume_deals_description")),
            "with_digital_terms": sum(1 for s in summary if s.get("digital_deals_description")),
            "unavailable_kept": sum(1 for s in summary if s["returned"] == "yes" and not s.get("fulfillment_sellable") and s["kept_in_output"] == "yes"),
            "records_written": len(records)}
        self.report["acceptance"] = self._acceptance(rows)
        self._step("mapped", **self.report["results"], acceptance=self.report["acceptance"]["overall"])

    def _acceptance(self, rows: dict) -> dict:
        """Pass / fail / inconclusive per criterion. A criterion the site does not serve today is inconclusive."""
        checks = []

        def add(name, upc, result, detail):
            checks.append({"check": name, "upc": upc, "result": result, "detail": detail})

        returned = [u for u in self.upcs if rows[u][1] is not None]
        if not returned:
            add("products_returned", "", "fail", "the site returned none of the products")
        elif len(returned) < len(self.upcs):
            add("products_returned", "", "inconclusive",
                f"{len(returned)} of {len(self.upcs)} returned; not returned: {[u for u in self.upcs if u not in returned]}")
        else:
            add("products_returned", "", "pass", f"all {len(self.upcs)} returned")
        echoes = [a.get("laf_echo") for a in self.site.attempts
                  if a.get("outcome") == "retailer_200" and "/product/v2/products" in a.get("path", "")]
        if self.transport != "oxylabs":
            pass                                      # the browser sends the header itself; nothing to echo
        elif not echoes:
            add("store_header_forwarded", "", "inconclusive", "no product answer to check")
        elif "mismatch" in echoes:
            add("store_header_forwarded", "", "fail", f"Oxylabs reports a different X-Laf-Object than sent: {echoes}")
        elif "missing" in echoes:
            add("store_header_forwarded", "", "inconclusive",
                f"Oxylabs' echoed request headers do not list X-Laf-Object for every product answer: {echoes}")
        else:
            add("store_header_forwarded", "", "pass",
                "Oxylabs' echoed request headers list X-Laf-Object equal to the one sent (supporting evidence only; "
                "the store and fulfilment context are proven by each product answer below)")
        wanted = self.site.sent_context()
        for upc in self.upcs:
            row, p, rec, meta = rows[upc]
            role = self.roles.get(upc, "any")
            if p is None:
                if role == "unavailable":
                    add("unavailable_retained", upc, "inconclusive", "not returned by the site today (a full run keeps it UPC-only)")
                continue
            sp = (p.get("price") or {}).get("storePrices") or {}
            priced = bool(usd((sp.get("regular") or {}).get("price")) or usd((sp.get("promo") or {}).get("price")))
            src = str(sp.get("sourceLocationId", ""))
            if echoed_context(p) != wanted:
                add("source_context", upc, "fail", f"answer echoes {echoed_context(p)}, sent {wanted}")
            elif priced and src != self.loc.location_id:
                add("source_context", upc, "fail", f"price present but sourceLocationId is '{src or 'missing'}': price not attributable to {self.loc.location_id} (blanked in the record)")
            elif priced:
                add("source_context", upc, "pass", f"echoes {wanted[0]}/{wanted[1]}/{wanted[2]}; price sourceLocationId {src}")
            else:
                add("source_context", upc, "pass", f"echoes {wanted[0]}/{wanted[1]}/{wanted[2]}; no price returned")
            sellable = "PICKUP" in meta["fulfillment"]
            if role in ("ppm-volume", "ppm-digital", "regular"):
                add("pickup_sellable", upc, "pass" if sellable else "inconclusive",
                    "PICKUP sellable at the store" if sellable else f"not sellable for PICKUP today (sellable: {meta['fulfillment'] or 'none'})")
                shown = set()
                for o in [sp.get("regular") or {}, sp.get("promo") or {}] + \
                         [x for s in p.get("fulfillmentSummaries") or [] for x in (s.get("regular") or {}, s.get("sale") or {})]:
                    if o.get("equivalizedUnitPriceString"):
                        shown.add(fix_text(o["equivalizedUnitPriceString"]))
                got = rec["ppm"] if rec else ""
                if not shown:
                    add("direct_ppm", upc, "inconclusive", "the source shows no price per measure for this product today")
                elif got and got in shown:
                    add("direct_ppm", upc, "pass", f"ppm '{got}' equals the string in the source answer")
                else:
                    add("direct_ppm", upc, "fail", f"ppm '{got}' not among the source strings {sorted(shown)}")
            if role == "ppm-volume":
                expected = _volume_terms(p)
                desc = rec["volume_deals_description"] if rec else ""
                price = rec["volume_deals_price"] if rec else ""
                if not expected:
                    add("volume_offer", upc, "inconclusive", "the source serves no quantity offer for this product today")
                else:
                    missing = [t for t in expected if t != "__price__" and t not in desc]
                    want_price = expected.get("__price__")
                    if want_price is not None and price != want_price:
                        missing.append(f"volume_deals_price '{price}' != source '{want_price}'")
                    if missing:
                        add("volume_offer", upc, "fail", f"source terms missing from the record: {missing}")
                    else:
                        add("volume_offer", upc, "pass", f"every source term present ({len([t for t in expected if t != '__price__'])}): '{desc[:200]}'")
            if role == "ppm-digital":
                desc = rec["digital_deals_description"] if rec else ""
                price = rec["digital_deals_price"] if rec else ""
                clip = [c for c in self._coupons_for(upc) if _is_clip_coupon(c)]
                n = int(row["coupons_returned"])
                if n == 0:
                    add("digital_coupon", upc, "inconclusive", "the site's coupon lookup returned no coupon for this UPC today")
                elif not clip:
                    add("digital_coupon", upc, "inconclusive", f"{n} coupon(s) returned but none is a clip coupon (points / cash back are not digital deals)")
                else:
                    missing = []
                    for c in clip:
                        for label, term in _coupon_terms(c):
                            if term not in desc:
                                missing.append(f"coupon {c.get('id')} {label}: '{term[:80]}'")
                    want_price = SiteMapper.digital_price(clip)
                    if price != want_price:
                        missing.append(f"digital_deals_price '{price}' != price stated by the source '{want_price}'")
                    if missing:
                        add("digital_coupon", upc, "fail", f"source terms missing from the record: {missing}")
                    else:
                        add("digital_coupon", upc, "pass", f"{len(clip)} clip coupon(s); title, requirement (quantity, eligibility, limits, dates), expiry and stated price all equal the source")
            if role == "unavailable":
                proof = _unavailable_proof(p)
                if rec is None:
                    add("unavailable_retained", upc, "fail", f"returned but not kept in products.csv ({row['exclusion_reason']})")
                elif sellable:
                    add("unavailable_retained", upc, "inconclusive", "sellable for PICKUP today, so retention of an unavailable product is not shown")
                elif not proof:
                    add("unavailable_retained", upc, "inconclusive", "kept, but the source gives no availability for it (no PICKUP fulfilment summary, no stock level): unavailability is not established")
                else:
                    add("unavailable_retained", upc, "pass", f"kept in products.csv; source shows {proof}; prices '{rec['regular_price'] or 'empty'}'")
        results = {c["result"] for c in checks}
        overall = "fail" if "fail" in results else "inconclusive" if "inconclusive" in results else "pass"
        return {"overall": overall, "checks": checks,
                "rule": "fail if any check fails; inconclusive if any criterion was not served by the site today; pass only if every check passed"}

    def _coupons_for(self, upc: str) -> list[dict]:
        return [c for c in self.coupons if upc in (c.get("upcs") or [])]

    # --- evidence -------------------------------------------------------------------------------------------
    def _finish(self, code: int, lock_timeout: float | None = None) -> None:
        if not self._lock.acquire(timeout=-1 if lock_timeout is None else lock_timeout):
            return                     # another writer holds the evidence (stalled): the watchdog ends the process
        try:
            self._write_evidence(code)
        finally:
            self._lock.release()

    def _write_evidence(self, code: int) -> None:
        if self._finished:
            return
        self._finished = True
        self.report["exit_code"] = code
        self.report["finished"] = _stamp()
        self.report["requests"] = {"sent": self.site.stats.requests, "retries": self.site.stats.retries,
                                   "errors": self.site.stats.errors, "seconds_in_requests": round(self.site.stats.total_seconds, 1),
                                   "elapsed_seconds": round(self.site.elapsed(), 1),
                                   "finalization": {"watchdog_after_deadline_s": WATCHDOG_GRACE_SECONDS,
                                                    "evidence_write_limit_s": WATCHDOG_WRITE_SECONDS},
                                   "by_outcome": dict(getattr(self.site, "outcomes", {}) or {}),
                                   "session_replacements": self.site.session_generation,
                                   "note": "every dispatched attempt counts; errors = every failed attempt (provider or retailer); "
                                           "browser start-up page assets are not JSON requests and are not counted"}
        if self.allowance is not None and self.allowance.reservation_id:
            self.report["allowance"]["jobs_this_run"] = self.site.stats.requests
            try:
                led = self.allowance.finalize(self.site.stats.requests, code, self.report["finished"]["utc"])
                self.report["allowance"]["final_update"] = "verified"
                self.report["allowance"]["counted_after"] = led["total_jobs_counted"]
                self.report["allowance"]["ledger_sha256_after_final"] = self.allowance.last_sha256
                self._write("cumulative-usage.json", encode(led).decode("utf-8"))
            except LedgerError as exc:
                # the reservation stays counted in full; the run is reported as stopped and the ledger as unresolved
                self.report["allowance"]["final_update"] = f"UNRESOLVED: {type(exc).__name__}: {exc}"
                self.report["stop_reason"] = (self.report.get("stop_reason") or "") +                     f"; ledger final update not verified ({exc}): reservation stays counted, ledger unresolved"
                self.report["transport"] = "stopped"
                code = 2
                self.report["exit_code"] = code
                self._exit_override = code
        index = []
        for att in list(self.site.attempts):
            entry = {k: att.get(k) for k in ("seq", "t", "elapsed_s", "path", "attempt", "transport", "outcome", "status",
                                             "content_type", "oxylabs_http", "job_id", "session_generation",
                                             "laf_echo", "ms", "error")}
            name = _body_file(att)
            if name:
                self._write(name, att["body"])
                entry.update(body_file=name, body_sha256=hashlib.sha256(att["body"].encode("utf-8")).hexdigest())
            if att.get("oxylabs_body") is not None:
                env = f"raw/{att['seq']:02d}-{_kind(att['path'])}.oxylabs.json"
                self._write(env, att["oxylabs_body"])
                entry.update(envelope_file=env, envelope_sha256=hashlib.sha256(att["oxylabs_body"].encode("utf-8")).hexdigest())
            index.append(entry)
        self._write("raw/index.jsonl", "".join(json.dumps(e, ensure_ascii=False) + "\n" for e in index))
        calls = [{k: v for k, v in c.items() if k != "raw"} for c in self.site.calls]
        self._write("requests.jsonl", "".join(json.dumps(c) + "\n" for c in calls))
        jobs = getattr(self.site, "jobs", None)
        if jobs is not None:
            self._write("oxylabs_jobs.jsonl", "".join(json.dumps(j, ensure_ascii=False) + "\n" for j in jobs))
        self._write("report.json", json.dumps(self.report, indent=1, ensure_ascii=False))
        lines = []
        for path in sorted(p for p in self.out.rglob("*") if p.is_file() and p.name != "manifest.sha256"):
            lines.append(f"{hashlib.sha256(path.read_bytes()).hexdigest()}  {path.relative_to(self.out).as_posix()}")
        self._write("manifest.sha256", "\n".join(lines) + "\n")
        log("info", "probe_finished", exit_code=code, transport=self.report["transport"], stop_reason=self.report["stop_reason"],
            acceptance=self.report["acceptance"]["overall"], out=str(self.out), requests=self.site.stats.requests)