diff --git a/.claude/skills/agentic-delivery/references/fanout-host-sizing.md b/.claude/skills/agentic-delivery/references/fanout-host-sizing.md index 7349ad503..814ddb29f 100644 --- a/.claude/skills/agentic-delivery/references/fanout-host-sizing.md +++ b/.claude/skills/agentic-delivery/references/fanout-host-sizing.md @@ -539,7 +539,19 @@ three classes. A red train is first re-run **alone and at low load** (`merge_train.py verify --rerun-cmd`: only the failing checks or specs, `nice -n 19`, `PERUN_LOW_LOAD=1`). A pass there is a load flake: the batch lands and the flake is logged. Only a red that reproduces is bisected. Measure it with `landing_watchdog.py check`: a low yield (PRs per landed train) or merges per hour below -target (policy `train_yield_floor`, `merges_per_hour_target`) says trains are being lost to reds or starved. +target (policy `train_yield_floor`, `merges_per_hour_target`) says trains are being lost to reds or starved. It also tells a dead +run from a slow one: + +- `Train run stuck: run 10617-70 posted landing on 3 PRs 140 min ago (expected <= 60 min), no outcome since. Check the train runner, + not the PRs.` The newest train activity is an in-flight marker past 1.5 times its expected time. Any later marker or landing + supersedes it, so a hand merge, or a landing stage that hangs after its first merge, silences this line; the stall line still + fires once nothing merges for the window. +- `Landing stalled: 0 merged to main in 60 min, 2 green PRs waiting. Fix landing before starting new work. No train run in 200 min: + start the train or check its runner.` The clause rides the stall line when train markers exist and the last train activity + (marker or landing) was an old outcome. +- After `landing_watchdog.py expect 'train_size>=12'` (quoted, or the shell reads `>=12` as a redirect): `Decision not in effect: + train_size>=12 set 2.0 h ago; the 2 trains since carried a median of 3 PRs. Check that the runner reads it.` A train counts by + when it began. ## A merge host is not accepted until it can be recovered without anyone's help diff --git a/.claude/skills/agentic-delivery/scripts/landing_watchdog.py b/.claude/skills/agentic-delivery/scripts/landing_watchdog.py index 7a40f19a7..77b65db85 100644 --- a/.claude/skills/agentic-delivery/scripts/landing_watchdog.py +++ b/.claude/skills/agentic-delivery/scripts/landing_watchdog.py @@ -47,7 +47,26 @@ yield: L of S trains landed ..." fires when fewer than half the recent trains landed a PR or one PR re-rode 5+ trains, and says it points at infra (timebox or load), not the PR. No ledger or zero trains = silent. Warn-only; shares the throttle, cache and CLI. -Throttle: the remote is asked (one `git fetch` of the default branch, one `gh pr list`) at most once per 15 minutes +Train liveness and decision effect (always on, no policy key): the train's progress marker is the `perun/train-stage` status on each open PR (merge_train.py `stage`). +`gh pr list` shows that status but not its description, so while it shows one, one `gh api graphql` call reads the markers of the open PRs into the branch: stage = first +word of the description, time = its `at=` unless that is more than 5 min ahead of the clock, else the status `createdAt`, eta = `eta<=Ns` (at most 24 h); a marker with +no stage word or no time is skipped, and no marker means no extra call. green() ignores that marker (it is not a check), so a PR left in a dead run or skipped once still +counts as waiting. The last train activity is the newest of every marker, whatever its run (a run id is minted per `merge_train.py stage` call, not per train, so it groups +nothing), and the newest first-parent commit on the branch (one local `git log -1`; a failure, or a time more than 5 min ahead of the clock, reads as none); an outcome beats a marker on a tie. +"Train run stuck: run R posted on K PRs X min ago (expected <= Y min), no outcome since. ..." fires only when that last activity is itself a queued, gating or +landing marker older than 1.5 x its eta (default 60 min), K counting the PRs that carry that run and stage. Any newer marker or any landing supersedes a live marker, so a +dead run's marker on a PR no later train touched stays quiet. When the stall line fires and the last activity is not a live marker but older than max(1.5 x expected, the check +window), the stall line gains " No train run in X min: start the train or check its runner."; a stuck run gets the stuck line and no clause, and fresh outcome markers +(trains that run but fail) get none. No markers at all = silent, so a repo without trains never alarms. `landing_watchdog.py expect 'train_size>=N' [--hours H]` (quote the +first argument, an unquoted `>=` is a shell redirect; N and H integers >= 1, H default 24; anything else exits 2) appends a decision to /perun-decisions.jsonl +(env PERUN_DECISIONS overrides), and "Decision not in effect: train_size>=N set X h ago; the K trains since carried a median of M PRs. ..." fires while the newest +unexpired decision is unmet by the median PR count of the >= 2 ledger trains started since it (train_yield.py stamps each train with its start time, else its end time; a +record with neither never counts), so a decision the runner never applied does not read as a slow train. Residuals: any landing supersedes a live marker, so a hand merge, +or a landing stage that hangs after its first merge, silences the stuck line (the stall line still fires once nothing merges for the window); markers come only from the +current heads of open PRs, so a re-push or a close mid-run drops one, and a failed second call reads as no markers; the `createdAt` fallback is the post time, not the run +start; train_size reads the local ledger only, so another host's trains are not seen. Warn-only; shares the throttle, cache and CLI; fails open. + +Throttle: the remote is asked (one `git fetch` of the default branch, one `gh pr list`, and one `gh api graphql` while an open PR carries a train marker) at most once per 15 minutes per repo; the verdict is cached between probes, so a standing stall costs no network call; a prompt's line is emitted once per session and again when it changes or after 15 minutes; a spawn always gets it, and a payload without session_id is not deduped. The cache is one small JSON file per repo under $PERUN_LANDING_DIR, else /perun/landing-watchdog ( = $XDG_CACHE_HOME or ~/.cache, beside the other Perun hook state). The stamp is written before the calls, so a @@ -68,6 +87,8 @@ import json import math import os +import re +import statistics import subprocess import sys import time @@ -86,6 +107,16 @@ LANE = "Lane stalled: {l} has {k} ready PRs, last merge {h} (limit {t} h). Land that lane before new work." LKG = "Last-known-good is {h:.1f} h old (limit {t} h): no full ci-tests pass recorded since. Check the merge gate." LINE = "Landing stalled: 0 merged to {b} in {n} min, {k} green PRs waiting. Fix landing before starting new work." +TRAIN_STAGE = "perun/train-stage" # the train's progress marker; same string as merge_train's STAGE_CONTEXT (merge_train imports this module, so not imported here) +LIVE = ("queued", "gating", "landing") # stages with no outcome yet +ETA_S = 3600 # expected run time when a marker carries no eta +STUCK = "Train run stuck: run {r} posted {s} on {k} PRs {x:.0f} min ago (expected <= {y:.0f} min), no outcome since. Check the train runner, not the PRs." +IDLE = " No train run in {x:.0f} min: start the train or check its runner." # appended to the stall line +DECISION = "Decision not in effect: train_size>={n} set {h:.1f} h ago; the {k} trains since carried a median of {m:g} PRs. Check that the runner reads it." +STAGES_Q = ("query($owner:String!,$repo:String!,$base:String!){repository(owner:$owner,name:$repo){pullRequests(states:OPEN,baseRefName:$base,first:100," + "orderBy:{field:CREATED_AT,direction:DESC}){nodes{commits(last:1){nodes{commit{statusCheckRollup{contexts(first:100){nodes{__typename " + "...on StatusContext{context,description,createdAt}}}}}}}}}}}") +EXPECT_USAGE = "usage: landing_watchdog.py expect 'train_size>=N' [--hours H] (quote the first argument; N and H integers >= 1, H default 24)" def _run(args, cwd, timeout=15): @@ -117,13 +148,19 @@ def blocker(pr, allow_forks=False): return "" +def _stage(r): + """True for the train's own progress marker (a StatusContext named TRAIN_STAGE), which is not a check.""" + return r.get("__typename") == "StatusContext" and r.get("context") == TRAIN_STAGE + + def green(pr, allow_forks=False): """True when the PR is queue-eligible (see blocker) with at least one check, all finished successfully. mergeable=UNKNOWN is GitHub still computing: counted as eligible here so stall signals do not flicker, while the - train itself still blocks it. + train itself still blocks it. The train's progress marker is ignored (pending while a run is live, failure after a + skip), as merge_train judges check runs only; any other pending status still blocks. """ - rolls = pr.get("statusCheckRollup") or [] + rolls = [r for r in pr.get("statusCheckRollup") or [] if not _stage(r)] return blocker(pr, allow_forks) in ("", "mergeable=UNKNOWN") and bool(rolls) and all(_check_ok(r) for r in rolls) @@ -217,8 +254,42 @@ def p0_exempt(prs, cwd, now): return 0 +def _stages(prs, branch, cwd): + """The train's progress markers on the open PRs into `branch` (their current heads), as statuses with a `description` and `createdAt`; [] when there are none. + + `gh pr list --json statusCheckRollup` exports a status's context, state and start time but not its description (checked on gh 2.99), and the stage, run id and + time live there. So when its output shows a marker, one `gh api graphql` call reads them; no marker, no extra call. Fails open: any gh or shape problem gives [].""" + if not any(_stage(s) for p in prs for s in p.get("statusCheckRollup") or []): + return [] + try: + out = json.loads(_run(["gh", "api", "graphql", "-f", "query=" + STAGES_Q, "-f", "base=" + branch, "-F", "owner={owner}", "-F", "repo={repo}"], cwd, timeout=30)) + return [s for p in out["data"]["repository"]["pullRequests"]["nodes"] for c in p["commits"]["nodes"] + for s in (c["commit"]["statusCheckRollup"] or {}).get("contexts", {}).get("nodes", []) if _stage(s)] + except Exception: # no marker is read, no line is added + return [] + + +def _marks(stages, now=None): + """[(run, stage, at, eta_s)] for the train progress markers in `stages` (see _stages): stage = first word of the status description, at = its `at=` unless that + is more than 300 s ahead of the clock, else the status createdAt, eta_s = `eta<=Ns` capped at 86400 else 0, run = `run=` else "?". A marker with no stage word or no + time is skipped, never guessed. The patterns are bounded, so a hand-made status cannot overflow a number or grow a line, and a made-up eta or future time cannot + keep a dead run from ever reading as stuck.""" + now, out = now or time.time(), [] + for r in stages: + d = str(r.get("description") or "") + at, eta, run = (re.search(x, d) for x in (r"\bat=(\d{1,10})\b", r"\beta<=(\d{1,9})s\b", r"\brun=([\w.-]{1,16})")) + try: + t = int(at[1]) if at and int(at[1]) <= now + 300 else _ts(r["createdAt"]) + except (KeyError, ValueError, TypeError): + continue + if d.split(): + out.append((run[1] if run else "?", d.split()[0], t, min(int(eta[1]), 86400) if eta else 0)) + return out + + def probe(cwd, minutes, allow_forks=False): - """(branch, merged_count, green_count, open_non_draft_count, green_ages_hours). Raises on any git/gh problem; callers fail open.""" + """(branch, merged_count, green_count, open_non_draft_count, green_ages_hours, train_markers, landed_at). `landed_at` is the epoch second of the newest first-parent + commit on origin/, None when that local `git log` fails. Raises on any other git/gh problem; callers fail open.""" branch = default_branch(cwd) _run(["git", "fetch", "-q", "origin", branch], cwd, timeout=20) merged = _run(["git", "log", "--first-parent", f"--since={minutes} minutes ago", "--format=%H", "origin/" + branch], cwd) @@ -226,7 +297,11 @@ def probe(cwd, minutes, allow_forks=False): "--json", "isDraft,mergeable,isCrossRepository,statusCheckRollup,createdAt,body,labels"], cwd, timeout=30)) now = time.time() ages = [a for a in (_age_h(p, now) for p in prs if green(p, allow_forks)) if a is not None] - return branch, len(merged.split()), sum(green(p, allow_forks) for p in prs), sum(not p.get("isDraft") for p in prs) - p0_exempt(prs, cwd, now), ages + try: + landed = int(_run(["git", "log", "-1", "--first-parent", "--format=%ct", "origin/" + branch], cwd)) + except Exception: # fail open: no landing time, the markers alone judge + landed = None + return branch, len(merged.split()), sum(green(p, allow_forks) for p in prs), sum(not p.get("isDraft") for p in prs) - p0_exempt(prs, cwd, now), ages, _marks(_stages(prs, branch, cwd), now), landed def settings(): @@ -334,9 +409,9 @@ def cache_file(cwd): return os.path.join(base, hashlib.sha1(top.encode()).hexdigest()[:16] + ".json") -def _stall(branch, merged, k, n_open, minutes, cap): - """The stall line and/or the WIP line, newline-joined, or None when neither applies.""" - lines = [LINE.format(b=branch, n=minutes, k=k)] if merged == 0 and k else [] +def _stall(branch, merged, k, n_open, minutes, cap, idle=None): + """The stall line and/or the WIP line, newline-joined, or None when neither applies. `idle` (minutes since the last train activity, from _runs) is appended to the stall line.""" + lines = [LINE.format(b=branch, n=minutes, k=k) + (IDLE.format(x=idle) if idle else "")] if merged == 0 and k else [] if cap and n_open > cap: lines.append(WIP.format(n=n_open, c=cap)) return "\n".join(lines) or None @@ -352,6 +427,73 @@ def _train_yield(cwd): return None +def _runs(marks, landed, minutes, now=None): + """(stuck line or None, idle minutes or None). `marks` is [(run, stage, at, eta_s)] (see _marks), `landed` the epoch second of the newest first-parent commit on the branch + or None; no markers = (None, None), so a repo without trains never alarms. + + The last activity is the newest of every marker and, when known and not more than 300 s ahead of the clock, a landing counted as an outcome at `landed`; on a tie in time an + outcome is the newer. Run ids do not + group markers: a run id is minted per `merge_train.py stage` call, not per train, so one train posts several. + Stuck: the last activity is itself a queued, gating or landing marker older than 1.5 x its expected time (eta, else ETA_S); the line names its run and stage and counts the + PRs carrying both. Any newer marker or any landing supersedes a live marker, so a dead run's marker on a PR no later train touched stays quiet. Residual: a hand merge, or a + landing stage that hangs after its first merge, therefore silences it (the stall line still fires once nothing merges for the window). + Idle: the last activity is not a live marker and is older than max(1.5 x expected, the check window); its age in minutes, which _stall appends to the stall line. A repo + that merges has no stall line, so a skipped marker from days ago never alarms there.""" + if not marks: + return None, None + now = now or time.time() + last = max(marks + ([("", "landed", landed, 0)] if landed and landed <= now + 300 else []), key=lambda m: (m[2], m[1] not in LIVE)) + age, eta = now - last[2], last[3] or ETA_S + if last[1] in LIVE: + return (STUCK.format(r=last[0], s=last[1], k=sum(m[:2] == last[:2] for m in marks), x=age / 60, y=eta / 60) if age > 1.5 * eta else None), None + return None, age / 60 if age > max(1.5 * eta, minutes * 60) else None + + +def decisions_path(cwd=None): + """Decision file: env PERUN_DECISIONS, else perun-decisions.jsonl under the git common dir of cwd (as train_yield.ledger_path). Raises RuntimeError outside a repo.""" + return os.environ.get("PERUN_DECISIONS") or os.path.join(_run(["git", "rev-parse", "--path-format=absolute", "--git-common-dir"], cwd), "perun-decisions.jsonl") + + +def expect(argv, now=None): + """Record `'train_size>=N' [--hours H]` (H default 24) in the decision file and return a one-line receipt (main prints it); quote the first argument, an unquoted `>=` is a shell redirect. + + Only the metric train_size with `>=`, N >= 1 and a whole-number H >= 1 are accepted; anything else raises ValueError(EXPECT_USAGE) and writes nothing. + Side effect: appends {"metric","op","value","at","until"} (epoch seconds) to decisions_path(). Raises OSError or RuntimeError when it cannot write.""" + m = re.fullmatch(r"train_size>=([0-9]{1,9})", argv[0]) if argv else None + hours = "24" if len(argv) == 1 else argv[2] if len(argv) == 3 and argv[1] == "--hours" else "" + if not (m and re.fullmatch(r"[0-9]{1,9}", hours) and int(m[1]) >= 1 and int(hours) >= 1): + raise ValueError(EXPECT_USAGE) + at = int(now or time.time()) + with open(decisions_path(os.getcwd()), "a") as f: + f.write(json.dumps({"metric": "train_size", "op": ">=", "value": int(m[1]), "at": at, "until": at + int(hours) * 3600}) + "\n") + return f"decision recorded: train_size>={int(m[1])} for {int(hours)} h" + + +def _decision(cwd, now=None): + """Decision-effect line, or None: the newest unexpired `train_size>=N` decision (see expect) while the >= 2 ledger trains started since it (train_yield.sizes_since) carried a median + of fewer than N PRs, i.e. the decision is not in effect. Met, expired, fewer than 2 trains, a record without a time, a missing or unreadable file = silent. + Fails open: never drops the other lines.""" + try: + sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) + import train_yield + now, ds = now or time.time(), [] + with open(decisions_path(cwd)) as f: + for ln in f.read().splitlines(): + try: + d = json.loads(ln) + if d["metric"] == "train_size" and d["op"] == ">=" and d["value"] >= 1 and d["at"] <= now < d["until"]: + ds.append(d) + except (ValueError, KeyError, TypeError): + continue + d = max(ds, key=lambda x: x["at"], default=None) + sizes = train_yield.sizes_since(train_yield.ledger_path(cwd), d["at"]) if d else [] + if len(sizes) > 1 and (m := statistics.median(sizes)) < d["value"]: + return DECISION.format(n=d["value"], h=(now - d["at"]) / 3600, k=len(sizes), m=m) + except Exception: # no file, no repo, no ledger: silent + pass + return None + + def _lkg(cwd): """Stale last-known-good alarm line (lkg.py), or None. Off unless policy `lkg_stale_h` > 0. Fails open: never drops the other lines.""" try: @@ -414,7 +556,8 @@ def times(): lane = _lanes(cwd, b, lanes if lanes is not None else lane_cfg()) except Exception: # fail open: lane lines never drop the others lane = None - return "\n".join(x for x in (_stall(b, merged, k, n_open, minutes, cap), extra, lane, _train_yield(cwd), _lkg(cwd), _red(cwd), _installer(cwd), _age(rest[0] if rest else [], ages_lim if ages_lim is not None else age_limits())) if x) or None + stuck, idle = _runs(rest[1] if len(rest) > 1 else [], rest[2] if len(rest) > 2 else None, minutes) + return "\n".join(x for x in (_stall(b, merged, k, n_open, minutes, cap, idle), extra, lane, _train_yield(cwd), stuck, _decision(cwd), _lkg(cwd), _red(cwd), _installer(cwd), _age(rest[0] if rest else [], ages_lim if ages_lim is not None else age_limits())) if x) or None def _event(data): @@ -494,6 +637,13 @@ def check(argv): def main(): if sys.argv[1:2] == ["check"]: return check(sys.argv[2:]) + if sys.argv[1:2] == ["expect"]: + try: + print(expect(sys.argv[2:])) + return 0 + except (ValueError, OSError, RuntimeError) as e: + sys.stderr.write(f"landing_watchdog: {e}\n") + return 2 try: data = json.load(sys.stdin) line = hook(data) diff --git a/.claude/skills/agentic-delivery/scripts/train_land.sh b/.claude/skills/agentic-delivery/scripts/train_land.sh index 6f21c0b9b..a59e924e3 100755 --- a/.claude/skills/agentic-delivery/scripts/train_land.sh +++ b/.claude/skills/agentic-delivery/scripts/train_land.sh @@ -112,7 +112,7 @@ export PERUN_GITHUB_ACTIONS PERUN_JOBS rm -f "$LOG.red" "$LOG.flaky" LOCK=$(git rev-parse --git-dir)/train-land.lock LT=${LOCK_TIMEOUT_MIN:-30} -TB=${TRAIN_BUDGET_MIN:-$(python3 "$HERE/perun_policy.py" get train_budget_min)} TR=${TRAIN_MAX_BISECT_ROUNDS:-$(python3 "$HERE/perun_policy.py" get train_max_bisect_rounds)} T0=$SECONDS UN=${TRAIN_BUDGET_UNIT_SECS:-60} # UN: seconds per budget minute (a test seam) +TB=${TRAIN_BUDGET_MIN:-$(python3 "$HERE/perun_policy.py" get train_budget_min)} TR=${TRAIN_MAX_BISECT_ROUNDS:-$(python3 "$HERE/perun_policy.py" get train_max_bisect_rounds)} T0=$SECONDS START=$(date +%s) UN=${TRAIN_BUDGET_UNIT_SECS:-60} # UN: seconds per budget minute (a test seam) [[ $TB =~ ^[0-9]+$ && $TR =~ ^[0-9]+$ ]] || { echo "train_land: TRAIN_BUDGET_MIN / TRAIN_MAX_BISECT_ROUNDS must be integers >= 0" >&2; exit 2; } lock_take "$LOCK" 0 "$LT" || exit 1 # Exclusive machine-wide heavy lease while the gate runs: heavy_gate.py denies other heavy commands ("gate running: wait"). @@ -144,7 +144,7 @@ unl() { local n; for n in $ALL; do [[ " $landed " == *" $n "* ]] || printf '%s ' split_report() { [ "$split" -eq 1 ] || return 0; local u; u=$(unl); echo "SPLIT done: landed #${landed# }; CANT (not landed) #${u% }; gates=$gates"; } stage() { { [ "${TRAIN_STAGES:-1}" != 0 ] && [ $# -ge 3 ]; } || return 0; python3 "$HERE/merge_train.py" stage --base "$BASE_BRANCH" --stage "$1" ${2:+--eta "$2"} ${@:3} &2 || true; } # stage NAME ETA_SECS PR... stage_end() { case $1 in 0) stage done "" $landed; stage skipped "" $(unl) ;; 4|5) stage deferred "" $(unl) ;; *) stage failed "" $(unl) ;; esac; } # EXIT: landed = done; rc 0 with PRs unlanded (SKIP/REFUSE) = skipped; DEFER = deferred; else failed -trap 'stage_end $?; kill "$HB" 2>/dev/null; rm -f "$TRAIN_LAND_COPY"; [ -z "$OWNP" ] || rm -f "$LOG.pid"; [ "$gates" -eq 0 ] || python3 "$HERE/train_yield.py" record "$T" "$ALL" "$landed" "$([ -e "$LOG.red" ] && echo red || { [ ! -e "$LOG.flaky" ] || cat "$LOG.flaky"; })" "$HEADS" || true; rm -f "$LOG.red" "$LOG.flaky"; split_report; clean_union_dirs; lock_drop "$PQ" || true; python3 "$HERE/perun_policy.py" heavy-exclusive-release >/dev/null 2>&1 || true; lock_drop "$LOCK" || true' EXIT # a loud foreign-claim drop must not abort the rest of the cleanup +trap 'stage_end $?; kill "$HB" 2>/dev/null; rm -f "$TRAIN_LAND_COPY"; [ -z "$OWNP" ] || rm -f "$LOG.pid"; [ "$gates" -eq 0 ] || python3 "$HERE/train_yield.py" record "$T" "$ALL" "$landed" "$([ -e "$LOG.red" ] && echo red || { [ ! -e "$LOG.flaky" ] || cat "$LOG.flaky"; })" "$HEADS" "$START" || true; rm -f "$LOG.red" "$LOG.flaky"; split_report; clean_union_dirs; lock_drop "$PQ" || true; python3 "$HERE/perun_policy.py" heavy-exclusive-release >/dev/null 2>&1 || true; lock_drop "$LOCK" || true' EXIT # a loud foreign-claim drop must not abort the rest of the cleanup # A landing train holds the push lock and is moving the base: wait it out before any gate work (a union composed mid-land is stale at its own land). lock_take "$PQ" "$PW" "$LT" || exit 1; lock_drop "$PQ" if [ -n "${CLEAN_ROOT:-}" ]; then # start-of-train cleanup of your own finished work; never blocks the train @@ -163,10 +163,10 @@ git checkout -q --detach "$REMOTE/$BASE_BRANCH" # HEAD (train_yield.py solo; an older head or no recorded head does not count) lands alone, no union, but only when every check on that exact head passed # (landing_watchdog._check_ok, at least one check) and the merge is pinned to that SHA. A PR whose checks are not all green stays in the train, as before. ci_head() { "$GH" pr view "$1" --json headRefOid,statusCheckRollup /dev/null | python3 -c 'import json,sys; sys.path.insert(0,sys.argv[1]) -from landing_watchdog import _check_ok -d=json.load(sys.stdin); r=d["statusCheckRollup"] +from landing_watchdog import _check_ok, _stage +d=json.load(sys.stdin); r=[x for x in d["statusCheckRollup"] or [] if not _stage(x)] if r and all(map(_check_ok,r)): print(d["headRefOid"]) -else: sys.exit(1)' "$HERE"; } # prints the head sha iff it has checks and all passed; a lookup or parse failure is not green +else: sys.exit(1)' "$HERE"; } # prints the head sha iff it has checks (the train's own perun/train-stage marker is not one) and all passed; a lookup or parse failure is not green cand=$(python3 "$HERE/train_yield.py" solo $HEADS 2>/dev/null) || cand= sol= pins= for n in $cand; do # at most ONE solo landing per train: the first eligible PR; the others stay in the train and are re-tested on the new base by the next one diff --git a/.claude/skills/agentic-delivery/scripts/train_yield.py b/.claude/skills/agentic-delivery/scripts/train_yield.py index a65e0463b..cf18cc047 100644 --- a/.claude/skills/agentic-delivery/scripts/train_yield.py +++ b/.claude/skills/agentic-delivery/scripts/train_yield.py @@ -5,12 +5,14 @@ PRs which stayed open; 3 landed (about 15%), and one PR rode 22 trains. That points at infra (a timebox, host load), not at the PR, and only the yield and the re-ride count show it. - train_yield.py record TAG "PR PR ..." "LANDED_PR ..." [OUTCOME ["PR:SHA ..."]] append one started train (train_land.sh calls this on exit) + train_yield.py record TAG "PR PR ..." "LANDED_PR ..." [OUTCOME ["PR:SHA ..." [START]]] append one started train (train_land.sh calls this on exit) train_yield.py report print yield and the PRs that re-rode, then the alarm if any train_yield.py solo PR:SHA... print the PRs that rode >= TRAIN_SOLO_AFTER (default 3, 0 = off) culprit-less ("unpinned"/"flaky") trains at that same head SHA -Ledger: env PERUN_TRAIN_LEDGER, else /train-yield.jsonl, one {"tag","prs","landed","outcome","heads"} per started train -("heads" = the head SHA each PR rode with; a record without it never counts toward `solo`). +Ledger: env PERUN_TRAIN_LEDGER, else /train-yield.jsonl, one {"tag","prs","landed","outcome","at","start","heads"} per started train +("at" = the epoch second it was recorded, which is the train's end; "start" = the epoch second the train began, when the caller passes START (train_land.sh does; +a START that is not 1 to 10 ASCII digits is ignored); landing_watchdog.py compares a recorded decision with "start", else with "at" for a record without one; +"heads" = the head SHA each PR rode with, and a record without "heads" never counts toward `solo`). Only the last WINDOW records count; a record with outcome "red" (BASE RED or a CULPRIT: the base or a PR failed, not infra) is skipped. A record with outcome "unpinned" (every PR passed the failing ids alone, no culprit) or "flaky" (no REAL id, nothing blamed) still counts as a ride and is what `solo` counts; the "may be an intermittent regression" CANT is ledgered as neither (outcome ""). yield = trains with at least one PR landed / trains started. A PR's re-rides = trains it rode minus one, for PRs that @@ -20,8 +22,10 @@ """ import json import os +import re import subprocess import sys +import time WINDOW, YIELD_FLOOR, RERIDE_AT, MIN_TRAINS, SOLO_AT = 50, 0.5, 5, 4, 3 SOLO_OUTCOMES = ("flaky", "unpinned") # culprit-less ride outcomes that `solo` counts @@ -36,11 +40,14 @@ def ledger_path(cwd=None): return os.path.join(top.stdout.strip(), "train-yield.jsonl") -def record(path, tag, prs, landed, outcome="", heads=None): - """Append one started train. `heads` ({pr: head sha} each PR rode with) is stored when given.""" - r = {"tag": tag, "prs": [str(p) for p in prs], "landed": [str(p) for p in landed], "outcome": outcome} +def record(path, tag, prs, landed, outcome="", heads=None, start=None): + """Append one started train, stamped with "at" (int epoch seconds, now: the train's end). `heads` ({pr: head sha} each PR rode with) and `start` (the epoch + second the train began) are stored when given.""" + r = {"tag": tag, "prs": [str(p) for p in prs], "landed": [str(p) for p in landed], "outcome": outcome, "at": int(time.time())} if heads: r["heads"] = {str(p): h for p, h in heads.items()} + if start is not None: + r["start"] = int(start) with open(path, "a") as f: f.write(json.dumps(r) + "\n") @@ -73,6 +80,27 @@ def stats(path): return len(recs), sum(bool(landed) for _, landed in recs), {p: n - 1 for p, n in rides.items() if n > 1} +def sizes_since(path, at): + """PR counts (len of "prs") of the trains among the last WINDOW records that began at or after epoch `at`, oldest first, whatever their outcome. + + A train began at its "start", else (a record written without one) at its "at", the time it was recorded, which is its end. A record with neither usable + never counts as after `at`; a missing or unreadable ledger or line gives nothing for it (never raises).""" + try: + with open(path) as f: + lines = f.read().splitlines()[-WINDOW:] + except OSError: + return [] + out = [] + for ln in lines: + try: + r = json.loads(ln) + if (r["start"] if "start" in r else r["at"]) >= at: + out.append(len(r["prs"])) + except (ValueError, KeyError, TypeError): + continue + return out + + def solo(path, heads, at=None): """The PRs in `heads` ({pr: current head sha}) that never landed and rode >= `at` trains (default $TRAIN_SOLO_AFTER, else SOLO_AT) that ended on culprit-less reds (outcome "unpinned" or "flaky") WITH THAT SAME HEAD: a ride whose recorded head differs from the PR's current head, or has none, is not counted. @@ -116,8 +144,9 @@ def alarm(path): def main(argv): try: path = ledger_path() - if argv[:1] == ["record"] and len(argv) in (4, 5, 6): - record(path, argv[1], argv[2].split(), argv[3].split(), argv[4] if len(argv) > 4 else "", parse_heads(argv[5].split()) if len(argv) == 6 else None) + if argv[:1] == ["record"] and len(argv) in (4, 5, 6, 7): + start = argv[6] if len(argv) == 7 and re.fullmatch(r"[0-9]{1,10}", argv[6]) else None + record(path, argv[1], argv[2].split(), argv[3].split(), argv[4] if len(argv) > 4 else "", parse_heads(argv[5].split()) if len(argv) > 5 else None, start) return 0 if argv[:1] == ["solo"]: print(" ".join(solo(path, parse_heads(argv[1:])))) diff --git a/changelog.d/watchdog-liveness-decision-effect.md b/changelog.d/watchdog-liveness-decision-effect.md new file mode 100644 index 000000000..0b84e6b4b --- /dev/null +++ b/changelog.d/watchdog-liveness-decision-effect.md @@ -0,0 +1,19 @@ +### Fixed +- `green()` in `.claude/skills/agentic-delivery/scripts/landing_watchdog.py` judged every status in a PR's check rollup, including the train's own progress marker (the `perun/train-stage` status: pending while a run is queued, gating or landing, failure after a skipped or failed run), while `merge_train.py` judges check runs only. A PR left in a dead run, or skipped once, therefore stopped counting as a green PR waiting, and every signal built on that count undercounted it. That one status is now ignored, both when judging the checks and when testing that the PR has any; any other pending status still blocks. +- `ci_head` in `train_land.sh` had the same defect, and it is a pre-existing one, not a regression of this change: it judged the whole rollup including the train's own marker, so after a skipped, failed or deferred train, where the marker is failure or error, the solo re-ride escape (a PR that rode trains without landing is landed alone once its own checks pass) could never see an all-green rollup for exactly the PRs it exists for, and never fired. It now drops the marker before judging, still requires at least one remaining check, and still reads a lookup or parse failure as not green. + +### Added +- Run liveness in `landing_watchdog.py`: the rate alarms say "slow" but could not tell a slow train from a dead one. The probe now also reads the train's progress markers from the open PRs (stage = first word of the status description, time = its `at=` else the status creation time, expected run time = `eta<=Ns` else 60 min; a marker with no stage word or no time is skipped) and the time of the newest commit on the default branch's first-parent line (a landing; a time more than 5 minutes ahead of the clock is ignored). The last train activity is the newest of every marker, whatever its run, and that landing; an outcome beats a live marker at the same second. Two warn-only messages come from it, a stuck line and a clause on the stall line, with the same throttle, cache, hook and `check` CLI, failing open. +- "Train run stuck: run R posted on K PRs X min ago (expected <= Y min), no outcome since. Check the train runner, not the PRs." fires only when the last train activity is itself a queued, gating or landing marker older than 1.5 times its expected time; it names that marker's run and stage and counts the PRs carrying the same run and stage. A run id comes from each `merge_train.py stage` call rather than from a whole train, so markers are not grouped by run: any newer marker or landing supersedes an older live one, which keeps a dead run's marker on a PR no later train re-posted from alarming for ever. +- " No train run in X min: start the train or check its runner." is a clause on the existing stall line (0 merged in the window with green PRs waiting), not a line of its own. It is added when markers exist, the last activity is not a live marker, and it is older than the larger of 1.5 times its expected time and the check window, with X counted from that last activity (a landing counts, so a healthy repo with an old skipped marker on an open PR stays quiet). A repo with no marker, or whose last marker is a fresh outcome, gets the plain stall line. +- A parsed `eta` is capped at 24 hours, so a made-up eta cannot hide a dead run for ever, and an `at=` more than 5 minutes ahead of the clock is not trusted: the status creation time is used instead (the marker is skipped when there is none), so a future stamp cannot mute the alarms. +- `gh pr list --json statusCheckRollup` exports a status's context, state and start time but not its description (checked on gh 2.99), and the stage, run id and time live in the description. So when that list shows a train marker, one extra `gh api graphql` call reads the markers of the open PRs into the default branch; with no marker listed there is no extra call, a failed or unreadable reply reads as no markers (silent), and the call shares the 15-minute throttle. The tests exercise it through a fake `gh`; it has not yet been run against a live repository. +- Decision effect: `landing_watchdog.py expect 'train_size>=N' [--hours H]` records a decision in `/perun-decisions.jsonl` (env `PERUN_DECISIONS` overrides; N and H are whole numbers >= 1, H defaults to 24, anything else exits 2 with usage). The first argument is quoted everywhere it is shown, because an unquoted `>=` is a shell redirect: bash runs the command without it and leaves a stray file named `=N`. While the newest unexpired decision is unmet, that is, at least two trains that began after it carried a median of fewer than N PRs, the hook adds "Decision not in effect: train_size>=N set X h ago; the K trains since carried a median of M PRs. Check that the runner reads it." A met or expired decision, fewer than two trains since, a ledger record without a time, and a missing or unreadable file are silent, so a decision the runner never applied no longer reads as a slow train. +- `train_yield.py record` takes an optional last argument, the epoch second the train began, and stamps each ledger line with `"start"` as well as `"at"` (the time the line was written, at the end of the train). `train_land.sh` takes the start epoch once, next to its elapsed-time clock, and passes it from its exit trap. The new `sizes_since(path, at)` returns the PR counts of the trains that began at or after a time, falling back to `"at"` for a record with no start, so a train that began before a decision and ended after it is not judged against it. A record written before the stamps never counts as after a decision, and the existing readers (`report`, `solo`, the yield alarm) ignore the new keys. +- Known limits, stated in the script's docstring: markers come only from the current heads of open PRs, so a re-push or a close in the middle of a run drops one; the creation-time fallback is the post time, not the run start; any landing supersedes a live marker, so a hand merge, or a landing stage that hangs after its first merge, silences the stuck line (the stall line still fires once nothing merges for the window); `train_size` reads the local ledger only, so trains run on another host are not seen. +- `fanout-host-sizing.md` shows one example each of the stuck line, the stall line with its idle clause, and the decision line, next to the yield and merge-rate alarms. + +### Changed +- Tests: `scripts/test_landing_watchdog.py` covers a pending train marker beside a passing check run (green) and another pending status (not green), the marker parser (eta cap, future stamp, missing creation time), the one-call marker fetch and its fail-open paths, the newest-commit time in the probe, stuck at 1.5 times the expected time plus and minus a minute (with an eta and with the default), supersession by a newer marker of another run and by a landing, a healthy repo with an old skipped marker, the idle clause and where it does not appear, tuples from older callers, the `expect` parser, its quoted form run through a shell, and an unmet, met, one-train, expired, untimed and started-before-the-decision case, through the hook and the CLI. `scripts/test_train_yield.py` covers the `"start"` and `"at"` stamps and `sizes_since`, and `scripts/test-train-solo-port.sh` covers the start stamp and a solo landing beside a failure or pending marker (a marker alone stays not green). + +size-budget-raise: .claude/skills/agentic-delivery/references/fanout-host-sizing.md 49827→50877 three alarm examples (stuck line, stall line with idle clause, decision line) and the quoted expect command diff --git a/scripts/test-train-solo-port.sh b/scripts/test-train-solo-port.sh index 99c478992..c9fda25d2 100644 --- a/scripts/test-train-solo-port.sh +++ b/scripts/test-train-solo-port.sh @@ -102,6 +102,8 @@ seed() { rm -f "$WORK/ledger.jsonl"; for i in 1 2 3; do python3 "$SC/train_yield out=$(run t1 1 2 3); rc=$? [ $rc -eq 3 ] && grep -q '^CANT: flake t1 never fails on the base' <<<"$out" && [ "$(last '["outcome"]')" = "" ] && [ "$(last '["heads"]["2"]')" = "$H2" ] ok $? "the intermittent-regression CANT is not ledgered flaky; the ride records each PR's head sha" +[ "$(last '["start"]')" -gt 1000000000 ] && [ "$(last '["start"]')" -le "$(last '["at"]')" ] +ok $? "the ride records the epoch second the train started, no later than its end stamp" out=$(RR=rerun2.sh TRAIN_SPLIT_DEPTH=0 run t2 1 2 3); rc=$? [ $rc -eq 3 ] && grep -q 'CANT: no culprit pinned for the real failures' <<<"$out" && [ "$(last '["outcome"]')" = unpinned ] ok $? "no culprit (every PR passes the failing ids alone) is ledgered unpinned" @@ -131,11 +133,18 @@ grep -q '1 PR(s) left (not landed): #2' <<<"$out" && [ "$(cat "$WORK/merged.log" ok $? "one PR left after the solo landing is reported as left (not landed), not silently skipped" for roll in '[{"__typename":"CheckRun","status":"COMPLETED","conclusion":"FAILURE"}]' '[{"__typename":"CheckRun","status":"IN_PROGRESS","conclusion":""}]' '[]' \ - '[{"__typename":"StatusContext","state":"FAILURE"}]'; do + '[{"__typename":"StatusContext","state":"FAILURE"}]' '[{"__typename":"StatusContext","context":"perun/train-stage","state":"FAILURE"}]'; do echo "$roll" >"$WORK/roll1"; seed "1" "1:$H1"; out=$(run t7 1 2 3) grep -q '#1 rode >= 3 flaky-only or unpinned trains but its head has no all-green checks' <<<"$out" && ! grep -q 'merged 1' "$WORK/merged.log" && [ -s "$WORK/verified" ] ok $? "solo candidate with checks $roll is not landed alone and stays in the train" done; rm -f "$WORK/roll1" +# The train's own progress marker (perun/train-stage) is pending or failure on every open PR, so it is not a check: a head whose other checks all passed lands alone. +for st in FAILURE PENDING; do + echo '[{"__typename":"CheckRun","status":"COMPLETED","conclusion":"SUCCESS"},{"__typename":"StatusContext","context":"perun/train-stage","state":"'$st'"}]' >"$WORK/roll1" + seed "1" "1:$H1"; out=$(run t7m 1 2 3) + grep -q 'landing alone' <<<"$out" && [ "$(head -1 "$WORK/merged.log")" = "merged 1" ] && [ -s "$WORK/verified" ] + ok $? "solo candidate whose passing checks sit beside the train's own $st marker still lands alone" +done; rm -f "$WORK/roll1" seed "1" "1:deadbeef"; out=$(run t8 1 2 3) ! grep -q 'landing alone' <<<"$out" && [ ! -s "$WORK/merged.log" ]; ok $? "head drift: rides recorded at an older head do not count toward solo" seed "1"; out=$(run t9 1 2 3) diff --git a/scripts/test_landing_watchdog.py b/scripts/test_landing_watchdog.py index d30b431e0..110c11eeb 100644 --- a/scripts/test_landing_watchdog.py +++ b/scripts/test_landing_watchdog.py @@ -1,12 +1,13 @@ #!/usr/bin/env python3 -"""Tests for landing_watchdog.py: stall, no stall, no green PR, throttle, off mode, fail-open, the policy keys and the -operating-layer wiring (install into a scratch repo, then fire the hook exactly as wired). +"""Tests for landing_watchdog.py: stall, no stall, no green PR, throttle, off mode, fail-open, the policy keys, the train +liveness and decision-effect lines and the operating-layer wiring (install into a scratch repo, then fire the hook exactly as wired). Fixtures are synthetic (a throwaway bare origin plus clone with backdated commits, a fake `gh` script on PATH); the real gh, the real policy file and the real cache are never touched. """ import json import os +import shlex import shutil import subprocess import sys @@ -23,10 +24,12 @@ import landing_watchdog as lw # noqa: E402 import perun_doctor # noqa: E402 import perun_policy as pp # noqa: E402 +import train_yield as ty # noqa: E402 FAKE_GH = """#!/bin/sh echo x >> "$FAKE_GH_COUNT" [ -n "$FAKE_GH_FAIL" ] && { echo "gh: boom" >&2; exit 1; } +[ "$2" = graphql ] && exec cat "$FAKE_GH_GQL" cat "$FAKE_GH_JSON" """ OK = {"__typename": "CheckRun", "status": "COMPLETED", "conclusion": "SUCCESS"} @@ -40,6 +43,27 @@ ] +def stage(state="PENDING"): + """The train's progress marker as `gh pr list --json statusCheckRollup` exports it (checked on gh 2.99.0): context, state, start time and target, no description.""" + return {"__typename": "StatusContext", "context": "perun/train-stage", "state": state, "targetUrl": "", "startedAt": "1970-01-01T00:16:40Z"} + + +def status(desc, **kw): + """The same marker as the GraphQL API returns it, with the description and createdAt (epoch 1000) that `gh pr list` leaves out.""" + return {"__typename": "StatusContext", "context": "perun/train-stage", "state": "PENDING", "description": desc, "createdAt": "1970-01-01T00:16:40Z", **kw} + + +def iso(t): + """An epoch second as the ISO time GitHub puts in createdAt.""" + return time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime(t)) + + +def reply(*prs): + """The GraphQL reply for the markers query: each item is one open PR's context nodes on its last commit (None = a commit with no checks).""" + return json.dumps({"data": {"repository": {"pullRequests": {"nodes": [ + {"commits": {"nodes": [{"commit": {"statusCheckRollup": None if c is None else {"contexts": {"nodes": c}}}}]}} for c in prs]}}}}) + + def git(cwd, *args, **env): subprocess.run(["git", "-C", str(cwd), *args], check=True, capture_output=True, text=True, env={**os.environ, **env}) @@ -48,7 +72,7 @@ class Base(unittest.TestCase): def setUp(self): self.tmp = Path(tempfile.mkdtemp(prefix="lw-")) self.addCleanup(shutil.rmtree, self.tmp, True) - self.bin, self.cache, self.count, self.prs = self.tmp / "bin", self.tmp / "cache", self.tmp / "count", self.tmp / "prs.json" + self.bin, self.cache, self.count, self.prs, self.gql = self.tmp / "bin", self.tmp / "cache", self.tmp / "count", self.tmp / "prs.json", self.tmp / "gql.json" self.bin.mkdir() (self.bin / "gh").write_text(FAKE_GH) (self.bin / "gh").chmod(0o755) @@ -63,7 +87,7 @@ def setUp(self): git(self.repo, "remote", "set-head", "origin", "main") self.set_prs([GREEN]) self.env = {"PATH": f"{self.bin}{os.pathsep}{os.environ['PATH']}", "PERUN_LANDING_DIR": str(self.cache), - "FAKE_GH_COUNT": str(self.count), "FAKE_GH_JSON": str(self.prs)} + "FAKE_GH_COUNT": str(self.count), "FAKE_GH_JSON": str(self.prs), "FAKE_GH_GQL": str(self.gql)} p = mock.patch.dict(os.environ, self.env) p.start() self.addCleanup(p.stop) @@ -82,6 +106,9 @@ def push(self): def set_prs(self, prs): self.prs.write_text(json.dumps(prs)) + def set_stages(self, *prs): + self.gql.write_text(reply(*prs)) + def gh_calls(self): return len(self.count.read_text().split()) if self.count.exists() else 0 @@ -342,5 +369,563 @@ def test_install_wires_the_hook_and_the_installed_hook_fires(self): self.assertEqual(run(PERUN_LANDING_WATCHDOG="off"), "") +class TrainStageIsNotACheck(unittest.TestCase): + """green() ignores the train's own progress marker (pending while a run is live, failure after a skip); any other pending status still blocks.""" + + def pr(self, *rolls): + return {**GREEN, "statusCheckRollup": list(rolls)} + + def test_a_marker_beside_a_passing_check_run_counts_green_in_every_state(self): + for state in ("PENDING", "FAILURE", "ERROR", "SUCCESS"): + with self.subTest(state=state): + self.assertTrue(lw.green(self.pr(OK, stage(state)))) + + def test_a_pending_status_with_another_context_still_blocks(self): + other = {"__typename": "StatusContext", "context": "ci/other", "state": "PENDING"} + self.assertFalse(lw.green(self.pr(OK, other))) + self.assertFalse(lw.green(self.pr(OK, stage(), other))) + + def test_the_marker_alone_is_not_a_check(self): + self.assertFalse(lw.green(self.pr(stage()))) + + def test_only_a_status_context_is_exempt(self): + run = {"__typename": "CheckRun", "name": "perun/train-stage", "status": "IN_PROGRESS", "conclusion": ""} + self.assertFalse(lw.green(self.pr(OK, run))) + + def test_the_constant_is_the_one_merge_train_posts(self): + import merge_train + self.assertEqual(lw.TRAIN_STAGE, merge_train.STAGE_CONTEXT) + + +class TrainMarks(unittest.TestCase): + """_marks: stage, time, eta and run id from the marker's description; what cannot be read is skipped, never guessed.""" + + def test_reads_stage_time_eta_and_run(self): + stages = [status("gating run=ab12cd34 at=1000 eta<=1200s"), + status("landing run=ab12cd34 at=1001"), + status("queued run=ff00ff00"), # no at=: the status createdAt (epoch 1000) + status("queued run=ee11ee11 at=2000", createdAt="1970-01-01T01:00:00Z"), # at= beats createdAt + status("gating at=5")] # no run id + self.assertEqual(lw._marks(stages), [("ab12cd34", "gating", 1000, 1200), ("ab12cd34", "landing", 1001, 0), ("ff00ff00", "queued", 1000, 0), + ("ee11ee11", "queued", 2000, 0), ("?", "gating", 5, 0)]) + + def test_a_marker_with_no_stage_word_or_no_time_is_skipped(self): + def drop(r, key): + return {k: v for k, v in r.items() if k != key} + for r in (status(""), status(" "), drop(status("x"), "description"), drop(status("gating run=x"), "createdAt"), + status("gating run=x", createdAt="yesterday"), status("gating run=x", createdAt=None), status("gating run=x", createdAt=5), + status("gating run=x at=99999999999999999999", createdAt=None)): + with self.subTest(marker=r): + self.assertEqual(lw._marks([r]), []) + + def test_a_hand_made_status_cannot_grow_a_line_or_overflow_a_number(self): + self.assertEqual(lw._marks([status("gating run=" + "a" * 40 + " at=5 eta<=999999999999s")]), [("a" * 16, "gating", 5, 0)]) + + def test_a_made_up_eta_is_capped_at_a_day(self): + for eta, want in (("999999999", 86400), ("86401", 86400), ("86400", 86400), ("100", 100)): + with self.subTest(eta=eta): + self.assertEqual(lw._marks([status(f"gating run=x at=5 eta<={eta}s")]), [("x", "gating", 5, want)]) + + def test_a_time_more_than_5_min_ahead_of_the_clock_falls_back_to_the_status_created_at(self): + now = 1_800_000_000 + self.assertEqual(lw._marks([status(f"gating run=x at={now + 300}")], now=now), [("x", "gating", now + 300, 0)], "5 min ahead is clock skew: trusted") + self.assertEqual(lw._marks([status(f"gating run=x at={now + 301}", createdAt=iso(now - 600))], now=now), [("x", "gating", now - 600, 0)]) + self.assertEqual(lw._marks([status("gating run=x at=9999999999", createdAt=iso(now - 600))], now=now), [("x", "gating", now - 600, 0)]) + nothing = {k: v for k, v in status("gating run=x at=9999999999").items() if k != "createdAt"} + for r in (status("gating run=x at=9999999999", createdAt=None), nothing): + with self.subTest(marker=r): + self.assertEqual(lw._marks([r], now=now), [], "no createdAt to fall back to: skipped, not guessed") + + +class TrainStages(unittest.TestCase): + """_stages: `gh pr list` leaves out a status's description, so one `gh api graphql` call reads the markers, and only while the list shows one.""" + + LISTED = [{"statusCheckRollup": [OK, stage()]}] + + def stages(self, out, prs=None): + with mock.patch.object(lw, "_run", return_value=out) as run: + return lw._stages(self.LISTED if prs is None else prs, "main", "/nonexistent"), run + + def test_no_marker_in_the_list_means_no_call(self): + other = {"__typename": "StatusContext", "context": "ci/other", "state": "PENDING"} + for prs in ([], [GREEN], [{"statusCheckRollup": None}], [{}], [{"statusCheckRollup": [OK, other]}]): + with self.subTest(prs=prs): + got, run = self.stages(reply([status("gating run=x at=5")]), prs) + self.assertEqual(got, []) + run.assert_not_called() + + def test_a_marker_in_the_list_means_one_call_that_returns_only_markers(self): + mine, later = status("gating run=ab12cd34 at=1000"), status("landing run=ab12cd34 at=1001") + other, lookalike = {**status("x"), "context": "ci/other"}, {**status("gating run=x at=5"), "__typename": "CheckRun"} + got, run = self.stages(reply([{"__typename": "CheckRun"}, mine, other], [lookalike], None, [later])) + self.assertEqual(got, [mine, later]) + run.assert_called_once() + args = run.call_args[0][0] + self.assertEqual(args[:4], ["gh", "api", "graphql", "-f"]) + self.assertTrue(args[4].startswith("query=query(")) + self.assertEqual(args[5:], ["-f", "base=main", "-F", "owner={owner}", "-F", "repo={repo}"]) + + def test_the_query_asks_for_the_open_prs_into_the_branch_and_the_text_gh_drops(self): + for want in ("states:OPEN", "baseRefName:$base", "first:100", "commits(last:1)", "...on StatusContext{context,description,createdAt}"): + self.assertIn(want, lw.STAGES_Q) + self.assertEqual(lw.STAGES_Q.count("{"), lw.STAGES_Q.count("}")) + + def test_a_failing_call_or_an_odd_reply_gives_nothing(self): + for out in ("not json", "[]", "{}", '{"data": null}', '{"data": {"repository": null}}', reply(5)): + with self.subTest(out=out): + self.assertEqual(self.stages(out)[0], []) + for exc in (RuntimeError("gh api failed"), subprocess.TimeoutExpired("gh", 30)): + with self.subTest(exc=exc), mock.patch.object(lw, "_run", side_effect=exc): + self.assertEqual(lw._stages(self.LISTED, "main", "/nonexistent"), []) + + +class RunLiveness(unittest.TestCase): + """_runs: (stuck line, idle minutes). The last train activity is the newest marker of any run or a landing; only a live marker that is itself the last activity can + be stuck, and an old outcome or landing is idle time for the stall line.""" + + NOW = 1_800_000_000 + NONE = (None, None) + + def runs(self, marks, landed=None, minutes=60): + return lw._runs(marks, landed, minutes, now=self.NOW) + + def mark(self, stage, age_min, eta=0, run="r1"): + return (run, stage, self.NOW - age_min * 60, eta) + + @staticmethod + def stuck(run, stage, k, x, y): + return f"Train run stuck: run {run} posted {stage} on {k} PRs {x} min ago (expected <= {y} min), no outcome since. Check the train runner, not the PRs." + + def test_stuck_fires_a_minute_past_1_5x_the_default_and_not_a_minute_before(self): + self.assertEqual(lw.LIVE, ("queued", "gating", "landing")) + for st in lw.LIVE: + with self.subTest(stage=st): + self.assertEqual(self.runs([self.mark(st, 91)]), (self.stuck("r1", st, 1, 91, 60), None)) + self.assertEqual(self.runs([self.mark(st, 89)]), self.NONE) + self.assertEqual(self.runs([self.mark("gating", 90)]), self.NONE, "exactly 1.5x is not older than 1.5x") + + def test_stuck_uses_the_eta_the_marker_carries(self): + self.assertEqual(self.runs([self.mark("gating", 31, eta=1200)]), (self.stuck("r1", "gating", 1, 31, 20), None)) + self.assertEqual(self.runs([self.mark("gating", 29, eta=1200)]), self.NONE) + self.assertEqual(self.runs([self.mark("gating", 100, eta=7200)]), self.NONE, "a longer eta moves the line out") + self.assertEqual(self.runs([self.mark("gating", 181, eta=7200)]), (self.stuck("r1", "gating", 1, 181, 120), None)) + self.assertEqual(self.runs([self.mark("gating", 140, eta=3600)]), (self.stuck("r1", "gating", 1, 140, 60), None)) + + def test_an_outcome_is_never_stuck_and_wins_a_tie_in_time(self): + for outcome in ("done", "failed", "skipped", "deferred", "odd"): + with self.subTest(outcome=outcome): + self.assertEqual(self.runs([self.mark(outcome, 500)]), (None, 500)) + for pair in ([self.mark("gating", 200), self.mark("skipped", 200)], [self.mark("skipped", 200), self.mark("gating", 200)]): + self.assertEqual(self.runs(pair), (None, 200), "the outcome is the newer of a tie") + + def test_a_newer_marker_of_any_run_supersedes_a_live_one(self): + # a run id is minted per `merge_train.py stage` call, not per train: one train posts queued, gating and landing under three ids + train = [self.mark("queued", 200, run="10529-3e"), self.mark("gating", 170, run="10574-e7"), self.mark("landing", 140, run="10617-70")] + self.assertEqual(self.runs(train), (self.stuck("10617-70", "landing", 1, 140, 60), None), "the newest marker names the run; the older ones are history") + self.assertEqual(self.runs([self.mark("gating", 300, run="a1"), self.mark("queued", 10, run="b2")]), self.NONE) + self.assertEqual(self.runs([self.mark("gating", 300, run="a1"), self.mark("done", 100, run="b2")]), (None, 100)) + dead = [self.mark("gating", 10 * 24 * 60, eta=10800, run="dead"), self.mark("done", 10, run="new")] + self.assertEqual(self.runs(dead), self.NONE, "a dead run's marker on a PR no later train re-posted is superseded, however old") + + def test_k_counts_the_markers_of_that_run_and_stage(self): + marks = [self.mark("gating", 140), self.mark("gating", 141), self.mark("queued", 150)] + self.assertEqual(self.runs(marks), (self.stuck("r1", "gating", 2, 140, 60), None)) + self.assertEqual(self.runs([self.mark("gating", 140), self.mark("gating", 141, run="r2")])[0], self.stuck("r1", "gating", 1, 140, 60), "another run is not counted") + + def test_a_landing_after_a_live_marker_supersedes_it_and_one_before_does_not(self): + live = [self.mark("gating", 140)] + self.assertEqual(self.runs(live, landed=self.NOW - 141 * 60), (self.stuck("r1", "gating", 1, 140, 60), None)) + self.assertEqual(self.runs(live, landed=self.NOW - 20 * 60), self.NONE, "a landing since: a hand merge or a train, the repo is moving") + self.assertEqual(self.runs(live, landed=self.NOW - 139 * 60), (None, 139.0), "superseded, so not stuck; the landing itself is old enough to be idle time") + self.assertEqual(self.runs(live, landed=self.NOW - 140 * 60), (None, 140.0), "an outcome wins a tie in time") + + def test_a_landing_time_ahead_of_the_clock_is_ignored_so_it_cannot_mute_a_stuck_run(self): + live = [self.mark("gating", 140)] + self.assertEqual(self.runs(live, landed=self.NOW + 300), self.NONE, "5 min ahead is clock skew: trusted, a landing since") + self.assertEqual(self.runs(live, landed=self.NOW + 301), (self.stuck("r1", "gating", 1, 140, 60), None)) + self.assertEqual(self.runs([self.mark("done", 500)], landed=self.NOW + 10 ** 6), (None, 500), "idle time stays measured from the marker") + + def test_idle_is_the_age_of_the_last_activity_past_the_larger_of_1_5x_expected_and_the_check_window(self): + for eta, minutes, limit in ((0, 60, 90), (7200, 60, 180), (0, 120, 120)): + with self.subTest(eta=eta, minutes=minutes): + self.assertEqual(self.runs([self.mark("done", limit + 1, eta=eta)], minutes=minutes), (None, limit + 1)) + self.assertEqual(self.runs([self.mark("done", limit - 1, eta=eta)], minutes=minutes), self.NONE) + + def test_idle_looks_at_the_newest_marker_of_every_run(self): + self.assertEqual(self.runs([self.mark("done", 500, run="a"), self.mark("done", 10, run="b")]), self.NONE) + + def test_a_landing_is_the_last_activity_when_it_is_the_newest(self): + old = [self.mark("skipped", 3 * 24 * 60)] + self.assertEqual(self.runs(old, landed=self.NOW - 10 * 60), self.NONE, "a repo that merges, with a skipped marker from days ago") + self.assertEqual(self.runs(old, landed=self.NOW - 200 * 60), (None, 200), "idle time counts from the landing, not from the old marker") + self.assertEqual(self.runs(old), (None, 3 * 24 * 60), "no landing time known: the marker alone") + + def test_a_stuck_last_marker_has_no_idle_time_and_a_live_one_not_yet_stuck_has_none_either(self): + self.assertEqual(self.runs([self.mark("gating", 140)]), (self.stuck("r1", "gating", 1, 140, 60), None)) + self.assertEqual(self.runs([self.mark("gating", 30)]), self.NONE) + + def test_a_made_up_eta_cannot_hide_a_dead_run_for_ever(self): + marks = lw._marks([status(f"gating run=x at={self.NOW - 37 * 3600} eta<=999999999s")], now=self.NOW) + self.assertEqual(self.runs(marks), (self.stuck("x", "gating", 1, 2220, 1440), None)) + + def test_a_future_time_cannot_mute_the_alarm(self): + now, far = self.NOW, self.NOW + 10 ** 6 + stale = status(f"gating run=b at={now - 200 * 60}") + for future in (status(f"done run=a at={far}", createdAt=iso(now - 300 * 60)), status(f"done run=a at={far}", createdAt=None)): + with self.subTest(future=future): # it falls back to a createdAt older than the stale run, or is skipped: either way the stale run is the last + self.assertEqual(self.runs(lw._marks([future, stale], now=now)), (self.stuck("b", "gating", 1, 200, 60), None)) + lone = status(f"queued run=x at={far}", createdAt=iso(now - 200 * 60)) + self.assertEqual(self.runs(lw._marks([lone], now=now)), (self.stuck("x", "queued", 1, 200, 60), None)) + + def test_no_markers_is_silent(self): + self.assertEqual(self.runs([]), self.NONE) + self.assertEqual(self.runs([], landed=self.NOW - 10 ** 6), self.NONE, "a landing alone is not a train") + + +class RunLinesThroughAll(unittest.TestCase): + """_all with canned probe results (no git, no network): where the stuck line and the idle clause land, and the probe tuple's optional tail.""" + + STALL = "Landing stalled: 0 merged to main in 60 min, 2 green PRs waiting. Fix landing before starting new work." + CLAUSE = " No train run in {} min: start the train or check its runner." + + def setUp(self): + self.now = int(time.time()) + p = mock.patch.multiple(lw, _lkg=lambda c: None, _red=lambda c: None, _installer=lambda c: None, _train_yield=lambda c: None, _decision=lambda c: None, + flow_limits=lambda: (0, 0), age_limits=lambda: (0, 0), lane_cfg=lambda: (0, {})) + p.start() + self.addCleanup(p.stop) + + def all(self, merged, marks=(), landed=None, k=2): + return lw._all("/x", ("main", merged, k, k, [], list(marks), landed), 60, 0, (0, 0), (0, {})) + + def stuck(self, run, k, x): + return f"Train run stuck: run {run} posted gating on {k} PRs {x} min ago (expected <= 60 min), no outcome since. Check the train runner, not the PRs." + + def test_a_healthy_repo_with_an_old_skipped_marker_is_quiet(self): + marks = [("old1", "skipped", self.now - 3 * 86400, 0)] + self.assertIsNone(self.all(5, marks)) + self.assertIsNone(self.all(5, marks, landed=self.now - 10 * 60)) + + def test_the_clause_rides_the_stall_line_and_only_when_the_last_activity_is_an_old_outcome(self): + old = [("old1", "skipped", self.now - 200 * 60, 0)] + self.assertEqual(self.all(0, old), self.STALL + self.CLAUSE.format(200)) + self.assertIsNone(self.all(0, old, k=0), "no green PR waiting: no stall line, so no clause") + self.assertEqual(self.all(0), self.STALL, "no markers: a repo without trains never gets the clause") + self.assertEqual(self.all(0, [("o", "done", self.now - 80 * 60, 0)]), self.STALL, "a fresh outcome (trains run but fail) gets none") + self.assertEqual(self.all(0, old, landed=self.now - 70 * 60), self.STALL, "a landing newer than the marker is fresh enough") + + def test_stuck_and_the_clause_never_fire_together(self): + live = [("r1", "gating", self.now - 140 * 60, 3600)] + self.assertEqual(self.all(0, live).splitlines(), [self.STALL, self.stuck("r1", 1, 140)]) + self.assertEqual(self.all(5, live), self.stuck("r1", 1, 140), "stuck does not need a stall line") + self.assertIsNone(self.all(5, live, landed=self.now - 20 * 60), "a landing since supersedes the live marker") + + def test_the_probe_tuple_may_stop_early(self): + for tail in ((), ([],), ([], []), ([], [], None)): + with self.subTest(length=4 + len(tail)): + self.assertEqual(lw._all("/x", ("main", 0, 2, 2, *tail), 60, 0, (0, 0), (0, {})), self.STALL) + + +class TrainLivenessHook(Base): + """The probe sees the marker in its `gh pr list`, reads the markers in one more call, and the hook adds the line.""" + + STALL = "Landing stalled: 0 merged to main in 60 min, 1 green PRs waiting. Fix landing before starting new work." + CLAUSE = " No train run in {} min: start the train or check its runner." + + def mark(self, desc, state="PENDING"): + """One green PR with the marker: the list shows it without the text, the GraphQL reply carries the text.""" + self.set_prs([{**GREEN, "statusCheckRollup": [OK, stage(state)]}]) + self.set_stages([{"__typename": "CheckRun"}, status(desc, state=state)]) + + def test_probe_returns_the_markers_and_asks_once_more_only_while_one_is_listed(self): + self.mark("gating run=ab12cd34 at=1000 eta<=1200s") + self.assertEqual(lw.probe(str(self.repo), 60)[5], [("ab12cd34", "gating", 1000, 1200)]) + self.assertEqual(self.gh_calls(), 2, "gh pr list and gh api graphql") + self.set_prs([GREEN]) + self.assertEqual(lw.probe(str(self.repo), 60)[5], []) + self.assertEqual(self.gh_calls(), 3, "gh pr list only") + + def test_probe_returns_the_newest_first_parent_commit_time_and_fails_open(self): + def newest(): + return int(subprocess.run(["git", "-C", str(self.repo), "log", "-1", "--first-parent", "--format=%ct", "origin/main"], capture_output=True, text=True, check=True).stdout) + before = newest() + self.assertEqual(lw.probe(str(self.repo), 60)[6], before) + self.commit(minutes_ago=10) + self.push() + self.assertEqual((lw.probe(str(self.repo), 60)[6], newest() > before), (newest(), True)) + real = lw._run + + def no_log(args, cwd, timeout=15): + if args[:3] == ["git", "log", "-1"]: + raise RuntimeError("git log failed") + return real(args, cwd, timeout) + with mock.patch.object(lw, "_run", no_log): + probed = lw.probe(str(self.repo), 60) + self.assertEqual((probed[6], probed[1]), (None, 1), "a failed landing time is None, the rest of the probe stands") + + def test_a_dead_run_is_named_and_its_pr_still_counts_as_waiting(self): + self.mark(f"gating run=ab12cd34 at={int(time.time()) - 150 * 60} eta<=3600s") + self.assertEqual(self.prompt().splitlines(), [self.STALL, "Train run stuck: run ab12cd34 posted gating on 1 PRs 150 min ago " + "(expected <= 60 min), no outcome since. Check the train runner, not the PRs."]) + + def test_a_live_run_and_a_skipped_pr_are_quiet_and_the_skipped_pr_still_waits(self): + self.mark(f"gating run=ab12cd34 at={int(time.time()) - 30 * 60}") + self.assertEqual(self.prompt(), self.STALL) + shutil.rmtree(self.cache) + self.mark(f"skipped run=ab12cd34 at={int(time.time()) - 5 * 60}", state="FAILURE") + self.assertEqual(self.prompt(), self.STALL) + + def test_an_old_outcome_with_a_green_pr_waiting_adds_the_clause_to_the_stall_line(self): + self.mark(f"done run=ab12cd34 at={int(time.time()) - 200 * 60}", state="SUCCESS") + self.assertEqual(self.prompt(), self.STALL + self.CLAUSE.format(200)) + + def test_the_clause_counts_from_the_last_landing_when_that_is_newer_than_every_marker(self): + self.mark(f"skipped run=ab12cd34 at={int(time.time()) - 3 * 24 * 3600}", state="FAILURE") + self.assertEqual(self.prompt(), self.STALL + self.CLAUSE.format(300), "the fixture's last landing is 300 min old, the marker 3 days") + + def test_a_healthy_repo_with_an_old_skipped_marker_is_quiet(self): + self.commit(minutes_ago=10) + self.push() + self.mark(f"skipped run=ab12cd34 at={int(time.time()) - 3 * 24 * 3600}", state="FAILURE") + self.assertIsNone(self.prompt()) + + def test_a_dead_runs_marker_is_superseded_by_a_later_landing(self): + self.commit(minutes_ago=20) + self.push() + self.mark(f"gating run=dead0001 at={int(time.time()) - 10 * 24 * 3600} eta<=10800s") + self.assertIsNone(self.prompt()) + + def test_a_failed_second_call_keeps_the_other_lines(self): + self.set_prs([{**GREEN, "statusCheckRollup": [OK, stage()]}]) # a marker is listed; no reply file, so the fake's `gh api graphql` fails + self.assertEqual(self.prompt(), self.STALL) + self.assertEqual(self.gh_calls(), 2) + + def test_the_second_call_shares_the_throttle(self): + self.mark(f"gating run=ab12cd34 at={int(time.time()) - 150 * 60}") + first = self.prompt() + self.assertIn("Train run stuck", first) + self.assertEqual((self.prompt(), self.gh_calls()), (first, 2)) + + def test_the_cli_check_reports_the_line(self): + self.mark(f"gating run=ab12cd34 at={int(time.time()) - 150 * 60}") + p = subprocess.run([sys.executable, str(SC / "landing_watchdog.py"), "check"], cwd=self.repo, capture_output=True, text=True, env=os.environ) + self.assertEqual(p.returncode, 1, p.stderr) + self.assertIn("Train run stuck: run ab12cd34 posted gating on 1 PRs 150 min ago", p.stdout) + + +class Expect(Base): + """`landing_watchdog.py expect`: the decision file and the exit codes.""" + + def expect(self, *args, **env): + return subprocess.run([sys.executable, str(SC / "landing_watchdog.py"), "expect", *args], cwd=self.repo, capture_output=True, text=True, env={**os.environ, **env}) + + def test_records_a_decision_with_the_default_and_an_explicit_window(self): + f = self.repo / ".git" / "perun-decisions.jsonl" + p = self.expect("train_size>=12") + self.assertEqual((p.returncode, p.stdout, p.stderr), (0, "decision recorded: train_size>=12 for 24 h\n", "")) + p = self.expect("train_size>=3", "--hours", "2") + self.assertEqual((p.returncode, p.stdout, p.stderr), (0, "decision recorded: train_size>=3 for 2 h\n", "")) + a, b = (json.loads(x) for x in f.read_text().splitlines()) + self.assertEqual(sorted(a), ["at", "metric", "op", "until", "value"]) + self.assertEqual((a["metric"], a["op"], a["value"], a["until"] - a["at"]), ("train_size", ">=", 12, 24 * 3600)) + self.assertEqual((b["value"], b["until"] - b["at"]), (3, 2 * 3600)) + self.assertLess(abs(a["at"] - time.time()), 60) + + def test_the_env_names_the_decision_file(self): + f = self.tmp / "elsewhere.jsonl" + self.assertEqual(self.expect("train_size>=5", PERUN_DECISIONS=str(f)).returncode, 0) + self.assertEqual(json.loads(f.read_text())["value"], 5) + self.assertFalse((self.repo / ".git" / "perun-decisions.jsonl").exists()) + + def test_anything_else_exits_2_with_usage_and_writes_nothing(self): + for args in ((), ("train_size>=0",), ("train_size>=-1",), ("train_size>=1.5",), ("train_size>=",), ("train_size>12",), ("train_size=12",), + ("train_size<=12",), ("size>=12",), ("train_size>=12 ",), ("train_size >= 12",), ("train_size>=1234567890",), + ("train_size>=12", "--hours"), ("train_size>=12", "--hours", "0"), ("train_size>=12", "--hours", "x"), ("train_size>=12", "--hours", "-1"), + ("train_size>=12", "--hours", "1.5"), ("train_size>=12", "--days", "2"), ("train_size>=12", "--hours", "2", "x"), + ("--hours", "2", "train_size>=12")): + with self.subTest(args=args): + p = self.expect(*args) + self.assertEqual((p.returncode, p.stdout), (2, "")) + self.assertIn("usage: landing_watchdog.py expect 'train_size>=N' [--hours H]", p.stderr) + self.assertFalse((self.repo / ".git" / "perun-decisions.jsonl").exists()) + + def test_the_documented_quoted_form_works_through_a_shell(self): + """Unquoted, a shell reads `>=12` as a redirect into a file named `=12` and passes only `train_size`.""" + cmd = shlex.join([sys.executable, str(SC / "landing_watchdog.py"), "expect"]) + " 'train_size>=12' --hours 2" + p = subprocess.run(cmd, shell=True, cwd=self.repo, capture_output=True, text=True, env=os.environ) + self.assertEqual((p.returncode, p.stdout, p.stderr), (0, "decision recorded: train_size>=12 for 2 h\n", "")) + self.assertFalse((self.repo / "=12").exists()) + + def test_every_place_that_shows_the_command_quotes_the_argument(self): + shown = {"usage": lw.EXPECT_USAGE, "module docstring": lw.__doc__, "host sizing guide": (ROOT / ".claude/skills/agentic-delivery/references/fanout-host-sizing.md").read_text()} + for name, text in {**shown, "expect docstring": lw.expect.__doc__}.items(): + with self.subTest(doc=name): + self.assertNotRegex(text, r"expect\s+train_size>=") + for name, text in shown.items(): + with self.subTest(doc=name): + self.assertIn("expect 'train_size>=", text) + + def test_an_unwritable_decision_file_exits_2_not_a_traceback(self): + p = self.expect("train_size>=5", PERUN_DECISIONS=str(self.tmp)) # a directory + self.assertEqual((p.returncode, p.stdout), (2, "")) + self.assertTrue(p.stderr.startswith("landing_watchdog: "), p.stderr) + self.assertNotIn("Traceback", p.stderr) + + +class DecisionEffect(unittest.TestCase): + """_decision: the newest unexpired `expect 'train_size>=N'` against the PRs the trains since actually carried.""" + + NOW = 1_800_000_000 + + def setUp(self): + d = Path(tempfile.mkdtemp(prefix="lw-dec-")) + self.addCleanup(shutil.rmtree, d, True) + self.dec, self.led = d / "decisions.jsonl", d / "ledger.jsonl" + p = mock.patch.dict(os.environ, {"PERUN_DECISIONS": str(self.dec), "PERUN_TRAIN_LEDGER": str(self.led)}) + p.start() + self.addCleanup(p.stop) + + def decide(self, n, set_h=3, left_h=21, **kw): + rec = {"metric": "train_size", "op": ">=", "value": n, "at": self.NOW - set_h * 3600, "until": self.NOW + left_h * 3600, **kw} + with open(self.dec, "a") as f: + f.write(json.dumps(rec) + "\n") + + def train(self, size, h_ago=1, **kw): + rec = {"tag": "t", "prs": [str(i) for i in range(size)], "landed": [], "outcome": "", "at": self.NOW - int(h_ago * 3600), **kw} + with open(self.led, "a") as f: + f.write(json.dumps(rec) + "\n") + + def line(self): + return lw._decision("/nonexistent", now=self.NOW) + + def test_unmet_with_two_trains_since_fires(self): + self.decide(12) + self.train(3) + self.train(3) + self.assertEqual(self.line(), "Decision not in effect: train_size>=12 set 3.0 h ago; the 2 trains since carried a median of 3 PRs. " + "Check that the runner reads it.") + + def test_it_is_a_median_not_a_mean_or_a_last(self): + self.decide(12) + for size in (2, 3, 4, 9): + self.train(size) + self.assertIn("the 4 trains since carried a median of 3.5 PRs", self.line()) + + def test_met_is_silent_and_a_median_equal_to_n_is_met(self): + self.decide(12) + for sizes, fires in (((12, 12), False), ((13, 20), False), ((10, 14), False), ((2, 12, 12), False), ((20, 20, 3), False), ((11, 12), True)): + with self.subTest(sizes=sizes): + self.led.write_text("") + for size in sizes: + self.train(size) + self.assertEqual(self.line() is not None, fires) + + def test_no_train_or_one_train_since_is_silent(self): + self.decide(12) + self.assertIsNone(self.line(), "no ledger") + self.train(3) + self.assertIsNone(self.line(), "one train is not a trend") + + def test_an_expired_decision_is_silent(self): + self.decide(12, set_h=30, left_h=-6) + self.train(3, h_ago=10) + self.train(3, h_ago=9) + self.assertIsNone(self.line()) + + def test_records_without_a_time_or_from_before_the_decision_do_not_count(self): + self.decide(12, set_h=3) + self.train(3, h_ago=5) + self.train(3, h_ago=4) + for _ in range(3): + with open(self.led, "a") as f: + f.write(json.dumps({"tag": "t", "prs": ["1"], "landed": [], "outcome": ""}) + "\n") + self.assertIsNone(self.line()) + self.train(3) + self.assertIsNone(self.line(), "one counted train since") + self.train(3) + self.assertIn("the 2 trains since", self.line()) + + def test_a_train_counts_by_when_it_began_not_when_it_ended(self): + self.decide(12, set_h=3) + self.train(3, h_ago=2, start=self.NOW - 4 * 3600) # began before the decision, recorded (ended) after it: the decision never reached it + self.train(12, h_ago=0.5, start=self.NOW - 3600) + self.assertIsNone(self.line(), "one train began since; counted by its end stamp the median would be 7.5 and the line would fire") + self.train(3, h_ago=0.1, start=self.NOW - 1800) + self.assertIn("the 2 trains since carried a median of 7.5 PRs", self.line()) + + def test_a_record_without_a_start_counts_by_its_end_stamp(self): + self.decide(12, set_h=3) + self.train(3, h_ago=2) + self.train(3, h_ago=1) + self.assertIn("the 2 trains since", self.line()) + self.led.write_text("") + self.train(3, h_ago=4) + self.train(3, h_ago=1) + self.assertIsNone(self.line(), "a record from before the decision does not count") + + def test_the_newest_unexpired_decision_is_the_one_checked(self): + self.train(4, h_ago=0.5) + self.train(4, h_ago=0.4) + self.decide(12, set_h=5) + self.assertIn("train_size>=12 set 5.0 h ago", self.line()) + self.decide(3, set_h=1) + self.assertIsNone(self.line(), "the newer decision is met") + + def test_a_newer_expired_decision_is_ignored(self): + self.train(4, h_ago=0.5) + self.train(4, h_ago=0.4) + self.decide(12, set_h=5) + self.decide(3, set_h=1, left_h=-0.5) + self.assertIn("train_size>=12 set 5.0 h ago", self.line()) + + def test_other_metrics_ops_and_bad_lines_are_ignored(self): + self.train(3) + self.train(3) + base = {"value": 12, "at": self.NOW - 3600, "until": self.NOW + 3600} + with open(self.dec, "w") as f: + f.write("not json\n[]\n5\n{}\n") + for bad in ({"metric": "merges", "op": ">="}, {"metric": "train_size", "op": "<="}, {"metric": "train_size", "op": ">=", "value": "12"}, + {"metric": "train_size", "op": ">=", "value": 0}, {"metric": "train_size", "op": ">=", "at": "x"}): + f.write(json.dumps({**base, **bad}) + "\n") + self.assertIsNone(self.line()) + self.decide(12) + self.assertIn("train_size>=12", self.line(), "a valid line after the junk is still found") + + def test_missing_or_unreadable_files_are_silent(self): + self.assertIsNone(self.line(), "neither file exists") + self.train(3) + self.train(3) + self.assertIsNone(self.line(), "no decision file") + os.mkdir(self.dec) + self.assertIsNone(self.line(), "a directory is not a decision file") + + +class DecisionHook(Base): + """End to end: `expect` and the ledger at their default paths (the git dir), then the hook.""" + + def test_the_hook_adds_the_line_while_the_trains_since_are_smaller_than_expected(self): + env = {k: v for k, v in os.environ.items() if k not in ("PERUN_DECISIONS", "PERUN_TRAIN_LEDGER")} + p = subprocess.run([sys.executable, str(SC / "landing_watchdog.py"), "expect", "train_size>=12"], cwd=self.repo, capture_output=True, text=True, env=env) + self.assertEqual(p.returncode, 0, p.stderr) + for tag in ("t1", "t2"): + ty.record(str(self.repo / ".git" / "train-yield.jsonl"), tag, [1, 2, 3], []) + with mock.patch.dict(os.environ, clear=False): + os.environ.pop("PERUN_DECISIONS", None) + os.environ.pop("PERUN_TRAIN_LEDGER", None) + lines = self.prompt().splitlines() + self.assertEqual(lines[-1], "Decision not in effect: train_size>=12 set 0.0 h ago; the 2 trains since carried a median of 3 PRs. Check that the runner reads it.") + + def test_a_train_that_began_before_the_decision_is_not_counted(self): + env = {k: v for k, v in os.environ.items() if k not in ("PERUN_DECISIONS", "PERUN_TRAIN_LEDGER")} + p = subprocess.run([sys.executable, str(SC / "landing_watchdog.py"), "expect", "train_size>=12"], cwd=self.repo, capture_output=True, text=True, env=env) + self.assertEqual(p.returncode, 0, p.stderr) + for tag in ("t1", "t2"): # recorded after the decision, but started a minute before it + ty.record(str(self.repo / ".git" / "train-yield.jsonl"), tag, [1, 2, 3], [], "", None, int(time.time()) - 60) + with mock.patch.dict(os.environ, clear=False): + os.environ.pop("PERUN_DECISIONS", None) + os.environ.pop("PERUN_TRAIN_LEDGER", None) + self.assertNotIn("Decision not in effect", self.prompt()) + + if __name__ == "__main__": unittest.main() diff --git a/scripts/test_train_yield.py b/scripts/test_train_yield.py index c2d42b5a2..f180deb1b 100644 --- a/scripts/test_train_yield.py +++ b/scripts/test_train_yield.py @@ -1,9 +1,11 @@ #!/usr/bin/env python3 -"""Tests for train_yield.py and its landing_watchdog line: yield, re-rides, empty ledger, zero trains.""" +"""Tests for train_yield.py and its landing_watchdog line: yield, re-rides, empty ledger, zero trains, the "at" and "start" stamps and sizes_since.""" +import json import os import subprocess import sys import tempfile +import time import unittest from pathlib import Path @@ -82,6 +84,78 @@ def test_watchdog_line_and_fail_open(self): ty.record(self.p, f"t{i}", [1, 2], []) self.assertIn("0 of 4", lw._train_yield("/nonexistent")) + def test_record_stamps_at_from_the_function_and_the_cli(self): + before = int(time.time()) + ty.record(self.p, "t1", [1, 2], []) + r = subprocess.run([sys.executable, ty.__file__, "record", "t2", "3", ""], env={**os.environ, "PERUN_TRAIN_LEDGER": self.p}, capture_output=True, text=True) + self.assertEqual(r.returncode, 0, r.stderr) + after = int(time.time()) + recs = [json.loads(ln) for ln in Path(self.p).read_text().splitlines()] + self.assertEqual([r["tag"] for r in recs], ["t1", "t2"]) + for r in recs: + self.assertIsInstance(r["at"], int) + self.assertTrue(before <= r["at"] <= after, (before, r["at"], after)) + + def test_sizes_since_counts_the_trains_stamped_at_or_after_whatever_their_outcome(self): + Path(self.p).write_text("\n".join([ + '{"prs": ["1", "2"], "at": 100}', + '{"prs": ["3"], "outcome": "red", "at": 200}', + '{"prs": ["1", "2", "3"]}', # no "at": never after a decision + "not json", + '{"prs": ["4", "5", "6", "7"], "at": 300}', + '{"prs": 5, "at": 400}', + '{"at": 400}', + '{"prs": ["1"], "at": "100"}', + ]) + "\n") + self.assertEqual(ty.sizes_since(self.p, 200), [1, 4]) + self.assertEqual(ty.sizes_since(self.p, 100), [2, 1, 4]) + self.assertEqual(ty.sizes_since(self.p, 301), []) + + def test_sizes_since_is_empty_for_a_missing_or_unreadable_ledger(self): + self.assertEqual(ty.sizes_since(self.p, 0), []) + self.assertEqual(ty.sizes_since(os.path.dirname(self.p), 0), []) + + def test_sizes_since_reads_only_the_window(self): + for i in range(ty.WINDOW + 5): + ty.record(self.p, f"t{i}", [1], []) + self.assertEqual(ty.sizes_since(self.p, 0), [1] * ty.WINDOW) + + def test_record_stores_start_only_when_given(self): + ty.record(self.p, "a", [1], []) + ty.record(self.p, "b", [1], [], "", None, 1234) + ty.record(self.p, "c", [1], [], "flaky", {1: "abc"}, "1235") + a, b, c = [json.loads(ln) for ln in Path(self.p).read_text().splitlines()] + self.assertNotIn("start", a) + self.assertEqual((b["start"], c["start"], c["heads"]), (1234, 1235, {"1": "abc"})) + + def test_cli_record_takes_start_as_the_seventh_argument_and_ignores_a_bad_one(self): + env = {**os.environ, "PERUN_TRAIN_LEDGER": self.p} + + def run(*a): + r = subprocess.run([sys.executable, ty.__file__, "record", "t", "1 2", "", *a], env=env, capture_output=True, text=True) + self.assertEqual((r.returncode, r.stderr), (0, "")) + return json.loads(Path(self.p).read_text().splitlines()[-1]) + + self.assertEqual(run("", "1:abc", "1700000000")["start"], 1700000000) + self.assertEqual(run("", "", "1700000001")["start"], 1700000001) # empty HEADS still lets START through + self.assertNotIn("heads", run("", "", "1700000001")) + self.assertNotIn("start", run("", "1:abc")) # the six-argument form is unchanged + for bad in ("", "abc", "-5", "1.5", "\u0661\u0662\u0663", "9" * 11, "9" * 5000): + self.assertNotIn("start", run("", "1:abc", bad), bad[:12]) + + def test_sizes_since_counts_a_train_by_when_it_began_else_by_its_end_stamp(self): + Path(self.p).write_text("\n".join([ + '{"prs": ["1", "2", "3"], "start": 100, "at": 500}', # began before 200, ended after: not counted + '{"prs": ["1", "2", "3", "4"], "start": 300, "at": 600}', + '{"prs": ["5"], "at": 250}', # recorded before "start" existed: its end stamp stands in + '{"prs": ["6", "7"], "start": 400}', # a start alone is enough + '{"prs": ["8"], "start": "x", "at": 900}', # an unusable start is skipped, not read as its end stamp + '{"prs": ["9"]}', + ]) + "\n") + self.assertEqual(ty.sizes_since(self.p, 200), [4, 1, 2]) + self.assertEqual(ty.sizes_since(self.p, 301), [2]) + self.assertEqual(ty.sizes_since(self.p, 0), [3, 4, 1, 2]) + if __name__ == "__main__": unittest.main()