# -*- coding: utf-8 -*-
"""ladder.py — the test record (`update_ladder:`) in `.felhom.yml`: read it, check it, write it.

`09-update-architecture.md` §3 decision 13 — *the test decides, not the tag*: the catalog holds only
tested steps, and an image move with no test record is refused at push time. §6.4 part 4 is the
build; part 5 (the box climbing the ladder) and the digest half of part 6 on the box are NOT this.

THE FORMAT, chosen for the two readers it has (spiked live 2026-09-23 on controller v0.266.0 and
v0.267.0 — the controller ignores the unknown top-level key and deploys, probes and badges as before):

    update_ladder:
      - {"from": {...}, "to": {...}, "digest": {...}, "verdict": "proven", ...}

ONE JSON OBJECT PER LINE. JSON is a subset of YAML's flow style, so the controller's YAML parser
reads it; and the catalog CI runner has NO PyYAML (it carries python3 and git only), so the gate
reads it with `json.loads` — no parser that can be missing, no degraded mode. A line under
`update_ladder:` that is not exactly `  - {json}` is a conviction, never a skip.

AN ENTRY (all keys required unless marked):
  from, to          {service: image ref} for EVERY service with an image: line, before / after
  digest            {service: "sha256:<64 hex>"} for every service in `to` — what the registry
                    served for that ref when the entry was written (decision 17)
  verdict           "proven" | "unrecorded" (backfill only: a live move with no record found)
  tested_at         RFC 3339, or null for "unrecorded"
  harness_version   int (2 = the memory watch), or null for "unrecorded"
  evidence          path of the verdict record(s), relative to the workspace root
  memory_peak_pct   the memory watch's worst container peak in % of its limit; null only on a
                    backfilled entry (harness v1 had no watch)
  marks             {"files_may_change": bool, "needs_person": null | "<why>", "memory_tight": bool}
  engine_conversion (optional) {"service", "engine": "postgres", "from": int, "to": int} — the box
                    CONVERTS this PostgreSQL major (controller v0.273.0, `09` §6.4 part 10); written by
                    `upgrade-test.py --write-ladder` only when the bench AND the box both converted it
  backfilled        (optional) "YYYY-MM-DD" — written by the backfill from an EXISTING record,
                    never by a new test; a new move may not carry it
  digest_from       (a RE-TEST only, `09` §3 decision 52) {service: "sha256:…"} — the digest the re-test
                    was run FROM. A re-test is an entry whose `from` equals its `to` (the same tags): the
                    catalog proved the same tag at a NEW digest (an upstream same-name fix). It needs
                    `digest_from` for every service, at least one service whose `digest` differs from it,
                    and `box_evidence` (both venues); `check-test-record.py` rule 2b ties `digest_from` to
                    the previous entry's `digest`. Written by `upgrade-test.py --write-ladder` from a
                    `--retest` verdict, never by hand.

STEP DEFINITIONS (`09` §6.4 part 5, controller v0.268.0): every entry but the NEWEST carries its own
complete compose file at `templates/<app>/steps/<step_key(to)>.yml` — the box climbs one step at a time
and pins exactly that file; the newest step's definition is the template's `docker-compose.yml`. The box's
catalog clone is `--depth 1`, so git history is not a place a box can read a step from, and the commit
that moved an image is not always the definition that works (romm 15f9ebf). `step_key` is computed the
SAME way by the controller (`stacks.StepKey`, pinned by TestLadder_StepKeyMatchesTheCatalog).

Every path that reads or writes the format is here, so the gate and the writer cannot disagree.
"""
import hashlib
import json
import re

LADDER_KEY_RE = re.compile(r"^update_ladder:\s*$")
ENTRY_RE = re.compile(r"^  - (\{.*\})\s*$")
SERVICE_RE = re.compile(r"^  ([A-Za-z0-9_-]+):\s*$")
IMAGE_RE = re.compile(r"^\s+image:\s*[\"']?([^\s\"'#]+)")
DIGEST_RE = re.compile(r"^sha256:[0-9a-f]{64}$")
TS_RE = re.compile(r"^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}(\.\d+)?(Z|[+-]\d{2}:\d{2})$")
DATE_RE = re.compile(r"^\d{4}-\d{2}-\d{2}$")
VERDICTS = ("proven", "unrecorded")
MEMORY_TIGHT_PCT = 80.0


def images_in(compose_text):
    """{service: image} — per service, from that service's OWN `image:` line (the same reading
    check-engine-major.py and check-catalog-since.py make)."""
    out, cur = {}, None
    for line in compose_text.splitlines():
        m = SERVICE_RE.match(line)
        if m:
            cur = m.group(1)
            continue
        mi = IMAGE_RE.match(line)
        if mi and cur and cur not in out:
            out[cur] = mi.group(1)
    return out


def parse(felhom_text):
    """(entries, raw_lines, errors). No ladder → ([], [], []). A malformed line is an ERROR."""
    lines = felhom_text.splitlines()
    start = None
    for i, l in enumerate(lines):
        if LADDER_KEY_RE.match(l):
            if start is not None:
                return [], [], ["update_ladder: appears twice"]
            start = i
    if start is None:
        return [], [], []
    entries, raws, errors = [], [], []
    for l in lines[start + 1:]:
        if not l.strip() or l.lstrip().startswith("#"):
            continue
        if not l.startswith(" "):
            break  # the next top-level key: the block has ended
        m = ENTRY_RE.match(l)
        if not m:
            errors.append("not a one-line JSON entry under update_ladder: %r" % l[:120])
            continue
        try:
            e = json.loads(m.group(1))
        except ValueError as ex:
            errors.append("entry is not valid JSON (%s): %r" % (ex, l[:120]))
            continue
        if not isinstance(e, dict):
            errors.append("entry is not an object: %r" % l[:120])
            continue
        entries.append(e)
        raws.append(m.group(1))
    if not entries and not errors:
        errors.append("update_ladder: is present but holds no entry")
    return entries, raws, errors


def check_entry(e):
    """Problems with ONE entry's shape, as sentences. Empty list = well-formed."""
    p = []
    for k in ("from", "to", "digest", "verdict", "tested_at", "harness_version", "evidence",
              "memory_peak_pct", "marks"):
        if k not in e:
            p.append("missing key %r" % k)
    if p:
        return p
    for k in ("from", "to", "digest"):
        if not isinstance(e[k], dict) or not e[k]:
            p.append("%r must be a non-empty {service: value} object" % k)
    if p:
        return p
    v = e["verdict"]
    if v not in VERDICTS:
        p.append("verdict %r is not allowed in a ladder (only %s — a failed or inconclusive test is "
                 "evidence, never a step a box may take)" % (v, "/".join(VERDICTS)))
    backfilled = e.get("backfilled")
    if backfilled is not None and not (isinstance(backfilled, str) and DATE_RE.match(backfilled)):
        p.append("backfilled must be a YYYY-MM-DD date")
    if v == "unrecorded" and backfilled is None:
        p.append("verdict 'unrecorded' exists only for the backfill of a move made before the gate")
    for svc, ref in e["to"].items():
        d = e["digest"].get(svc)
        if not (isinstance(d, str) and DIGEST_RE.match(d)):
            p.append("no sha256 digest for service %r (%s)" % (svc, ref))
    for svc in e["digest"]:
        if svc not in e["to"]:
            p.append("digest names a service %r that `to` does not" % svc)
    if e["from"] == e["to"] and v == "proven" and backfilled is None:  # a RE-TEST (decision 52)
        df = e.get("digest_from")
        if not isinstance(df, dict) or set(df) != set(e["to"]) or not all(isinstance(x, str) and DIGEST_RE.match(x) for x in df.values()):
            p.append("a re-test (from == to) needs digest_from: {service: sha256} for every service — the digest it was tested FROM")
        elif all(df[s] == e["digest"].get(s) for s in e["to"]):
            p.append("a re-test (from == to) whose digest is the same as its digest_from tests nothing new — no new digest")
        if not (isinstance(e.get("box_evidence"), str) and e["box_evidence"].strip()):
            p.append("a re-test (from == to) must cite BOTH venues: box_evidence is missing")
    elif e.get("digest_from") is not None:
        p.append("digest_from belongs only to a re-test (from == to)")
    conv = e.get("engine_conversion")
    if conv is not None:  # `09` §6.4 part 10 — the box converts ONLY on this mark
        if not (isinstance(conv, dict) and set(conv) == {"service", "engine", "from", "to"}
                and conv["engine"] == "postgres" and isinstance(conv["from"], int) and isinstance(conv["to"], int)
                and conv["from"] < conv["to"] and conv["service"] in e["to"]):
            p.append("engine_conversion must be {service (in `to`), engine: postgres, from < to (ints)}")
    marks = e["marks"]
    if not isinstance(marks, dict) or set(marks) != {"files_may_change", "needs_person", "memory_tight"}:
        p.append("marks must be exactly {files_may_change, needs_person, memory_tight}")
        marks = {}
    if v == "proven":
        if not (isinstance(e["tested_at"], str) and TS_RE.match(e["tested_at"])):
            p.append("a proven entry needs tested_at as an RFC 3339 time")
        if not (isinstance(e["evidence"], str) and e["evidence"].strip()):
            p.append("a proven entry must cite its evidence")
        peak = e["memory_peak_pct"]
        if backfilled is None:
            if not isinstance(e["harness_version"], int) or e["harness_version"] < 2:
                p.append("a new proven entry needs harness_version >= 2 (the memory watch)")
            if not isinstance(peak, (int, float)) or isinstance(peak, bool):
                p.append("a new proven entry needs memory_peak_pct from the memory watch")
        if isinstance(peak, (int, float)) and not isinstance(peak, bool) and marks:
            tight = peak > MEMORY_TIGHT_PCT
            if bool(marks.get("memory_tight")) != tight:
                p.append("marks.memory_tight=%s disagrees with memory_peak_pct=%s (tight above %d%%)"
                         % (marks.get("memory_tight"), peak, MEMORY_TIGHT_PCT))
    return p


def entry_line(e):
    """The one line the writer emits for an entry — key order fixed so diffs stay readable."""
    order = ["from", "to", "digest", "verdict", "tested_at", "harness_version", "evidence",
             "box_evidence", "memory_peak_pct", "marks", "backfilled", "note"]
    ordered = {k: e[k] for k in order if k in e}
    for k in e:
        if k not in ordered:
            ordered[k] = e[k]
    return "  - " + json.dumps(ordered, ensure_ascii=False, separators=(", ", ": "))


LADDER_HEADER = (
    "\n# update_ladder — the test record: one tested step per line, oldest first (JSON flow mappings,\n"
    "# `09-update-architecture.md` §6.4 part 4). WRITTEN BY scripts/upgrade-test.py, never by hand;\n"
    "# gated by scripts/check-test-record.py. An image: move without a proven entry here is refused.\n"
    "update_ladder:\n")


def append_entry(felhom_text, e):
    """Return felhom_text with `e` appended as the ladder's newest line (creating the block at the
    END of the file when absent — it is a top-level key and nothing may follow it inside the block)."""
    line = entry_line(e)
    lines = felhom_text.splitlines()
    start = None
    for i, l in enumerate(lines):
        if LADDER_KEY_RE.match(l):
            start = i
    if start is None:
        body = felhom_text.rstrip("\n") + "\n" + LADDER_HEADER + line + "\n"
        return body
    end = start + 1
    while end < len(lines) and (not lines[end].strip() or lines[end].startswith(" ")
                                or lines[end].lstrip().startswith("#")):
        end += 1
    # insert after the last entry line of the block
    last = start
    for j in range(start + 1, end):
        if ENTRY_RE.match(lines[j]):
            last = j
    lines.insert(last + 1, line)
    return "\n".join(lines) + "\n"


def step_key(to):
    """The 16-hex name of a step's definition: sha256 of `to` as canonical JSON (keys sorted, no spaces).
    The controller computes the same string (stacks.StepKey)."""
    return hashlib.sha256(json.dumps(to, sort_keys=True, separators=(",", ":")).encode()).hexdigest()[:16]


def step_file(to):
    """The step definition's path RELATIVE to the template directory."""
    return "steps/%s.yml" % step_key(to)


def step_meta_file(to):
    """The step's own `.felhom.yml` (R-664, 2026-09-24): its probe, memory limits and applied record — the
    box judges the step with it. RELATIVE to the template directory."""
    return "steps/%s.felhom.yml" % step_key(to)


def strip_ladder_block(felhom_text):
    """A `.felhom.yml` without its update_ladder block (and the header comment the writer puts above it):
    a step's own copy carries no ladder — only the template's is read for one."""
    lines = felhom_text.splitlines()
    for i, l in enumerate(lines):
        if l.startswith("# update_ladder") or l.startswith("update_ladder:"):
            return "\n".join(lines[:i]).rstrip("\n") + "\n"
    return felhom_text