"""Window driver for the #304 premise-line follow-up lane. Six arms over two cells: GREEN at base, GREEN at tip, RED at tip under HERTZ_RIG_TMP_IN_REPO=1. The control's whole point is WHICH LINE reds, so this driver does not settle for a pass/fail bit -- it extracts the panic's file:line out of the producer's own output and compares it to the premise line and to the product line the premise line is supposed to stand in front of. Two rules this rig has already paid for, kept structural here: - AN ARM RECORDS THE CONDITIONS IT ACTUALLY RESOLVED. Window 6's arm B drove the TMP switch through env_extra (child-only) while lane_run.priv_for reads the PARENT env, so the arm measured nothing and its PASS read exactly like a control that fired and disagreed. Every arm here records the TMP base priv_for() actually returns, and a control arm REFUSES to call itself a measurement when that base matches the base its paired arm resolved. - A FILTER THAT CANNOT EXPRESS THE HUNT RETURNS A CLEAN ZERO. The line extraction is asserted to have FOUND something on any red arm; "no line matched" is reported as UNREADABLE, never as agreement. """ import io, json, os, re, subprocess, sys, time sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) import lane_run as L TREE = os.environ.get("HERTZ_LANE_TREE") if not TREE or not os.path.isdir(TREE): raise SystemExit("HERTZ_LANE_TREE must name this lane's worktree: %r" % TREE) BASE = os.environ.get("HERTZ_BASE_SHA", "4b54a95d") TIP = os.environ.get("HERTZ_TIP_SHA", "58120cfe") DEADLINE = int(os.environ.get("HERTZ_ARM_DEADLINE", "900")) # The two cells, with the line each one's premise assert guards and the product # line it stands in front of. A red at PRODUCT under the control means the # premise assert is in the wrong place -- doyle's caution, made checkable. CELLS = { "registryhost": { "filter": "registryhost::tests::recent_projects_for_dedups_newest_first_excludes_spt_internal", "file": "registryhost.rs", }, "projwriter": { "filter": "projwriter::tests::batched_complexity_counters_hold", "file": "projwriter.rs", }, } START_FILE = os.environ.get("HERTZ_START_FILE") if not START_FILE or not os.path.isfile(START_FILE): raise SystemExit( "REFUSING TO RUN: set HERTZ_START_FILE to the START body. A window opens " "only on the gater's word, and the START body is the record of it.") START_BODY = io.open(START_FILE, "rb").read().decode("utf-8", "replace") REPORT = os.path.join(L.EVID, "window-premise-%s.json" % time.strftime( "%H%M%SZ", time.gmtime())) if os.path.exists(REPORT): raise SystemExit("report path already exists, refusing to overwrite: %s" % REPORT) PANIC_LINE = re.compile(r"([A-Za-z0-9_]+\.rs):(\d+):(\d+)") def git(*args): r = subprocess.run(("git",) + args, cwd=TREE, capture_output=True, text=True) if r.returncode != 0: raise SystemExit("git %s failed: %s" % (" ".join(args), r.stderr.strip())) return r.stdout.strip() def head_sha(): return git("rev-parse", "--short", "HEAD") def panic_sites(label, want_file): """Every : the producer printed for the cell's own file. Returns (sites, unreadable). An empty list on a RED arm is UNREADABLE, not agreement: it means this filter could not express the hunt. """ sites = [] for name in (label + ".stdout", label + ".raw"): p = os.path.join(L.EVID, name) if not os.path.isfile(p): continue text = io.open(p, "rb").read().decode("utf-8", "replace") for m in PANIC_LINE.finditer(text): if m.group(1) == want_file: sites.append(int(m.group(2))) return sorted(set(sites)) RUN_COUNT = re.compile(r"(\d+) tests? run") def tests_run(label): """How many cells nextest actually EXECUTED. None when the summary is unreadable -- which is a hole, never a zero. "1 test run" is singular, so a filter anchored on the plural would itself read a clean zero here. """ for name in (label + ".raw", label + ".stdout"): p = os.path.join(L.EVID, name) if not os.path.isfile(p): continue text = io.open(p, "rb").read().decode("utf-8", "replace") hits = RUN_COUNT.findall(text) if hits: return int(hits[-1]) return None def arm(label, cell, sha, control): """One producer. Records the TMP base it ACTUALLY resolved, from the parent env, which is the only place lane_run.priv_for reads it from.""" if control: os.environ["HERTZ_RIG_TMP_IN_REPO"] = "1" else: os.environ.pop("HERTZ_RIG_TMP_IN_REPO", None) resolved_tmp = L.priv_for(label) spec = CELLS[cell] # --no-tests=fail AND a parsed count, because a nextest filter that matches # NOTHING exits 0 and reads exactly like a pass. --success-output immediate # because a nextest GREEN prints the cell's own output nowhere otherwise, # and the green arms are where the premise assert must be seen executing. argv = ["cargo", "nextest", "run", "-p", "spt-daemon", "--lib", "--no-fail-fast", "--no-tests", "fail", "--success-output", "immediate", "-E", "test(=%s)" % spec["filter"]] obs = L.run(label, argv, TREE, DEADLINE) rc = obs["producer_exit"] sites = panic_sites(label, spec["file"]) ran = tests_run(label) if ran != 1: verdict = "NON-MEASUREMENT" # zero-match filter, or a widened one elif rc == 0: verdict = "GREEN" else: verdict = "RED" return { "label": label, "cell": cell, "sha": sha, "control": control, "producer_exit": rc, "tests_run": ran, "verdict": verdict, "stop_reason": obs["stop_reason"], "elapsed_seconds": obs["elapsed_seconds"], "survivors": obs["survivors"], "resolved_tmp": resolved_tmp, "tmp_in_repo": os.path.abspath(resolved_tmp).lower().startswith( os.path.abspath(L.EVID).lower()), "panic_lines": sites, "unreadable": (rc != 0 and not sites), } def main(): out = {"start_body": START_BODY, "base": BASE, "tip": TIP, "tree": TREE, "pool": L.POOL, "utc_start": time.strftime( "%Y-%m-%dT%H:%M:%SZ", time.gmtime()), "arms": []} # TIP first: the pool is already warm for it, and the base arms are a # two-file incremental step away. # VERIFY THE SUBJECT before any arm: an arm run against the wrong checkout # is a measurement of something else that reads as an answer. head = head_sha() if not (head.startswith(TIP) or TIP.startswith(head)): raise SystemExit("expected HEAD at tip %s, found %s" % (TIP, head)) if subprocess.run(("git", "status", "--porcelain"), cwd=TREE, capture_output=True, text=True).stdout.strip(): raise SystemExit("tree is dirty; arms must run on committed bytes") for cell in CELLS: out["arms"].append(arm("tip-%s" % cell, cell, head_sha(), control=False)) for cell in CELLS: out["arms"].append(arm("ctl-%s" % cell, cell, head_sha(), control=True)) git("checkout", "--detach", BASE) try: for cell in CELLS: out["arms"].append(arm("base-%s" % cell, cell, head_sha(), control=False)) finally: git("checkout", "test/304-premise-lines") # An arm pair that resolved the SAME tmp base measured nothing (window 6's # defect). Refuse to call the control a measurement in that case. for cell in CELLS: t = next(a for a in out["arms"] if a["label"] == "tip-%s" % cell) c = next(a for a in out["arms"] if a["label"] == "ctl-%s" % cell) if t["resolved_tmp"] == c["resolved_tmp"]: c["verdict"] = "NON-MEASUREMENT" c["why"] = ("control resolved the same TMP base as its paired arm; " "the switch did not reach lane_run.priv_for") out["utc_end"] = time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()) json.dump(out, open(REPORT, "x"), indent=2) print(json.dumps([{k: a[k] for k in ("label", "verdict", "producer_exit", "tests_run", "panic_lines", "tmp_in_repo", "unreadable")} for a in out["arms"]], indent=2)) print("report:", REPORT) if __name__ == "__main__": main()