Landing watchdog: train liveness, decision effect, solo escape past the train marker

.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
 

.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} </dev/null >&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 2>/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

.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 <git-common-dir>/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 <git-common-dir>/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:]))))

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=<epoch>` 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 <stage> 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 `<git-common-dir>/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