REPORT: the 2026-10-25 night — committed fixes, CI by hash, teardown. Docs only.

The live proof found a fresh agent answering waiting before it had watched one minute; a quiet
box's torrent app would be held for nothing.

.claude/rules/unprompted-work.md

@@ -50,6 +50,12 @@ it **first** in the morning note. A decision you cannot write in that shape is o
    on Tester 1, demo-hp and demo-felhom through the normal update path and the box walk's checks are green
    on all three — not at night unless the night brief says so. **Agent releases still stop for the
    operator's vouch.** The golden is baked weekly and vouched by the operator; CC never vouches. **No `--no-verify`.**
+   **A version number is pushed once** — `felhom-controller/controller/build.sh` now refuses a second
+   push of a tag that exists, and a registry it cannot reach is a refusal too. This is a FACT about
+   what the build does, not a new rule: a fix for something already pushed is the next patch version,
+   and a box on a bad image is moved by the normal update path, never by re-tagging. Earned on
+   2026-10-15, when 0.315.0 was pushed three times from three different commits and a cache kept
+   serving the first one (`09-update-architecture.md` §0).
 7. **An enumerated gap becomes a row in the same session — or, if it is small, is fixed in it** (the size rule).
    Prose is not a record.
 8. **Hungarian text is searched with ASCII fragments**, with a positive and a negative control.

CHANGELOG.md

@@ -1,4 +1,101 @@
-## Unreleased (next: v0.164.0; 2026-10-24 attended release run) — the weekly upload line test (R-1247)
+## 2026-10-25 (night, unattended) — a BYO host loses our SSH drop-in in ONE update (R-1285) — committed, NOT released
+
+- **The cause, measured 2026-10-25 on a BYO test host:** the bundle that brings a new `felhom-os-apply` is planned by
+  the OLD one on disk, which did not know D43's appliance-only policy — so a BYO host upgraded from 0.164.0 kept our
+  drop-in until its next bundle. A change inside the wrapper alone cannot fix the run that delivers it.
+- **The fix — the new AGENT asks the new wrapper:** `internal/osupdate/bundle.go` runs a policy pass right after every
+  bundle install (`mode: "bundle-policy"`, host layer, no envelope); the wrapper on disk is the bundle's own by then.
+  `configs/felhom-os-apply` `Bundle.policy_pass`: on a root-owned BYO record it removes the appliance-only files WE
+  wrote (our header), keeps the previous copy beside the bundle's, `sshd -t` then reloads ssh — a failing `sshd -t` puts
+  the file back. It writes nothing, so it needs no signature; an appliance and the owner's own file are left alone. Never
+  fatal for the agent: an older wrapper refuses the unknown mode and the next bundle does it. Report key `bundle_policy`.
+- Tests: `TestR1285_AnInstalledBundleIsFollowedByThePolicyPass` (a refused bundle runs no pass);
+  `ByoKeepsItsSSH.test_the_policy_pass_*` (three). Red-proofs: felhom.eu
+  `documentation/audits/night-2026-10-25/redproof-r1285.txt`.
+
+## 2026-10-25 (night, unattended) — a signed job's window is at most 24 h (R-1280, D44 #22) — committed, NOT released (rides the next agent release; the vouch is the operator's)
+
+- `internal/authz/verifier.go`: `MaxWindow = 24h`; a blob whose `expires_at - issued_at` is longer is refused
+  (`ErrWindowTooLong`) before the expiry and nonce checks — a job signed with a long window by mistake no longer stays
+  usable that long, and the nonce store no longer keeps its record that long.
+- `configs/felhom-os-apply`: the root wrapper holds the same 24 h (`MAX_WINDOW_S`, refusal R3 „the signed job's window
+  is longer than 24 h").
+- `cmd/felhom-opsign`: `-ttl` above 24 h (or ≤ 0) is refused before anything is signed.
+- Tests: `TestVerify_RejectsPerStage/window_longer_than_24_h` and `…/window_of_exactly_24_h_is_accepted`;
+  `test_felhom_os_apply.DockerLane.test_a_window_longer_than_24h_is_refused` / `…_of_24h_is_accepted`. Both refusals
+  seen failing with their check switched off: felhom.eu `documentation/audits/night-2026-10-25/redproof-r1280.txt`.
+
+## 2026-10-25 (evening, attended) — the `published` gate follows D50 = A: the newest 5 plus the pinned MinAgent and day-0 agent (R-1290; no binary change)
+
+- **D50 = A** (operator, 2026-10-25): the registry keeps the newest 5 released agent versions, the hub's MinAgent floor and
+  the day-0 manifest's agent. `scripts/retention-policy.json`: `generic_versions_kept` 10 → 5, new `pinned`
+  (`hub_min_agent` 0.158.1, `day0_agent` 0.165.0 — copies of hub state, because CI holds no hub secret; the session that
+  records a vouch or a new MinAgent updates them in the same commit). `check-published-versions.py` asserts the window
+  plus every pin; a pin with no tag FAILS. CI was red since the operator's hand prune of 0.159.0–0.160.0 (job 2334).
+- **Decoys** (`test_gate_decoys.py`, via `FELHOM_RETENTION_POLICY`, test only): a pinned version older than the window is
+  asserted; its 404 convicts; a pin with no tag convicts. Seen failing on the old gate (two LIVE HOLES): felhom.eu
+  `documentation/audits/night-2026-10-25/redproof-d50.txt`.
+
+## v0.165.0 (2026-10-25) — D43 = A: BYO hosts keep their own SSH settings (R-1257); the line test for a box that seeds all day (R-1250) — released in the attended release run; **VOUCHED by the operator 2026-10-25 18:38:49 CET** (binary `e9d04a47…`, bundle `8dea1f34…`)
+
+**Found at the release (felhom.eu R-1285, P4):** on a BYO host upgraded from 0.164.0 the drop-in is removed only on the bundle run AFTER the one that installs this wrapper — the first run is planned by the old wrapper. Proven on a BYO test host (VM 9371 on demo-hp): the second `agent_config_update` logged `BUNDLE REMOVED … (ours, on a BYO host — D43)`.
+
+### BYO hosts keep their own SSH settings (R-1257, D43 = A) — 2026-10-24 night
+
+- **The root-key-only sshd drop-in (`/etc/ssh/sshd_config.d/10-felhom-root-key-only.conf`, R-982 / D4) goes only to an
+  appliance** — operator ruling D43 = A (2026-10-24): a BYO host belongs to its owner (`11` §1). The table entry's policy
+  is now `appliance`: written only when the ROOT-OWNED install record (`/var/lib/felhom-install/state.json`) says
+  `appliance`. On a `byo` record nothing is written, and a copy WE wrote (this exact path, beginning with our header
+  line) is REMOVED, its previous copy kept for the undo, then `sshd -t` and a reload. A box whose record is absent or
+  agent-writable proves nothing and is left exactly as it is. The bundle report lists `removed`; the record says
+  `removed` / `skipped`. The installer records the mode before it installs the bundle, so a new appliance still gets it.
+- **X3: the „in force" check asks sshd about each of the host's own LAN addresses** (`ip -4 -o addr show scope
+  global`), not only 192.0.2.1 — a `Match Address <LAN>` block that re-allows root passwords no longer reads as in force.
+- **X4: a failing `sshd -t` puts back ONLY the SSH step** (`undo_sshd_step`: the previous copy, or none; sshd is not
+  reloaded, so it never ran the refused state) and the rest of the bundle stays; it used to refuse and undo the whole
+  bundle. The report carries `sshd_step_undone`; the record names what is on disk.
+- Tests (`configs/test_felhom_config_bundle.py`, class `ByoKeepsItsSSH`, 7 tests; each rule seen failing with it taken
+  out): a BYO host gets no drop-in; our copy comes off a BYO host; the owner's own file stays; no record → left as it
+  is; a LAN `Match` block is not in force; every address key-only passes; a failing `sshd -t` undoes only the SSH step.
+  The fake box now carries an appliance record by default (every earlier test unchanged).
+
+### The line test for a box that seeds all day (R-1250) — 2026-10-24
+
+Attended brief 2026-10-24, Part 4 (commit only; the release rides the next normal train). The weekly test (v0.164.0)
+starts only after 10 quiet minutes; a torrent app that seeds all day never gives them, so that box never got a tested
+line and qBittorrent never got its 50 % limit. Now:
+
+- **`linetest.Tester.WaitingForQuiet`** — due, inside the window, tries left, tunnel up, and only the quiet missing.
+  `GET /host/line-test` carries it as `waiting_for_quiet`; a controller with a seeding torrent app acts only on it.
+  It needs the full watched window (11 minute samples): a freshly started agent is not known to be busy, so it never
+  invites a hold (found in the live proof, 2026-10-24: the first build said `true` before one minute was watched;
+  `TestFreshTesterIsNotWaiting`).
+- **`linetest.Tester.RunNow`** and **`POST /host/line-test/run`** (synchronous, about 55 s) — a test on request while the
+  controller holds its app's upload low: every weekly rule holds (due, window, `MaxTriesPerDay`, tunnel, the one heavy-op
+  gate) except the 10 quiet minutes, which become a 5 s settle and a 10 s quiet check (below 2 Mbit/s, so a request
+  never measures the family's own upload); the three runs are 5 s apart instead of 30. A refusal answers its code
+  (`not-due`, `outside-window`, `tries-used`, `no-tunnel`, `heavy-op`, `not-quiet`, `server-busy`); a failed request
+  spends one of the day's tries. `Tick` and `RunNow` now share one `ready()` check.
+- Tests (each seen failing): `TestSeedingBoxIsWaitingForQuietAndRunNowMeasuresIt` (the sleeps, the 60 s bound),
+  `TestRunNowRefusesAnUplinkThatIsStillBusy`, `TestRunNowKeepsTheWeeklyRules`, `TestRunNowFailuresSpendTheDaysTries`,
+  `TestQuietBoxIsNotWaiting`, `TestLineTestRunRoute`; a compile-time assertion that the Tester serves the request half.
+
+**No sudoers or config-bundle change. MinController: none** — an older controller never calls the route.
+
+## 2026-10-24 (review round 2, unattended) — `.claude/rules/unprompted-work.md` aligned with its four siblings
+
+- The copy lacked item 6's paragraph „A version number is pushed once …" that the felhom.eu and felhom-controller copies
+  have carried since 2026-10-15; the file's header says „change all five or none". The paragraph states a fact
+  (`felhom-controller/controller/build.sh` refuses an existing tag and an unreachable registry). Now byte-identical to
+  the other four (md5 b1891bd4). No code change. Record: felhom.eu `documentation/audits/review-round-2-2026-10-24/EDITS.md`.
+
+## v0.164.0 — the weekly upload line test (R-1247) (2026-10-24) — RELEASED on the operator's ruling (attended release run); vouched by the operator 09:44:58 CEST
+
+Binary sha256 `1b803d8b70f8afbe1546f6989f35b6b0e29c9ed63f6acb3843d3c04ef0d24b2b`; config bundle sha256
+`1d7b54d39d0c1aff682ba0c0a7f91aea0f9407a10d2f5a8bbfdc858d32c13341` (configs unchanged — no bundle job sent). Built from
+`f7e1dcb` (CI 2277 green) in a fresh clone by `scripts/release-agent.sh`. Delivered by one signed `agent_update` per box
+(07:25Z): Tester 1, demo-hp and demo-felhom — COMPLETED, self-check 76/76, committed after the 1-minute dwell, the line
+test started (felhom.eu `audits/line-test-2026-10-24/agent-0164-boxes.txt`).
 
 Operator ruling 2026-10-24 07:26: once a week the box's HOST measures its upload line with a 10-second iperf3 upload to an
 iperf3 server on ep0 that answers **only inside the WireGuard tunnel** (felhom.eu `scripts/ep0-iperf3/`; installed the same

CONTEXT.md

@@ -1,5 +1,10 @@
 # CONTEXT — felhom-agent working state
 
+> **2026-10-24 night (unattended) — D43 = A on `main` (`c8d3c17`), NOT released (next v0.165.0, with R-1250).** The sshd
+> drop-in's bundle policy is `appliance` (install record `/var/lib/felhom-install/state.json` mode); BYO: nothing written,
+> our copy removed; no proof: untouched. X3: `lan_addresses()` for `sshd -T`. X4: `undo_sshd_step` — only the SSH file is
+> put back. Release + the operator's vouch: Monday's list (felhom.eu STATUS).
+
 > **2026-10-24 (attended release run) — v0.164.0: the weekly upload line test (R-1247; operator ruling 07:26).**
 > `internal/linetest`: the agent's own iperf3 client (operator's pick — nothing installed on hosts) to ep0's tunnel-only
 > server `10.77.0.1:5201`, Tuesday 10:00-15:59, after 10 quiet minutes, under the heavy-op gate, best of 3; `line_test`

REPORT.md

@@ -1,36 +1,26 @@
-# REPORT — day 2026-10-23: agent 0.163.0 released, delivered to the three test boxes, vouched by the operator
+# REPORT — D50 retention gate, R-1280, R-1285: committed, not released (2026-10-25 night)
 
-**What:** the day brief's Part 2. Design read: `03-host-agent.md` (the tunnel, the bundle), `11` §5.4.2 (the step
+**Plain words first.** Three agent changes are on `main`. None is released. The release (v0.166.0) is Monday's and
-bundle, R-880).
+waits for the operator's vouch (felhom.eu `STATUS.md`, the Monday list).
 
-## Release
+## Each change, before → after
 
-- Code: `1109092` (D4 v2: the sshd drop-in `/etc/ssh/sshd_config.d/10-felhom-root-key-only.conf`,
+| Row | Commit | Before | After | Proof |
-  `PermitRootLogin prohibit-password`, `try-reload-or-restart`, an `sshd -T` read-back; RootKeyOnly 6 tests). CI 2246
+|---|---|---|---|---|
-  success. CHANGELOG `b4721ee` (v0.163.0 header, both sha256), CI 2249 success.
+| R-1290 (D50 = A) | `feaaa0e` | the published-versions gate knew no keep rule | it asserts the newest 5 released versions plus the pinned MinAgent floor (0.158.1) and the day-0 agent (0.165.0) in `scripts/retention-policy.json`; a pin with no tag fails | decoys seen failing; CI 2339 ✓ |
-- `scripts/release-agent.sh 0.163.0`: binary sha256 `561b5316cd511de9fa9d8b8be4553da7add684f715d3bea533ca330ff7f3763c`,
+| R-1280 | `d912307` | a signed job's window could be any length | at most 24 h — the verifier (`ErrWindowTooLong`), the root wrapper (`MAX_WINDOW_S`) and `felhom-opsign -ttl`; exactly 24 h is accepted | red-proof; unit tests; not live |
-  config bundle sha256 `3afb88d9fbbc4adc16083fa52a98cd4974583c6378a77a1a6e0ed8ae6d1b80c8`; tag `v0.163.0`; verified by
+| R-1285 | `abcc1e6` | a BYO host upgraded from an old wrapper kept our SSH drop-in until its NEXT bundle (the old wrapper planned the new bundle) | after every bundle install the agent runs the new wrapper's `bundle-policy` pass; the file is removed in one update; a failing `sshd -t` puts it back | red-proof; live on a BYO test VM: one signed run gave `BUNDLE POLICY REMOVED`; the test package was deleted after (204, then 404) |
-  an independent download.
-- **Vouched by the operator** 07:58:46 UTC — read back from the hub DB (with its WAL) and from the running hub's
-  `/configuration` page: agent 0.163.0, both sha256 as above.
 
-## Delivery — three signed jobs per box (felhom-opsign, key felhom-op-1)
+## CI (by commit hash, `jobs` endpoint)
 
-The D4 drop-in is a NEW bundle path, and the boxes ran bundle 0.160.0 (sha256 `6b0609…`), so: `agent_update 0.163.0`
+`feaaa0e` 2339 ✓ · `d912307` 2344 ✓ · `abcc1e6` 2345 ✓ — all `run_attempt` 1.
-→ `agent_config_update 0.163.0-step1` (built with `scripts/build-step-bundle.py` from 0.160.0: only
-`felhom-os-apply` replaced; sha256 `6cbc0669…`, uploaded and read back) → `agent_config_update 0.163.0`.
 
-| Box | agent | bundle | self-check | swap = RAM (pool guest 9201) | drop-in, `sshd -T` | sshd reloaded | key login | felhom-sshd |
+## Teardown
-|---|---|---|---|---|---|---|---|---|
-| Tester 1 | 0.163.0 | 0.163.0 | ok | 4096 / 4096 | in place, `without-password` | 09:27:38 SIGHUP | ok | active, listening, `sshd -t` ok |
-| demo-hp | 0.163.0 | 0.163.0 | ok | 25898 / 25898 | in place, `without-password` | 09:19:44 SIGHUP | ok | active, listening, `sshd -t` ok |
-| demo-felhom | 0.163.0 | 0.163.0 | ok | 12288 / 12288 | in place, `without-password` | 09:24:42 SIGHUP | ok | active, listening, `sshd -t` ok |
 
-**Not proven, said plainly:** the real root password over SSH was not tried (CC does not hold it); a dummy password is
+The BYO test VM (demo-hp 9372) was uninstalled through the product and destroyed. Its hub host record needs the
-refused, but OpenSSH logs that only as „Failed password", so the policy proof is `sshd -T` + the reload. felhom-sshd
+operator's delete (felhom.eu `STATUS.md`).
-(:8822) admits only the operator peer over the tunnel, so a connection from DooPlex times out by design. The console
-login is not touched by the drop-in (`passwd -S root` = P on all three); not tried at a physical console.
 
-Installer 1.36.0 (the uninstall removes the drop-in) was published the same day (felhom.eu). R-982 and R-1204 closed.
+## Evidence
-Evidence: felhom.eu `documentation/audits/day-2026-10-23/07-agent-0163-boxes.txt`, `10-vouch-readback.txt`,
+
-`11-installer-1360.txt`.
+felhom.eu `documentation/audits/night-2026-10-25/` (`redproof-d50.txt`, `redproof-r1280.txt`, `redproof-r1285.txt`,
+`rig/r1285-live.txt`).

cmd/felhom-agent/linetest_wiring_test.go

@@ -5,8 +5,16 @@ import (
 	"go/parser"
 	"go/token"
 	"testing"
+
+	"gitea.dooplex.hu/admin/felhom-agent/internal/linetest"
+	"gitea.dooplex.hu/admin/felhom-agent/internal/localapi"
 )
 
+// R-1250: SetLineTest takes a LineTestReader and finds the request half by a type assertion, so a Tester that lost
+// WaitingForQuiet or RunNow would compile and silently answer 503 to every seeding box. This line makes it a build
+// failure instead.
+var _ localapi.LineTestRequester = (*linetest.Tester)(nil)
+
 // R-1247: the weekly line test has the fstrim shape (component + reporter seam + goroutine) plus a local-API reader,
 // so its wiring is asserted from the AST. It must share the ONE heavy-op gate (heavyOps) — or it could run beside a
 // backup — and take its server from the tunnel manager, never a literal address.

cmd/felhom-opsign/main.go

@@ -123,6 +123,10 @@ func run() error {
 
 	now := time.Now().UTC()
 	issued := now
+	if *ttl <= 0 || *ttl > authz.MaxWindow {
+		fmt.Fprintf(os.Stderr, "felhom-opsign: -ttl %s refused: a box accepts a window of at most %s (R-1280)\n", *ttl, authz.MaxWindow)
+		os.Exit(2)
+	}
 	expires := now.Add(*ttl)
 
 	// Canonical OpBlob bytes — the EXACT bytes the agent verifier authenticates over.

configs/felhom-os-apply

@@ -168,7 +168,10 @@ KERNEL_ACTIVE = ("staged", "oneshot", "judging", "reverting")
 #         agent-unit → User=felhom-agent; dropin → a [Unit] section (SF-3); nft → nft -c -f; plain → none;
 #         sshd → `sshd -t -f` on the snippet alone (R-982).
 # policy: replace → always written when it differs; if-absent → only when the box has none (a setting the operator may
-#         have tuned); oob → only on a box with the OOB belt (/etc/felhom-sshd exists — the installer's --no-oob leaves none).
+#         have tuned); oob → only on a box with the OOB belt (/etc/felhom-sshd exists — the installer's --no-oob leaves none);
+#         appliance → only on a box whose ROOT-OWNED install record says appliance (D43 = A, operator 2026-10-24: a BYO
+#         host keeps its own settings, `11` §1). On a BYO box nothing is written and a copy WE wrote (this exact path, our
+#         first line) is removed; a box with no readable record is left exactly as it is (it cannot prove which it is).
 # Order matters: the sudoers files come LAST, so a referenced wrapper is in place before the line that allows it.
 # R-861 (agent v0.146.0): the signed agent update, verified here (mode agent_update), then the A/B flip.
 SELFUPDATE_OP = "agent_update"
@@ -184,9 +187,13 @@ BUNDLE_PREV_DIR = "/var/lib/felhom-os-apply/bundle-prev"  # the previous copies
 BUNDLE_KEEP = 3
 BUNDLE_MAX_BYTES = 4 * 1024 * 1024
 BUNDLE_RE = re.compile(r"^bundle-[0-9]+\.[0-9]+\.[0-9]+(-[0-9A-Za-z.]+)?\.json$")
+# R-1280: the longest validity window a signed job may carry — the agent's authz.MaxWindow, the same 24 h.
+MAX_WINDOW_S = 24 * 3600
 OOB_DIR = "/etc/felhom-sshd"
 # R-982: the host's own sshd drop-in (after_install reloads sshd for it; undo reloads after putting it back).
 SSHD_DROPIN = "/etc/ssh/sshd_config.d/10-felhom-root-key-only.conf"
+# D43 = A (R-1257): the first line every copy of the drop-in we ship carries — what proves a file on a BYO host is ours.
+SSHD_DROPIN_HEADER = b"# MANAGED BY FELHOM (the agent's config bundle) \xe2\x80\x94 R-982"
 
 BUNDLE_FILES = [
     ("/usr/local/sbin/felhom-mkfs-guarded", "felhom-mkfs-guarded.sh", 0o755, "bash", "replace"),
@@ -227,7 +234,8 @@ BUNDLE_FILES = [
     ("/etc/systemd/system/felhom-oob-nft.service", "felhom-oob-nft.service", 0o644, "unit", "oob"),
     # R-982, operator ruling D4 (2026-10-22): the host's own sshd (port 22, the LAN): root by key only. The console
     # password stays; felhom-sshd (the operator's door) reads nothing from sshd_config.d. Reloaded in after_install.
-    ("/etc/ssh/sshd_config.d/10-felhom-root-key-only.conf", "felhom-root-ssh-key-only.conf", 0o644, "sshd", "replace"),
+    # D43 = A (operator 2026-10-24, R-1257): appliances only — a BYO host keeps its own SSH settings (`11` §1).
+    ("/etc/ssh/sshd_config.d/10-felhom-root-key-only.conf", "felhom-root-ssh-key-only.conf", 0o644, "sshd", "appliance"),
     ("/etc/sudoers.d/felhom-op", "felhom-op.sudoers", 0o440, "sudoers", "oob"),
     ("/etc/sudoers.d/felhom-agent", "felhom-agent.sudoers", 0o440, "sudoers", "replace"),
 ]
@@ -483,7 +491,8 @@ class Apply:
 
     def check_plan(self, plan):
         mode = plan.get("mode", "apply")
-        if mode not in ("apply", "inventory", "health", "facts", "live-restore-on", "bundle", "agent_update", "oom-check") \
+        if mode not in ("apply", "inventory", "health", "facts", "live-restore-on", "bundle", "bundle-policy", "agent_update",
+                        "oom-check") \
                 + KERNEL_MODES:
             raise Refused("R11", f"unknown mode {mode!r}")
         if mode == "agent_update":
@@ -494,6 +503,10 @@ class Apply:
             if plan.get("layer") != "host":
                 raise Refused("R11", "bundle is a host-layer mode")
             return mode, "host", 0, "bundle"
+        if mode == "bundle-policy":
+            if plan.get("layer") != "host":
+                raise Refused("R11", "bundle-policy is a host-layer mode")
+            return mode, "host", 0, "bundle-policy"
         layer = plan.get("layer")
         if layer not in ("guest", "host", "docker", "pve", "kernel"):
             raise Refused("R12", f"layer {layer!r} is not guest, host, docker, pve or kernel")
@@ -655,6 +668,8 @@ class Apply:
             iss = calendar.timegm(time.strptime(op["issued_at"], "%Y-%m-%dT%H:%M:%SZ"))
         except (KeyError, ValueError, TypeError):
             raise Refused("R3", "the signed job has no readable time window")
+        if exp - iss > MAX_WINDOW_S:
+            raise Refused("R3", "the signed job's window is longer than 24 h (R-1280)")
         if now > exp or now < iss - 120:
             raise Refused("R3", "the signed job is expired or not yet valid")
         nonce = op.get("nonce")
@@ -1064,6 +1079,8 @@ class Apply:
             return self.facts()
         if self.mode == "bundle":
             return Bundle(self).from_plan(plan)
+        if self.mode == "bundle-policy":
+            return Bundle(self).policy_pass()
         if self.mode == "agent_update":
             return self.agent_update(plan)
         if self.layer in ("host", "pve", "kernel"):
@@ -1873,6 +1890,7 @@ class Bundle:
         self.a, self.r = apply, apply.r
         self.report = apply.report
         self.sshd_dropin_not_in_force = False
+        self.sshd_step_undone = None  # X4 (R-1257): why the SSH step alone was put back ("" = not), the rest stays
 
     # ---------- reading and checking ----------
     def load_file(self, path):
@@ -1982,14 +2000,40 @@ class Bundle:
             return b"", None, None  # a symlink or a directory: always replaced by the real file
         return self.r.read_bytes(dest), stat.S_IMODE(st.st_mode), st.st_uid
 
+    def install_mode(self):
+        """The box's ROOT-OWNED install record's mode ("appliance", "byo", …), or None when it proves nothing (absent,
+        agent-writable, unreadable) — the same proof check_appliance asks for the host layer (R12)."""
+        try:
+            st = self.r.stat(INSTALL_STATE)
+            if st.st_uid != 0 or (st.st_mode & 0o022):
+                return None
+            mode = json.loads(self.r.read_file(INSTALL_STATE)).get("mode")
+        except (OSError, ValueError, AttributeError):
+            return None
+        return mode if isinstance(mode, str) and mode else None
+
     def plan_writes(self, files):
-        """Decide per file: write / same / kept (if-absent and present) / skipped (oob on a box without the belt)."""
+        """Decide per file: write / same / kept (if-absent and present) / skipped (oob on a box without the belt; an
+        appliance-only file on a box that is not one) / remove (an appliance-only file WE wrote, on a BYO box)."""
         oob = self.r.isdir(OOB_DIR)
+        install_mode = self.install_mode()
         plan = []
         for dest, content, mode, check, policy in files:
             if policy == "oob" and not oob:
                 plan.append((dest, content, mode, "skipped", None, None))
                 continue
+            if policy == "appliance" and install_mode != "appliance":
+                old, old_mode, _ = self.live(dest)
+                ours = old is not None and old.startswith(SSHD_DROPIN_HEADER)
+                if install_mode == "byo" and ours:
+                    self.r.log(f"os-apply: BUNDLE {dest}: a BYO host keeps its own settings (D43) — removing the copy we wrote")
+                    plan.append((dest, content, mode, "remove", old, old_mode))
+                else:
+                    why = "a BYO host keeps its own settings (D43)" if install_mode == "byo" else \
+                        f"the install record proves no appliance (mode={install_mode!r}) — left as it is"
+                    self.r.log(f"os-apply: BUNDLE {dest} skipped: {why}")
+                    plan.append((dest, content, mode, "skipped", None, None))
+                continue
             old, old_mode, old_uid = self.live(dest)
             if policy == "if-absent" and old is not None:
                 plan.append((dest, content, mode, "kept", old, old_mode))
@@ -2027,11 +2071,47 @@ class Bundle:
         ver, files = self.parse(data, sha)
         return self.install(ver, sha, files, "installer", False)
 
+    def policy_pass(self):
+        """R-1285 (D43 = A): the appliance-only files' policy, applied by THIS wrapper. A bundle that brings a new wrapper
+        is planned by the OLD one, which did not know the policy — so on a BYO host upgraded from 0.164.0 our SSH drop-in
+        stayed until the next bundle. The agent runs this pass right after every bundle install (the wrapper on disk is
+        then the new one). It only REMOVES a file WE wrote (our header) from a box whose root-owned install record says
+        byo, keeps the previous copy beside the bundle's, and reloads sshd; it writes nothing, so it needs no signature.
+        A failing `sshd -t` puts the file back (R-1257 X4's rule). Report key `bundle_policy`."""
+        log = self.r.log
+        install_mode = self.install_mode()
+        removed = []
+        if install_mode == "byo":
+            stamp = time.strftime("%Y%m%dT%H%M%SZ", time.gmtime(self.r.now()))
+            prev = os.path.join(BUNDLE_PREV_DIR, f"{stamp}-policy")
+            for dest, _src, _mode, _check, policy in BUNDLE_FILES:
+                if policy != "appliance":
+                    continue
+                old, old_mode, _ = self.live(dest)
+                if old is None or not old.startswith(SSHD_DROPIN_HEADER):
+                    continue
+                self.r.put_file(prev + dest, old, 0o600)
+                self.r.remove(dest)
+                removed.append((dest, old, old_mode))
+                log(f"os-apply: BUNDLE POLICY REMOVED {dest} (ours, on a BYO host — D43, R-1285)")
+            if any(d == SSHD_DROPIN for d, _, _ in removed):
+                rc, out, err = self.r.host(["/usr/sbin/sshd", "-t"], 120)
+                if rc != 0:
+                    for d, old, old_mode in removed:
+                        self.r.put_file(d, old, old_mode)
+                    log(f"os-apply: BUNDLE POLICY WARNING sshd -t failed without our drop-in (rc={rc}) — put back")
+                    removed = []
+                else:
+                    self.must(["systemctl", "try-reload-or-restart", "ssh"], "systemctl try-reload-or-restart ssh")
+        self.report["bundle_policy"] = {"install_mode": install_mode, "removed": [d for d, _, _ in removed]}
+        log(f"os-apply: BUNDLE POLICY DONE install_mode={install_mode} removed={len(removed)}")
+        return 0
+
     # ---------- install, self-check, undo ----------
     def install(self, ver, sha, files, authority, bootstrap):
         log = self.r.log
         plan = self.plan_writes(files)
-        counts = {k: sum(1 for p in plan if p[3] == k) for k in ("write", "same", "kept", "skipped")}
+        counts = {k: sum(1 for p in plan if p[3] == k) for k in ("write", "same", "kept", "skipped", "remove")}
         log(f"os-apply: BUNDLE START agent={ver} sha={sha[:16]} authority={authority} files={len(plan)} "
             f"write={counts['write']} same={counts['same']} kept={counts['kept']} skipped={counts['skipped']}")
         stamp = time.strftime("%Y%m%dT%H%M%SZ", time.gmtime(self.r.now()))
@@ -2039,10 +2119,17 @@ class Bundle:
         done = []  # (dest, old bytes or None, old mode)
         rep = {"agent_version": ver, "sha256": sha, "authority": authority, "prev_dir": prev,
                "written": [p[0] for p in plan if p[3] == "write"], "kept": [p[0] for p in plan if p[3] == "kept"],
-               "skipped": [p[0] for p in plan if p[3] == "skipped"], "same": counts["same"]}
+               "skipped": [p[0] for p in plan if p[3] == "skipped"], "same": counts["same"],
+               "removed": [p[0] for p in plan if p[3] == "remove"]}
         self.report["bundle"] = rep
         try:
             for dest, content, mode, action, old, old_mode in plan:
+                if action == "remove":  # D43 = A: our copy off a BYO host (the previous copy kept, the undo restores it)
+                    self.r.put_file(prev + dest, old, 0o600)
+                    done.append((dest, old, old_mode))
+                    self.r.remove(dest)
+                    log(f"os-apply: BUNDLE REMOVED {dest} (ours, on a BYO host — D43)")
+                    continue
                 if action != "write":
                     continue
                 if old is not None:
@@ -2062,22 +2149,56 @@ class Bundle:
         rep["signers_created"] = False
         if self.sshd_dropin_not_in_force:
             rep["sshd_dropin_not_in_force"] = True  # R-982 A2: written and reloaded, but sshd -T shows root passwords allowed
+        if self.sshd_step_undone:
+            rep["sshd_step_undone"] = self.sshd_step_undone[:300]  # X4: only the SSH step was put back
         if bootstrap and not self.r.lexists(TRUST_SIGNERS):
             self.r.put_file(TRUST_SIGNERS, pinned_signers_line().encode(), 0o644)
             rep["signers_created"] = True
             log(f"os-apply: BUNDLE created {TRUST_SIGNERS} with exactly the pinned key {PINNED_OPERATOR_KEY_ID}")
         record = {"format": BUNDLE_FORMAT, "agent_version": ver, "bundle_sha256": sha, "authority": authority,
                   "installed_at": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime(self.r.now())),
-                  "files": {p[0]: (sha256_hex(p[1]) if p[3] in ("write", "same") else
+                  "files": {p[0]: self.record_entry(p) for p in plan}}
-                                   (sha256_hex(p[4]) if p[3] == "kept" else "skipped")) for p in plan}}
         self.r.put_file(BUNDLE_RECORD, (json.dumps(record, indent=2, sort_keys=True) + "\n").encode(), 0o644)
         self.prune()
         log(f"os-apply: BUNDLE DONE agent={ver} written={counts['write']} same={counts['same']} "
             f"self-check=ok signers-created={rep['signers_created']}")
         return 0
 
+    def record_entry(self, p):
+        """The record's word for one plan entry: the sha it installed, the sha it kept, or what it did instead."""
+        dest, content, _, action, old, _ = p
+        if dest == SSHD_DROPIN and self.sshd_step_undone:
+            return sha256_hex(old) if old is not None else "absent"  # X4: the SSH step was put back
+        if action in ("write", "same"):
+            return sha256_hex(content)
+        if action == "kept":
+            return sha256_hex(old)
+        return "removed" if action == "remove" else "skipped"
+
+    def lan_addresses(self):
+        """X3 (R-1257): the host's own global IPv4 addresses — the ones a LAN client reaches sshd on."""
+        rc, out, _ = self.r.host(["ip", "-4", "-o", "addr", "show", "scope", "global"], 30)
+        if rc != 0:
+            return []
+        return sorted(set(re.findall(r"\binet (\d+\.\d+\.\d+\.\d+)/", out)))
+
+    def undo_sshd_step(self, plan, why):
+        """X4 (R-1257): `sshd -t` refused the host's whole config with our step in it — put back ONLY that file (the
+        previous copy, or none) and keep the rest of the bundle. sshd was not reloaded, so it never ran the refused state."""
+        for dest, _, _, action, old, old_mode in plan:
+            if dest != SSHD_DROPIN or action not in ("write", "remove"):
+                continue
+            if old is None:
+                self.r.remove(dest)
+            else:
+                self.r.put_file(dest, old, old_mode if old_mode is not None else 0o644)
+        self.sshd_step_undone = why
+        self.r.log(f"os-apply: BUNDLE WARNING sshd -t refused the host's config with the SSH step — ONLY that step is put "
+                   f"back, the rest of the bundle stays: {why[:200]} (R-1257)")
+
     def after_install(self, plan):
         written = {p[0] for p in plan if p[3] == "write"}
+        removed = {p[0] for p in plan if p[3] == "remove"}
         dests = {p[0] for p in plan if p[3] != "skipped"}
         if any(d.startswith("/etc/systemd/system/") for d in written):
             self.must(["systemctl", "daemon-reload"], "systemctl daemon-reload")
@@ -2098,16 +2219,29 @@ class Bundle:
         # felhom-sshd is not touched. Then the POSITIVE check: sshd's own effective view for root. A box whose
         # sshd_config has no Include of sshd_config.d, an earlier drop-in or a Match block can keep the password route
         # — that is said loudly in the log and the report, never silently recorded as done (A2).
-        if SSHD_DROPIN in written:
+        # R-1257 X4: a failing `sshd -t` puts back ONLY the SSH step (undo_sshd_step); it used to refuse and undo the
-            self.must(["/usr/sbin/sshd", "-t"], "sshd -t (the host's whole sshd config)")
+        # whole bundle. X3: the in-force check asks sshd about each of the host's own LAN addresses — a probe from a
+        # documentation address (192.0.2.1) read a `Match Address <LAN>` block that re-allows passwords as in force.
+        if SSHD_DROPIN in written or SSHD_DROPIN in removed:
+            rc, out, err = self.r.host(["/usr/sbin/sshd", "-t"], 120)
+            if rc != 0:
+                self.undo_sshd_step(plan, f"sshd -t failed (rc={rc}): {(out + err).strip()[-200:]}")
+                return
             self.must(["systemctl", "try-reload-or-restart", "ssh"], "systemctl try-reload-or-restart ssh")
-            rc, out, err = self.r.host(["/usr/sbin/sshd", "-T", "-C", "user=root,host=felhom-check,addr=192.0.2.1"], 60)
+        if SSHD_DROPIN in written:
-            eff = re.findall(r"^permitrootlogin\s+(\S+)", out, re.M | re.I)
+            addrs = self.lan_addresses()
-            if rc != 0 or eff != ["without-password"]:
+            if not addrs:
-                self.r.log(f"os-apply: BUNDLE WARNING the root-key-only drop-in is NOT in force: sshd -T says "
+                self.r.log("os-apply: BUNDLE WARNING no global IPv4 address read for the sshd in-force check — "
-                           f"permitrootlogin={eff or '?'} (rc={rc}) — check sshd_config's Include, an earlier drop-in "
+                           "probing 192.0.2.1 only (R-1257)")
-                           f"or a Match block (R-982)")
+                addrs = ["192.0.2.1"]
-                self.sshd_dropin_not_in_force = True
+            for addr in addrs:
+                rc, out, err = self.r.host(["/usr/sbin/sshd", "-T", "-C", f"user=root,host=felhom-check,addr={addr}"], 60)
+                eff = re.findall(r"^permitrootlogin\s+(\S+)", out, re.M | re.I)
+                if rc != 0 or eff != ["without-password"]:
+                    self.r.log(f"os-apply: BUNDLE WARNING the root-key-only drop-in is NOT in force for {addr}: sshd -T "
+                               f"says permitrootlogin={eff or '?'} (rc={rc}) — check sshd_config's Include, an earlier "
+                               f"drop-in or a Match block (R-982, R-1257)")
+                    self.sshd_dropin_not_in_force = True
 
     def must(self, argv, what):
         rc, out, err = self.r.host(argv, 120)

configs/test_felhom_config_bundle.py

@@ -45,8 +45,10 @@ class St:
 class Box:
     """An in-memory host. files: path -> bytes; modes/uids per path; dirs: a set."""
 
-    def __init__(self, bundle_bytes, signed, oob=False, signers=True):
+    def __init__(self, bundle_bytes, signed, oob=False, signers=True, install_mode="appliance"):
         self.files, self.modes, self.uids = {}, {}, {}
+        if install_mode is not None:  # the installer's ROOT-OWNED record (D43 = A reads it for the sshd drop-in)
+            self.put(osapply.INSTALL_STATE, json.dumps({"completed": [], "mode": install_mode}).encode(), 0o644)
         self.dirs = {osapply.OOB_DIR} if oob else set()
         self.put(osapply.TRUST_FILE, json.dumps({"host_id": HOST, "ring0_slow_lane": False}).encode(), 0o644)
         if signers:
@@ -513,6 +515,157 @@ class RootKeyOnly(unittest.TestCase):
         reloads = [i for i, c in enumerate(box.calls) if c == ["systemctl", "try-reload-or-restart", "ssh"]]
         self.assertEqual(len(reloads), 2, box.calls)   # after_install, then the undo
 
+class ByoKeepsItsSSH(unittest.TestCase):
+    """D43 = A (operator 2026-10-24, R-1257): the root-key-only drop-in goes ONLY to a box whose root-owned install record
+    says appliance; on a BYO host nothing is written and a copy WE wrote is removed; a box with no record is left as it
+    is. X3: the in-force check asks sshd about each of the host's own LAN addresses. X4: a failing `sshd -t` puts back
+    only the SSH step. RED-PROOF (2026-10-24, felhom-agent f5c3d77 + these tests): with the drop-in's policy back to
+    `replace`, test_a_byo_host_gets_no_dropin and test_our_copy_comes_off_a_byo_host fail; with the probe back to
+    192.0.2.1 only, test_a_match_block_for_the_lan_is_not_in_force fails; with `sshd -t` back to `must`,
+    test_a_failing_sshd_t_undoes_only_the_ssh_step fails (rc 3, the whole bundle undone)."""
+    DEST = "/etc/ssh/sshd_config.d/10-felhom-root-key-only.conf"
+    OURS = (REPO / "configs" / "felhom-root-ssh-key-only.conf").read_bytes()
+
+    def test_a_byo_host_gets_no_dropin(self):
+        data, sha = real_bundle()
+        box = Box(data, signed_job(sha), install_mode="byo")
+        rc, rep = run(box)
+        self.assertEqual(rc, 0, rep)
+        self.assertNotIn(self.DEST, box.files)
+        self.assertNotIn(["/usr/sbin/sshd", "-t"], box.calls)
+        self.assertNotIn(["systemctl", "try-reload-or-restart", "ssh"], box.calls)
+        self.assertIn(self.DEST, rep["bundle"]["skipped"])
+        rec = json.loads(box.files[osapply.BUNDLE_RECORD])
+        self.assertEqual(rec["files"][self.DEST], "skipped")
+
+    def test_our_copy_comes_off_a_byo_host(self):
+        data, sha = real_bundle()
+        box = Box(data, signed_job(sha), install_mode="byo")
+        box.put(self.DEST, self.OURS, 0o644)   # an earlier bundle wrote it (before D43)
+        rc, rep = run(box)
+        self.assertEqual(rc, 0, rep)
+        self.assertNotIn(self.DEST, box.files)
+        self.assertEqual(rep["bundle"]["removed"], [self.DEST])
+        self.assertTrue(any(k.endswith(self.DEST) and k.startswith(osapply.BUNDLE_PREV_DIR) for k in box.files),
+                        "the previous copy must be kept for the undo")
+        self.assertLess(box.calls.index(["/usr/sbin/sshd", "-t"]), box.calls.index(["systemctl", "try-reload-or-restart", "ssh"]))
+        self.assertEqual(json.loads(box.files[osapply.BUNDLE_RECORD])["files"][self.DEST], "removed")
+
+    def test_the_owners_own_file_stays_on_a_byo_host(self):
+        data, sha = real_bundle()
+        box = Box(data, signed_job(sha), install_mode="byo")
+        box.put(self.DEST, b"# my own\nPermitRootLogin yes\n", 0o644)
+        rc, rep = run(box)
+        self.assertEqual(rc, 0, rep)
+        self.assertEqual(box.files[self.DEST], b"# my own\nPermitRootLogin yes\n")
+        self.assertNotIn(["systemctl", "try-reload-or-restart", "ssh"], box.calls)
+
+    def test_a_box_with_no_record_is_left_as_it_is(self):
+        for setup in ("none", "agent-writable"):
+            data, sha = real_bundle()
+            box = Box(data, signed_job(sha), install_mode=None if setup == "none" else "byo")
+            if setup == "agent-writable":
+                box.modes[osapply.INSTALL_STATE] = 0o666   # proves nothing
+            box.put(self.DEST, self.OURS, 0o644)
+            rc, rep = run(box)
+            self.assertEqual(rc, 0, rep)
+            self.assertEqual(box.files.get(self.DEST), self.OURS, setup)
+            self.assertIn(self.DEST, rep["bundle"]["skipped"], setup)
+
+    def _fake(self, box, addrs=("192.168.0.50",), match=None, sshd_t_rc=0):
+        orig = box.host
+        def host(argv, timeout=600, stdin=None):
+            if argv[:3] == ["ip", "-4", "-o"]:
+                box.calls.append(argv)
+                return 0, "".join(f"2: vmbr0    inet {a}/24 brd x scope global vmbr0\\       valid_lft forever\n" for a in addrs), ""
+            if argv[:2] == ["/usr/sbin/sshd", "-T"]:
+                box.calls.append(argv)
+                val = "yes" if match and f"addr={match}" in argv[-1] else "without-password"
+                return 0, f"port 22\npermitrootlogin {val}\n", ""
+            if argv == ["/usr/sbin/sshd", "-t"]:
+                box.calls.append(argv)
+                return (sshd_t_rc, "", "/etc/ssh/sshd_config line 9: Bad configuration option") if sshd_t_rc else (0, "", "")
+            return orig(argv, timeout, stdin)
+        box.host = host
+
+    # R-1285: the policy pass the agent runs right after every bundle install, with the wrapper NOW on disk. No
+    # signature (it only removes a file we wrote). RED-PROOF: make policy_pass return before the loop → our copy stays.
+    POLICY_PLAN = {"release_id": "bundle-policy-0.166.0", "layer": "host", "mode": "bundle-policy"}
+
+    def test_the_policy_pass_removes_our_copy_from_a_byo_host(self):
+        data, sha = real_bundle()
+        box = Box(data, None, install_mode="byo")
+        box.put(PLAN, json.dumps(self.POLICY_PLAN).encode(), 0o600, uid=999)
+        box.put(self.DEST, self.OURS, 0o644)   # the old wrapper's bundle run left it
+        rc, rep = run(box)
+        self.assertEqual(rc, 0, rep)
+        self.assertNotIn(self.DEST, box.files)
+        self.assertEqual(rep["bundle_policy"], {"install_mode": "byo", "removed": [self.DEST]}, rep)
+        self.assertTrue(any(k.endswith(self.DEST) and k.startswith(osapply.BUNDLE_PREV_DIR) for k in box.files))
+        self.assertLess(box.calls.index(["/usr/sbin/sshd", "-t"]), box.calls.index(["systemctl", "try-reload-or-restart", "ssh"]))
+
+    def test_the_policy_pass_leaves_an_appliance_and_the_owners_file(self):
+        for mode, content in (("appliance", None), ("byo", b"# my own\nPermitRootLogin yes\n")):
+            data, sha = real_bundle()
+            box = Box(data, None, install_mode=mode)
+            box.put(PLAN, json.dumps(self.POLICY_PLAN).encode(), 0o600, uid=999)
+            body = content if content is not None else self.OURS
+            box.put(self.DEST, body, 0o644)
+            rc, rep = run(box)
+            self.assertEqual(rc, 0, rep)
+            self.assertEqual(box.files.get(self.DEST), body, mode)
+            self.assertEqual(rep["bundle_policy"]["removed"], [], mode)
+            self.assertNotIn(["systemctl", "try-reload-or-restart", "ssh"], box.calls, mode)
+
+    def test_the_policy_pass_puts_our_copy_back_when_sshd_t_fails(self):
+        data, sha = real_bundle()
+        box = Box(data, None, install_mode="byo")
+        box.put(PLAN, json.dumps(self.POLICY_PLAN).encode(), 0o600, uid=999)
+        box.put(self.DEST, self.OURS, 0o644)
+        self._fake(box, sshd_t_rc=1)
+        rc, rep = run(box)
+        self.assertEqual(rc, 0, rep)
+        self.assertEqual(box.files.get(self.DEST), self.OURS)
+        self.assertEqual(rep["bundle_policy"]["removed"], [])
+        self.assertNotIn(["systemctl", "try-reload-or-restart", "ssh"], box.calls)
+
+    def test_a_match_block_for_the_lan_is_not_in_force(self):
+        data, sha = real_bundle()
+        box = Box(data, signed_job(sha))
+        self._fake(box, addrs=("192.168.0.50", "10.77.0.4"), match="192.168.0.50")
+        rc, rep = run(box)
+        self.assertEqual(rc, 0, rep)
+        probed = [c[-1] for c in box.calls if c[:2] == ["/usr/sbin/sshd", "-T"]]
+        self.assertTrue(any("addr=192.168.0.50" in p for p in probed) and any("addr=10.77.0.4" in p for p in probed), probed)
+        self.assertFalse(any("addr=192.0.2.1" in p for p in probed), probed)
+        self.assertTrue(rep["bundle"].get("sshd_dropin_not_in_force"), rep)
+        self.assertTrue(any("NOT in force for 192.168.0.50" in l for l in box.logs), box.logs)
+
+    def test_the_lan_check_passes_when_every_address_is_key_only(self):
+        data, sha = real_bundle()
+        box = Box(data, signed_job(sha))
+        self._fake(box, addrs=("192.168.0.50", "10.77.0.4"))
+        rc, rep = run(box)
+        self.assertEqual(rc, 0, rep)
+        self.assertNotIn("sshd_dropin_not_in_force", rep["bundle"])
+
+    def test_a_failing_sshd_t_undoes_only_the_ssh_step(self):
+        for before in (None, b"# MANAGED BY FELHOM (the agent's config bundle) \xe2\x80\x94 R-982 old\nPermitRootLogin prohibit-password\n"):
+            data, sha = real_bundle()
+            box = Box(data, signed_job(sha))
+            if before is not None:
+                box.put(self.DEST, before, 0o644)
+            self._fake(box, sshd_t_rc=255)
+            rc, rep = run(box)
+            self.assertEqual(rc, 0, rep)   # the bundle stays
+            self.assertEqual(box.files.get(self.DEST), before, "only the SSH step is put back")
+            self.assertIn("/usr/local/sbin/felhom-os-apply", box.files, "the rest of the bundle stays")
+            self.assertIn("sshd -t failed", rep["bundle"].get("sshd_step_undone", ""), rep)
+            self.assertNotIn(["systemctl", "try-reload-or-restart", "ssh"], box.calls, "sshd never ran the refused state")
+            want = "absent" if before is None else hashlib.sha256(before).hexdigest()
+            self.assertEqual(json.loads(box.files[osapply.BUNDLE_RECORD])["files"][self.DEST], want)
+
+
 class TrustBootstrap(unittest.TestCase):
     def test_missing_signers_verifies_against_the_pinned_key_and_creates_it(self):
         data, sha = real_bundle()

configs/test_felhom_os_apply.py

@@ -833,6 +833,14 @@ class DockerLane(unittest.TestCase):
     def test_expired_signed_job_is_refused(self):
         self.refused(docker_fake(signed=signed_job(expires="2026-10-04T11:55:00Z")), "R3")
 
+    def test_a_window_longer_than_24h_is_refused(self):
+        # R-1280. RED-PROOF: drop the MAX_WINDOW_S check → the job (current, signed, 24 h + 1 s) runs.
+        self.refused(docker_fake(signed=signed_job(issued="2026-10-03T11:59:59Z", expires="2026-10-04T12:00:00Z")), "R3")
+
+    def test_a_window_of_24h_is_accepted(self):
+        rc, rep = run(docker_fake(signed=signed_job(issued="2026-10-03T12:30:00Z", expires="2026-10-04T12:30:00Z")))
+        self.assertNotEqual((rep.get("refused") or {}).get("code"), "R3", rep)
+
     def test_replayed_signed_job_is_refused(self):
         f = docker_fake(signed=signed_job())
         f.nonces = {"n1": f.clock + 600}

internal/authz/errors.go

@@ -22,6 +22,8 @@ var (
 	ErrExpired = errors.New("authz: op expired")
 	// ErrNotYetValid: now < issued_at (minus clock-skew tolerance).
 	ErrNotYetValid = errors.New("authz: op not yet valid")
+	// ErrWindowTooLong: expires_at - issued_at > MaxWindow (R-1280).
+	ErrWindowTooLong = errors.New("authz: op validity window too long")
 	// ErrReplay: the nonce was already recorded in the window.
 	ErrReplay = errors.New("authz: replay (nonce already seen)")
 )

internal/authz/verifier.go

@@ -20,6 +20,11 @@ const Namespace = "felhom-op-v1"
 // the validity window stays an honest upper bound.
 const DefaultClockSkew = 2 * time.Minute
 
+// MaxWindow is the longest validity window a signed op may carry (R-1280, D44 #22, operator 2026-10-25): a job signed
+// with a long window by mistake would stay usable that long, and the nonce store would keep its record as long. The
+// root wrapper (configs/felhom-os-apply, MAX_WINDOW_S) and felhom-opsign (-ttl) hold the same 24 h.
+const MaxWindow = 24 * time.Hour
+
 // KeyRole tags a pinned operator key (doc 04 §3 two-key model).
 type KeyRole string
 
@@ -150,6 +155,9 @@ func (v *Verifier) Verify(blob, sigArmored []byte) (*VerifiedOp, error) {
 	if now.Before(op.IssuedAt.Add(-v.ClockSkew)) {
 		return nil, fmt.Errorf("%w: issued_at=%s now=%s", ErrNotYetValid, op.IssuedAt, now)
 	}
+	if w := op.ExpiresAt.Sub(op.IssuedAt); w > MaxWindow {
+		return nil, fmt.Errorf("%w: issued_at=%s expires_at=%s (%s, the most is %s)", ErrWindowTooLong, op.IssuedAt, op.ExpiresAt, w, MaxWindow)
+	}
 	if now.After(op.ExpiresAt) {
 		return nil, fmt.Errorf("%w: expires_at=%s now=%s", ErrExpired, op.ExpiresAt, now)
 	}

internal/authz/verify_test.go

@@ -117,6 +117,26 @@ func TestVerify_RejectsPerStage(t *testing.T) {
 		rejects(t, err, ErrTarget)
 	})
 
+	// R-1280: a window longer than 24 h is refused even while it is current; exactly 24 h is accepted.
+	// RED-PROOF: drop the MaxWindow check → the 24 h + 1 s case verifies.
+	t.Run("window longer than 24 h", func(t *testing.T) {
+		blob := canonicalBlob("guest_destroy", "demo-felhom", "9001", "op", "n-win-1", "{}",
+			refNow.Add(-time.Minute), refNow.Add(-time.Minute).Add(MaxWindow+time.Second))
+		sig := mintArmor(t, pub.Marshal(), Namespace, "sha512", blob, sign)
+		v := atRefNow(New(signerSet(pub, "op"), NewMemoryNonceStore(), "demo-felhom"))
+		_, err := v.Verify(blob, sig)
+		rejects(t, err, ErrWindowTooLong)
+	})
+	t.Run("window of exactly 24 h is accepted", func(t *testing.T) {
+		blob := canonicalBlob("guest_destroy", "demo-felhom", "9001", "op", "n-win-2", "{}",
+			refNow.Add(-time.Minute), refNow.Add(-time.Minute).Add(MaxWindow))
+		sig := mintArmor(t, pub.Marshal(), Namespace, "sha512", blob, sign)
+		v := atRefNow(New(signerSet(pub, "op"), NewMemoryNonceStore(), "demo-felhom"))
+		if _, err := v.Verify(blob, sig); err != nil {
+			t.Fatalf("a 24 h window must verify: %v", err)
+		}
+	})
+
 	t.Run("expired", func(t *testing.T) {
 		blob := canonicalBlob("guest_destroy", "demo-felhom", "9001", "op", "n-exp-1", "{}",
 			refNow.Add(-2*time.Hour), refNow.Add(-time.Hour))

internal/linetest/linetest.go

@@ -19,6 +19,10 @@
 //   - No tunnel (no operational wg-felhom with a server address) → no test and no value; the controller then keeps its
 //     passive floor and sets no qBittorrent limit.
 //   - The last result is persisted, so a restart neither loses it nor re-tests.
+//   - A box whose torrent app seeds all day is never quiet for 10 minutes (R-1250). For it the controller holds the
+//     app's upload limit very low and asks for a test on request (RunNow): every rule above except the 10 quiet
+//     minutes still holds, the quiet is checked over RequestQuietWindow instead, and the runs are RequestRunGap
+//     apart, so the whole request ends in about 55 s.
 package linetest
 
 import (
@@ -56,6 +60,11 @@ const (
 	TickInterval       = time.Minute
 	FirstTickDelay     = 5 * time.Minute
 	GateName           = "line-test"
+
+	// A test on request (R-1250): the controller holds the app's limit while it waits, so it is short.
+	RequestSettle      = 5 * time.Second  // the held app's last packets leave before the quiet is read
+	RequestQuietWindow = 10 * time.Second // the uplink must send below QuietMbps over this window
+	RequestRunGap      = 5 * time.Second
 )
 
 // ScheduleText is carried on the host report.
@@ -258,30 +267,52 @@ func (t *Tester) Run(ctx context.Context) {
 	}
 }
 
-// Tick is one look.
+// Refusal is why a test on request did not run (RunNow). Code is stable for the controller.
-func (t *Tester) Tick(ctx context.Context) {
+type Refusal struct {
+	Code string // not-due | outside-window | tries-used | no-tunnel | heavy-op | not-quiet
+	Msg  string
+}
+
+func (r *Refusal) Error() string { return "linetest: " + r.Code + ": " + r.Msg }
+
+// ready applies every rule except the quiet one: due this week, inside the window (the slot on the due day), tries
+// left today, the tunnel up. It answers the server address and the local day.
+func (t *Tester) ready() (addr, day string, refusal *Refusal) {
 	lt := t.now().In(t.loc)
 	t.mu.Lock()
 	st := t.st
-	samples := append([]txSample(nil), t.samples...)
 	t.mu.Unlock()
 	if !Due(st, lt) {
-		return
+		return "", "", &Refusal{"not-due", "measured this week"}
 	}
 	anchor := weekAnchor(lt)
 	anchorDay := lt.YearDay() == anchor.YearDay() && lt.Year() == anchor.Year()
 	if !EligibleAt(lt, t.slot, anchorDay) {
-		return
+		return "", "", &Refusal{"outside-window", "a test starts 10:00-15:59 local, from this host's slot on the due day"}
 	}
-	day := lt.Format("2006-01-02")
+	day = lt.Format("2006-01-02")
 	if st.TriesDay == day && st.TriesToday >= MaxTriesPerDay {
-		return
+		return "", "", &Refusal{"tries-used", fmt.Sprintf("%d tries today", st.TriesToday)}
 	}
 	ip, ok := t.target()
 	if !ok {
-		t.logger.Debug("linetest: no operational tunnel — no test")
+		return "", "", &Refusal{"no-tunnel", "no operational tunnel"}
+	}
+	return netip.AddrPortFrom(ip, ServerPort).String(), day, nil
+}
+
+// Tick is one look.
+func (t *Tester) Tick(ctx context.Context) {
+	addr, day, refusal := t.ready()
+	if refusal != nil {
+		if refusal.Code == "no-tunnel" {
+			t.logger.Debug("linetest: no operational tunnel — no test")
+		}
 		return
 	}
+	t.mu.Lock()
+	samples := append([]txSample(nil), t.samples...)
+	t.mu.Unlock()
 	if q, worst := quiet(samples); !q {
 		t.logger.Debug("linetest: uplink not quiet — waiting", "worst_minute_mbps", fmt.Sprintf("%.1f", worst), "samples", len(samples))
 		return
@@ -292,16 +323,80 @@ func (t *Tester) Tick(ctx context.Context) {
 		return
 	}
 	defer release()
-	t.measure(ctx, netip.AddrPortFrom(ip, ServerPort).String(), day)
+	t.measure(ctx, addr, day, RunGap)
 }
 
-func (t *Tester) measure(ctx context.Context, addr, day string) {
+// WaitingForQuiet reports that a measurement is due and could start now but the uplink has not been quiet for
+// QuietMinutes — the case a seeding torrent app makes permanent (R-1250). The controller reads it from
+// GET /host/line-test and only then holds its app's limit and calls RunNow.
+func (t *Tester) WaitingForQuiet() bool {
+	if _, _, refusal := t.ready(); refusal != nil {
+		return false
+	}
+	t.mu.Lock()
+	samples := append([]txSample(nil), t.samples...)
+	t.mu.Unlock()
+	if len(samples) < QuietMinutes+1 {
+		return false // not yet watched for 10 minutes (a fresh start): not known to be busy, so no hold
+	}
+	q, _ := quiet(samples)
+	return !q
+}
+
+// RunNow is a test on request (POST /host/line-test/run, R-1250): every rule of the weekly test except the 10 quiet
+// minutes, which the caller replaces by holding its own upload down. The uplink must still be quiet over
+// RequestQuietWindow (after RequestSettle), so a request never measures the family's upload against itself. A
+// *Refusal says why nothing ran; any other error is a failed measurement (the same as a failed weekly attempt).
+func (t *Tester) RunNow(ctx context.Context) (mbps float64, runs []float64, err error) {
+	addr, day, refusal := t.ready()
+	if refusal != nil {
+		return 0, nil, refusal
+	}
+	release, busy, ok := t.gate.TryAcquire(GateName)
+	if !ok {
+		return 0, nil, &Refusal{"heavy-op", busy}
+	}
+	defer release()
+	if t.uplink == nil {
+		return 0, nil, &Refusal{"not-quiet", "no uplink counter"}
+	}
+	if !t.sleep(ctx, RequestSettle) {
+		return 0, nil, ctx.Err()
+	}
+	tx0, err0 := t.uplink()
+	at0 := t.now()
+	if !t.sleep(ctx, RequestQuietWindow) {
+		return 0, nil, ctx.Err()
+	}
+	tx1, err1 := t.uplink()
+	if err0 != nil || err1 != nil || tx1 < tx0 {
+		return 0, nil, &Refusal{"not-quiet", "uplink counter unreadable or reset"}
+	}
+	secs := t.now().Sub(at0).Seconds()
+	if secs <= 0 {
+		secs = RequestQuietWindow.Seconds()
+	}
+	if m := float64(tx1-tx0) * 8 / secs / 1e6; m >= QuietMbps {
+		t.logger.Info("linetest: test on request refused — the uplink is not quiet", "mbps", fmt.Sprintf("%.1f", m))
+		return 0, nil, &Refusal{"not-quiet", fmt.Sprintf("the uplink sent %.1f Mbit/s while held", m)}
+	}
+	t.logger.Info("linetest: test on request (a held torrent app, R-1250)", "server", addr)
+	if err := t.measure(ctx, addr, day, RequestRunGap); err != nil {
+		return 0, nil, err
+	}
+	t.mu.Lock()
+	defer t.mu.Unlock()
+	return t.st.Mbps, append([]float64(nil), t.st.RunsMbps...), nil
+}
+
+// measure runs the runs, gap apart, and stores the result. It returns nil only when a value was stored.
+func (t *Tester) measure(ctx context.Context, addr, day string, gap time.Duration) error {
 	start := t.now()
 	var runs []float64
 	var lastErr error
 	busy := false
 	for i := 0; i < RunsPerMeasurement; i++ {
-		if i > 0 && !t.sleep(ctx, RunGap) {
+		if i > 0 && !t.sleep(ctx, gap) {
 			break
 		}
 		rctx, cancel := context.WithTimeout(ctx, RunTimeout)
@@ -356,6 +451,13 @@ func (t *Tester) measure(ctx context.Context, addr, day string) {
 	if err := t.saveLocked(); err != nil {
 		t.logger.Warn("linetest: state save failed — the result will not survive a restart", "path", t.statePath, "err", err)
 	}
+	if len(runs) > 0 {
+		return nil
+	}
+	if lastErr == nil {
+		lastErr = errors.New("linetest: no run succeeded")
+	}
+	return lastErr
 }
 
 // Value is the measured line (ok=false: never measured).

internal/linetest/linetest_test.go

@@ -431,3 +431,144 @@ func TestQuietNeedsTenFullMinutes(t *testing.T) {
 		t.Fatal("a gap in the samples must not read quiet")
 	}
 }
+
+// ── a test on request (R-1250): a torrent app that seeds all day ───────────────────────────────────────────────────
+
+// seedingRig: due and eligible, but every minute of the last ten sent at the line's speed — the weekly Tick never runs.
+func seedingRig(t *testing.T, outs ...runOut) *rig {
+	r := newRig(t, outs...)
+	for i := 0; i < QuietMinutes+1; i++ {
+		r.minute(600_000_000) // 80 Mbit/s: a seed at the line's speed
+	}
+	return r
+}
+
+func TestSeedingBoxIsWaitingForQuietAndRunNowMeasuresIt(t *testing.T) {
+	r := seedingRig(t, runOut{mbps: 88}, runOut{mbps: 92}, runOut{mbps: 90})
+	r.t.Tick(context.Background())
+	if r.run.calls != 0 {
+		t.Fatal("the weekly Tick tested a seeding box")
+	}
+	if !r.t.WaitingForQuiet() {
+		t.Fatal("a due, eligible, seeding box must report waiting_for_quiet")
+	}
+	var slept []time.Duration
+	r.t.sleep = func(_ context.Context, d time.Duration) bool { slept = append(slept, d); return true }
+	mbps, runs, err := r.t.RunNow(context.Background()) // the controller holds the app: the counter stands still
+	if err != nil || mbps != 92 || len(runs) != RunsPerMeasurement || r.run.calls != RunsPerMeasurement {
+		t.Fatalf("RunNow = %.1f %v %v, calls %d; want the best of 3 = 92", mbps, runs, err, r.run.calls)
+	}
+	want := []time.Duration{RequestSettle, RequestQuietWindow, RequestRunGap, RequestRunGap}
+	if len(slept) != len(want) {
+		t.Fatalf("slept %v, want %v", slept, want)
+	}
+	var total time.Duration
+	for i := range want {
+		if slept[i] != want[i] {
+			t.Fatalf("slept %v, want %v", slept, want)
+		}
+		total += slept[i]
+	}
+	if total+RunsPerMeasurement*TestTime > 60*time.Second {
+		t.Fatalf("a request holds the app %v + runs; it must end inside 60 s", total)
+	}
+	if v, _, ok := r.t.Value(); !ok || v != 92 {
+		t.Fatalf("the requested measurement was not stored: %.1f %v", v, ok)
+	}
+	if r.t.WaitingForQuiet() {
+		t.Fatal("measured this week: no longer waiting")
+	}
+	if _, _, err := r.t.RunNow(context.Background()); refusalCode(err) != "not-due" {
+		t.Fatalf("a second request this week: %v, want not-due", err)
+	}
+}
+
+func refusalCode(err error) string {
+	var ref *Refusal
+	if errors.As(err, &ref) {
+		return ref.Code
+	}
+	return ""
+}
+
+// The held app is not the only uploader: a request still refuses to measure a busy uplink.
+func TestRunNowRefusesAnUplinkThatIsStillBusy(t *testing.T) {
+	r := seedingRig(t, runOut{mbps: 90}, runOut{mbps: 90}, runOut{mbps: 90})
+	r.t.uplink = func() (uint64, error) { r.tx += 20_000_000; return r.tx, nil } // 20 MB per 10 s = 16 Mbit/s
+	if _, _, err := r.t.RunNow(context.Background()); refusalCode(err) != "not-quiet" || r.run.calls != 0 {
+		t.Fatalf("RunNow on a busy uplink: %v, calls %d; want not-quiet and no run", err, r.run.calls)
+	}
+}
+
+// Every other rule of the weekly test holds for a request.
+func TestRunNowKeepsTheWeeklyRules(t *testing.T) {
+	r := seedingRig(t)
+	r.gate.busy = "backup:felhom-pbs"
+	if _, _, err := r.t.RunNow(context.Background()); refusalCode(err) != "heavy-op" {
+		t.Fatalf("beside a backup: %v, want heavy-op", err)
+	}
+	r.gate.busy = ""
+	r.tunnel = false
+	if _, _, err := r.t.RunNow(context.Background()); refusalCode(err) != "no-tunnel" {
+		t.Fatalf("no tunnel: %v, want no-tunnel", err)
+	}
+	if r.t.WaitingForQuiet() {
+		t.Fatal("no tunnel: a request could not run, so it must not be invited")
+	}
+	r.tunnel = true
+	r.now = time.Date(2026, 10, 20, 3, 0, 0, 0, time.UTC) // Tuesday 03:00 — the night chain
+	if _, _, err := r.t.RunNow(context.Background()); refusalCode(err) != "outside-window" {
+		t.Fatalf("at night: %v, want outside-window", err)
+	}
+	if r.run.calls != 0 {
+		t.Fatalf("%d runs on refused requests", r.run.calls)
+	}
+}
+
+// A failed request counts as a try, so a controller cannot request more than MaxTriesPerDay tests a day.
+func TestRunNowFailuresSpendTheDaysTries(t *testing.T) {
+	var outs []runOut
+	for i := 0; i < MaxTriesPerDay*RunsPerMeasurement; i++ {
+		outs = append(outs, runOut{err: errors.New("connection refused")})
+	}
+	r := seedingRig(t, outs...)
+	for i := 0; i < MaxTriesPerDay; i++ {
+		if _, _, err := r.t.RunNow(context.Background()); err == nil || refusalCode(err) != "" {
+			t.Fatalf("try %d: %v, want a failed measurement", i+1, err)
+		}
+	}
+	if _, _, err := r.t.RunNow(context.Background()); refusalCode(err) != "tries-used" {
+		t.Fatalf("after %d failures: %v, want tries-used", MaxTriesPerDay, err)
+	}
+}
+
+// A quiet box tests itself on the weekly Tick: it must not invite the controller to hold its app.
+func TestQuietBoxIsNotWaiting(t *testing.T) {
+	r := newRig(t)
+	if r.t.WaitingForQuiet() {
+		t.Fatal("ten quiet minutes: the weekly Tick runs; waiting_for_quiet must be false")
+	}
+}
+
+// Live proof 2026-10-24 (demo-hp): a freshly started tester answered waiting_for_quiet before it had watched a single
+// minute, so a quiet box's torrent app would be held for nothing. Waiting needs the full window, seen busy.
+func TestFreshTesterIsNotWaiting(t *testing.T) {
+	r := &rig{run: &fakeRunner{}, gate: &fakeGate{}, now: time.Date(2026, 10, 20, 12, 0, 0, 0, time.UTC), tunnel: true}
+	r.t = New(r.run, func() (netip.Addr, bool) { return netip.MustParseAddr("10.77.0.1"), r.tunnel }, r.gate,
+		func() (uint64, error) { return r.tx, nil }, "host-x", filepath.Join(t.TempDir(), "line-test.json"), nil)
+	r.t.loc = time.UTC
+	r.t.now = func() time.Time { return r.now }
+	if r.t.WaitingForQuiet() {
+		t.Fatal("no minute watched yet: waiting_for_quiet must be false")
+	}
+	for i := 0; i < QuietMinutes; i++ {
+		r.minute(600_000_000)
+	}
+	if r.t.WaitingForQuiet() {
+		t.Fatal("10 samples are 9 minutes: still not a full window")
+	}
+	r.minute(600_000_000)
+	if !r.t.WaitingForQuiet() {
+		t.Fatal("a full busy window: waiting_for_quiet must be true")
+	}
+}

internal/localapi/linetest.go

@@ -1,8 +1,12 @@
 package localapi
 
 import (
+	"context"
+	"errors"
 	"net/http"
 	"time"
+
+	"gitea.dooplex.hu/admin/felhom-agent/internal/linetest"
 )
 
 // LineTestReader is the R-1247 weekly upload line test (internal/linetest.Tester) as the local API reads it.
@@ -10,6 +14,13 @@ type LineTestReader interface {
 	Value() (mbps float64, at time.Time, ok bool)
 }
 
+// LineTestRequester is the R-1250 half: a due test the uplink never lets start, and a test on request. The Tester
+// satisfies it; a reader without it answers waiting_for_quiet:false and 503 on the POST.
+type LineTestRequester interface {
+	WaitingForQuiet() bool
+	RunNow(ctx context.Context) (mbps float64, runs []float64, err error)
+}
+
 // LineTestResponse is GET /host/line-test. Present=false: never measured, or the job is not wired (no tunnel, disabled,
 // or an agent before the route answers 404 — the controller reads all three as „unknown").
 type LineTestResponse struct {
@@ -17,6 +28,19 @@ type LineTestResponse struct {
 	Mbps       float64 `json:"mbps,omitempty"`
 	MeasuredAt string  `json:"measured_at,omitempty"` // RFC3339 UTC
 	Method     string  `json:"method,omitempty"`      // "iperf3-tunnel": a 10-second upload to ep0 inside the tunnel
+	// WaitingForQuiet: a measurement is due and could start now, but the uplink has not been quiet for 10 minutes
+	// (R-1250). The controller with a seeding torrent app then holds its limit and POSTs /host/line-test/run.
+	WaitingForQuiet bool `json:"waiting_for_quiet,omitempty"`
+}
+
+// LineTestRunResponse is POST /host/line-test/run. OK=false with Refused: nothing ran (Refused is linetest.Refusal's
+// code, or "server-busy"); OK=false without it: the runs failed.
+type LineTestRunResponse struct {
+	OK       bool      `json:"ok"`
+	Mbps     float64   `json:"mbps,omitempty"`
+	RunsMbps []float64 `json:"runs_mbps,omitempty"`
+	Refused  string    `json:"refused,omitempty"`
+	Error    string    `json:"error,omitempty"`
 }
 
 // SetLineTest wires the weekly line test (main.go builds it after the local API server).
@@ -27,11 +51,39 @@ func (s *Server) handleLineTest(w http.ResponseWriter, r *http.Request, vmid int
 		writeOK(w, LineTestResponse{Present: false})
 		return
 	}
+	var waiting bool
+	if q, ok := s.lineTest.(LineTestRequester); ok {
+		waiting = q.WaitingForQuiet()
+	}
 	mbps, at, ok := s.lineTest.Value()
 	if !ok {
-		writeOK(w, LineTestResponse{Present: false})
+		writeOK(w, LineTestResponse{Present: false, WaitingForQuiet: waiting})
 		return
 	}
-	s.logger.Debug("local-api: /host/line-test served", "vmid", vmid, "mbps", mbps, "measured_at", at.UTC().Format(time.RFC3339))
+	s.logger.Debug("local-api: /host/line-test served", "vmid", vmid, "mbps", mbps, "measured_at", at.UTC().Format(time.RFC3339), "waiting_for_quiet", waiting)
-	writeOK(w, LineTestResponse{Present: true, Mbps: mbps, MeasuredAt: at.UTC().Format(time.RFC3339), Method: "iperf3-tunnel"})
+	writeOK(w, LineTestResponse{Present: true, Mbps: mbps, MeasuredAt: at.UTC().Format(time.RFC3339), Method: "iperf3-tunnel", WaitingForQuiet: waiting})
+}
+
+// handleLineTestRun is POST /host/line-test/run (R-1250): a test on request, synchronous (about 55 s). The answer is
+// always 200 with the outcome in the body, except 503 when no line test is wired.
+func (s *Server) handleLineTestRun(w http.ResponseWriter, r *http.Request, vmid int) {
+	q, ok := s.lineTest.(LineTestRequester)
+	if s.lineTest == nil || !ok {
+		writeErr(w, http.StatusServiceUnavailable, "line test not configured")
+		return
+	}
+	s.logger.Info("local-api: line test requested", "vmid", vmid)
+	mbps, runs, err := q.RunNow(r.Context())
+	var ref *linetest.Refusal
+	switch {
+	case err == nil:
+		writeOK(w, LineTestRunResponse{OK: true, Mbps: mbps, RunsMbps: runs})
+	case errors.As(err, &ref):
+		s.logger.Info("local-api: line test request refused", "vmid", vmid, "reason", ref.Code, "detail", ref.Msg)
+		writeOK(w, LineTestRunResponse{Refused: ref.Code, Error: ref.Msg})
+	case errors.Is(err, linetest.ErrBusy):
+		writeOK(w, LineTestRunResponse{Refused: "server-busy", Error: err.Error()})
+	default:
+		writeOK(w, LineTestRunResponse{Error: err.Error()})
+	}
 }

internal/localapi/linetest_test.go

@@ -1,12 +1,16 @@
 package localapi
 
 import (
+	"context"
 	"encoding/json"
+	"errors"
 	"log/slog"
 	"net/http/httptest"
 	"strings"
 	"testing"
 	"time"
+
+	"gitea.dooplex.hu/admin/felhom-agent/internal/linetest"
 )
 
 type fixedLine struct {
@@ -49,3 +53,64 @@ func TestLineTestRoute(t *testing.T) {
 		t.Fatalf("measured: %+v", got)
 	}
 }
+
+type requestableLine struct {
+	fixedLine
+	waiting bool
+	mbps    float64
+	err     error
+	calls   int
+}
+
+func (r *requestableLine) WaitingForQuiet() bool { return r.waiting }
+func (r *requestableLine) RunNow(context.Context) (float64, []float64, error) {
+	r.calls++
+	if r.err != nil {
+		return 0, nil, r.err
+	}
+	return r.mbps, []float64{r.mbps - 1, r.mbps}, nil
+}
+
+func lineTestRun(t *testing.T, s *Server) (int, LineTestRunResponse) {
+	t.Helper()
+	rec := httptest.NewRecorder()
+	s.handleLineTestRun(rec, httptest.NewRequest("POST", "/host/line-test/run", nil), 9201)
+	var env struct {
+		OK   bool                `json:"ok"`
+		Data LineTestRunResponse `json:"data"`
+	}
+	_ = json.Unmarshal(rec.Body.Bytes(), &env)
+	return rec.Code, env.Data
+}
+
+// R-1250: the controller acts only on waiting_for_quiet, and reads a refusal by its code — never as a measurement.
+func TestLineTestRunRoute(t *testing.T) {
+	s := &Server{logger: slog.Default()}
+	if code, _ := lineTestRun(t, s); code != 503 {
+		t.Fatalf("not wired: HTTP %d, want 503", code)
+	}
+	s.SetLineTest(fixedLine{ok: true, mbps: 90})
+	if code, _ := lineTestRun(t, s); code != 503 {
+		t.Fatalf("a reader without the request half: HTTP %d, want 503", code)
+	}
+	l := &requestableLine{fixedLine: fixedLine{ok: false}, waiting: true, mbps: 93.5}
+	s.SetLineTest(l)
+	if got := lineTestBody(t, s); got.Present || !got.WaitingForQuiet {
+		t.Fatalf("GET while waiting: %+v, want present:false waiting_for_quiet:true", got)
+	}
+	if code, got := lineTestRun(t, s); code != 200 || !got.OK || got.Mbps != 93.5 || len(got.RunsMbps) != 2 {
+		t.Fatalf("a measured request: HTTP %d %+v", code, got)
+	}
+	l.err = &linetest.Refusal{Code: "not-quiet", Msg: "the uplink sent 16.0 Mbit/s while held"}
+	if _, got := lineTestRun(t, s); got.OK || got.Refused != "not-quiet" || got.Mbps != 0 {
+		t.Fatalf("a refusal: %+v, want ok:false refused:not-quiet", got)
+	}
+	l.err = linetest.ErrBusy
+	if _, got := lineTestRun(t, s); got.OK || got.Refused != "server-busy" {
+		t.Fatalf("server busy: %+v, want refused:server-busy", got)
+	}
+	l.err = errors.New("dial tcp 10.77.0.1:5201: i/o timeout")
+	if _, got := lineTestRun(t, s); got.OK || got.Refused != "" || got.Error == "" {
+		t.Fatalf("a failed run: %+v, want ok:false, no refusal, the error", got)
+	}
+}

internal/localapi/server.go

@@ -577,6 +577,7 @@ func (s *Server) Handler() http.Handler {
 	// R-1247: the weekly upload line test's last measurement — the controller's qBittorrent limit and upload watch
 	// read it. Read-only; never measured (or the job not wired) = present:false.
 	mux.HandleFunc("GET /host/line-test", s.withGuest(s.handleLineTest))
+	mux.HandleFunc("POST /host/line-test/run", s.withGuest(s.handleLineTestRun)) // R-1250: a test on request
 	// Disk management (slice 8C) — self-scoped; format routes through the data-bearing classifier+gate.
 	mux.HandleFunc("GET /disks", s.withGuest(s.handleDisks))
 	mux.HandleFunc("GET /disks/candidates", s.withGuest(s.handleDiskCandidates))

internal/osupdate/bundle.go

@@ -8,6 +8,7 @@ import (
 	"encoding/json"
 	"fmt"
 	"io"
+	"log/slog"
 	"net/http"
 	"os"
 	"path/filepath"
@@ -114,12 +115,30 @@ func (e ConfigUpdateExecutor) Execute(ctx context.Context, op string, params jso
 		return fmt.Errorf("agent_config_update: failed (previous files restored): %s", wr.Failed)
 	}
 	lg.Warn("osupdate: config bundle INSTALLED", "bundle", string(wr.Bundle), "pass_seconds", wr.PassSeconds)
+	e.policyPass(ctx, lg, p.AgentVersion)
 	if e.AfterInstall != nil {
 		e.AfterInstall(ctx)
 	}
 	return nil
 }
 
+// policyPass (R-1285) asks the wrapper NOW on disk — after this install, the bundle's own — to apply the
+// appliance-only files' policy (mode "bundle-policy"): on a BYO host it removes the SSH drop-in we wrote. The bundle run
+// itself was planned by the wrapper that was on disk BEFORE it, which may predate the policy (0.164.0). Never fatal:
+// the bundle is installed either way; an older wrapper refuses the unknown mode and the next bundle does it.
+func (e ConfigUpdateExecutor) policyPass(ctx context.Context, lg *slog.Logger, ver string) {
+	runID := "bundle-policy-" + e.Leg.now().UTC().Format("20060102T150405Z")
+	wr, err := e.Leg.call(ctx, runID, map[string]any{"release_id": "bundle-policy-" + ver, "layer": LayerHost, "mode": "bundle-policy"})
+	switch {
+	case err != nil:
+		lg.Warn("osupdate: the bundle policy pass did not run", "err", err)
+	case wr.refused():
+		lg.Info("osupdate: the bundle policy pass was refused (a wrapper before 0.166.0?) — the next bundle applies it", "refused", string(wr.Refused))
+	default:
+		lg.Info("osupdate: bundle policy pass done", "bundle_policy", string(wr.BundlePolicy))
+	}
+}
+
 func (e ConfigUpdateExecutor) download(ctx context.Context, url, dest string) (string, error) {
 	hc := e.HTTPClient
 	if hc == nil {

internal/osupdate/bundle_test.go

@@ -38,6 +38,9 @@ func (b *bundleWrapper) Run(_ context.Context, name string, args ...string) ([]b
 	var plan map[string]any
 	_ = json.Unmarshal(raw, &plan)
 	b.plans = append(b.plans, plan)
+	if plan["mode"] == "bundle-policy" { // R-1285: the pass after an install carries no bundle
+		return []byte(`OSAPPLY-REPORT {"mode":"bundle-policy","bundle_policy":{"install_mode":"byo","removed":[]}}` + "\n"), nil, nil
+	}
 	body, _ := os.ReadFile(plan["bundle"].(string))
 	b.bodies = append(b.bodies, body)
 	return []byte("OSAPPLY-REPORT " + b.rep + "\n"), nil, nil
@@ -153,3 +156,27 @@ func TestBundleURL(t *testing.T) {
 func (b *bundleWrapper) RunStdin(ctx context.Context, _ io.Reader, name string, args ...string) ([]byte, []byte, error) {
 	return b.Run(ctx, name, args...)
 }
+
+// R-1285: right after an install the agent asks the wrapper now on disk for the appliance-only policy pass — the bundle
+// run itself was planned by the OLD wrapper (on a BYO host upgraded from 0.164.0 our SSH drop-in stayed one bundle
+// longer). A refused or failed bundle runs no pass.
+// RED-PROOF: drop the policyPass call → one plan only.
+func TestR1285_AnInstalledBundleIsFollowedByThePolicyPass(t *testing.T) {
+	body := []byte(`{"format":1,"agent_version":"0.143.0","files":[]}`)
+	srv, sha := serveBundle(t, body)
+	w := &bundleWrapper{t: t, rep: `{"mode":"bundle","bundle":{"agent_version":"0.143.0","written":["/usr/local/sbin/felhom-os-apply"]}}`}
+	e, _ := bundleExec(t, w, srv.URL)
+	if err := e.Execute(signedCtx(), OpConfigUpdate, params("0.143.0", sha)); err != nil {
+		t.Fatal(err)
+	}
+	if len(w.plans) != 2 || w.plans[1]["mode"] != "bundle-policy" || w.plans[1]["layer"] != "host" || w.plans[1]["signed"] != nil {
+		t.Fatalf("after the install the agent must run the policy pass (host layer, no envelope): %v", w.plans)
+	}
+
+	w2 := &bundleWrapper{t: t, rep: `{"mode":"bundle","refused":{"code":"R3","reason":"x"}}`}
+	e2, _ := bundleExec(t, w2, srv.URL)
+	_ = e2.Execute(signedCtx(), OpConfigUpdate, params("0.143.0", sha))
+	if len(w2.plans) != 1 {
+		t.Fatalf("a refused bundle must run no policy pass: %v", w2.plans)
+	}
+}

internal/osupdate/leg.go

@@ -123,7 +123,8 @@ type WrapperReport struct {
 	Undo                bool            `json:"undo"`
 	LiveRestore         json.RawMessage `json:"live_restore"`
 	Facts               json.RawMessage `json:"facts"`
-	Bundle              json.RawMessage `json:"bundle"` // the config bundle's result (R-840, mode "bundle")
+	Bundle              json.RawMessage `json:"bundle"`        // the config bundle's result (R-840, mode "bundle")
+	BundlePolicy        json.RawMessage `json:"bundle_policy"` // the appliance-only policy pass (R-1285, mode "bundle-policy")
 	// OOMCheck (R-528, `09` decision 157): the docker layer's memory-kill check, {result, oom_killed, oom_event,
 	// exit_code, image, detail}. Carried to the hub UNCHANGED; the agent never reads it.
 	OOMCheck json.RawMessage `json:"oom_check"`

scripts/check-published-versions.py

@@ -98,7 +98,8 @@ TAG_RE = re.compile(r"^v(\d+\.\d+\.\d+)$")
 # THE retention number, read from the one file that owns it. A check and the policy it enforces
 # must read the same number from the same place, or they drift and the drift looks like a defect
 # in something else — which is exactly what happened on 2026-08-08/09 (R-287).
-RETENTION_FILE = os.path.join(os.path.dirname(os.path.abspath(__file__)), "retention-policy.json")
+RETENTION_FILE = os.environ.get("FELHOM_RETENTION_POLICY") or \
+    os.path.join(os.path.dirname(os.path.abspath(__file__)), "retention-policy.json")  # the env: decoy tests only
 
 
 def retention_kept():
@@ -114,6 +115,21 @@ def retention_kept():
         raise ValueError("generic_versions_kept must be a positive int, got %r" % (n,))
     return n
 
+
+def retention_pinned():
+    """D50 = A: the versions kept besides the newest N — the hub's MinAgent floor and the day-0 agent. Copies of hub
+    state the vouching session writes into the policy file (CI holds no hub secret). Missing key = no pins."""
+    with open(RETENTION_FILE, encoding="utf-8") as fh:
+        pins = json.load(fh).get("pinned") or {}
+    if not isinstance(pins, dict):
+        raise ValueError("pinned must be an object, got %r" % (pins,))
+    out = {}
+    for name, v in pins.items():
+        if not isinstance(v, str) or not re.match(r"^\d+\.\d+\.\d+$", v):
+            raise ValueError("pinned %s must be X.Y.Z, got %r" % (name, v))
+        out[name] = v
+    return out
+
 tried = []
 
 
@@ -211,6 +227,7 @@ def main():
     all_versions = versions
     try:
         keep = retention_kept()
+        pins = retention_pinned()
     except Exception as e:
         inconclusive("cannot read the retention policy (%s): %s" % (RETENTION_FILE, e))
 
@@ -218,10 +235,15 @@ def main():
     # lexically: "0.9.0" > "0.10.0" as strings, and that would silently drop the wrong end.
     def _key(v):
         return tuple(int(x) for x in v.split("."))
-    versions = sorted(all_versions, key=_key)[-keep:]
+    window = sorted(all_versions, key=_key)[-keep:]
+    # D50 = A: the pinned versions are asserted wherever they sit. A pin with no tag cannot be installed at all —
+    # a failure, never a skip.
+    unpinned_tags = sorted(("%s=%s" % (k, v)) for k, v in pins.items() if v not in all_versions)
+    versions = sorted(set(window) | set(v for v in pins.values() if v in all_versions), key=_key)
     dropped = [v for v in all_versions if v not in versions]
 
-    print("  %d released version(s); retention policy keeps the newest %d" % (len(all_versions), keep))
+    print("  %d released version(s); retention policy keeps the newest %d plus the pinned %s" % (
+        len(all_versions), keep, ", ".join("%s=%s" % kv for kv in sorted(pins.items())) or "(none)"))
     print("  verifying: %s" % ", ".join(versions))
     if dropped:
         # NEVER silent. A bounded check that does not say what it stopped covering is how a
@@ -232,6 +254,9 @@ def main():
         print("    their configs resolve; what is no longer asserted is the BINARY's presence.")
 
     bad = []
+    for pin in unpinned_tags:
+        bad.append((pin, ["the pinned version has NO git tag — nothing can install it (D50 = A)"]))
+        print("  FAIL pinned %s: no git tag v%s" % (pin, pin.split("=")[1]))
     for v in versions:
         pkg_url = "%s/api/packages/%s/generic/%s/%s/%s" % (GITEA_BASE, OWNER, PKG, v, PKG)
         raw_url = "%s/%s/%s/raw/tag/v%s/%s" % (GITEA_BASE, OWNER, REPO, v, PROBE_CONFIG)

scripts/retention-policy.json

@@ -1,41 +1,33 @@
 {
   "_comment": [
-    "THE retention number for published agent artifacts. One file, read by everything that",
+    "THE retention rule for published agent artifacts. One file, read by everything that depends on it, because a check",
-    "depends on it, because a check and the policy it enforces must read the same number from the",
+    "and the policy it enforces must read the same rule from the same place or they drift — and the drift looks like a",
-    "same place or they drift — and the drift looks like a defect in something else.",
+    "defect in something else (R-287, 2026-08-08/09; again R-1290, 2026-10-25).",
     "",
-    "WHAT WENT WRONG WITHOUT IT (2026-08-08/09). The registry stopped serving felhom-agent",
+    "D50 = A (operator ruling 2026-10-25, felhom.eu `09` §3 decision 262): the registry keeps the NEWEST 5 released agent",
-    "0.120.0 and older, while scripts/check-published-versions.py demanded that EVERY git tag",
+    "versions, PLUS the hub's MinAgent floor, PLUS the version the hub's day-0 artifact manifest names. The operator prunes",
-    "still be downloadable. Both rules are individually sensible; together they are impossible.",
+    "by hand to that rule (on 2026-10-25 he removed 0.159.0–0.160.0 and kept 0.158.1, the floor). CI asserts exactly these",
-    "CI went red at a commit whose own run had been green the day before, on a true finding that",
+    "and never more. Before D50 the number was 10 (the R-267 newest-10 prune, R-287/R-291).",
-    "no one could act on. The red will return at the next publish unless the two read one number.",
     "",
-    "HOW THE NUMBER WAS ARRIVED AT. generic_versions_kept is 10 because that is what the registry",
+    "THE PINS ARE COPIES OF HUB STATE, because CI cannot read the hub without a secret it must not hold. The session that",
-    "keeps: the deleter was ESTABLISHED on 2026-08-10 (R-287) — the newest-10 prune of generic packages",
+    "records an agent vouch (day0_agent) or declares a new MinAgent with a floor (hub_min_agent) updates `pinned` in the",
-    "run under R-267 (felhom-agent and felhom-golden generic held exactly 10 afterwards). Until then this",
+    "same commit (felhom-build-deploy skill, agent release). A pin that names a version with no tag is a FAILURE, never a",
-    "file called the 10 an observed state with no located ruling; that is superseded (corrected",
+    "skip — a pin nobody can install is the defect the pin exists to catch.",
-    "2026-10-05, R-291). Container packages are not pruned by that rule.",
     "",
-    "SO THIS FILE IS A FLOOR, NOT A LICENCE. It says: CI may assume nothing older than the newest",
+    "SO THIS FILE IS A FLOOR, NOT A LICENCE: it says what CI may assume is still downloadable. It does not authorise",
-    "N generic versions is still downloadable. It does NOT authorise deleting anything, and the",
+    "deleting anything else.",
-    "operator should confirm or replace the number — at which point this file changes and both",
-    "readers follow it in the same commit.",
     "",
-    "THE DEEPER BOUND, recorded so a future session does not have to re-derive it: the principled",
+    "NEVER retire a git TAG to satisfy this. felhom-host-install.sh fetches an agent's config files from",
-    "limit is the hub's vouched min_agent floor (0.127.0 on 2026-08-09). Nothing can install an",
+    "raw/tag/v<version>/configs/, so deleting a tag retires the ability to install that version at all."
-    "agent below it — the hub refuses to vouch one and boxes update to the floor — so a released",
-    "version below the floor being un-downloadable costs nothing real. Bounding on the floor would",
-    "be better than bounding on a count, and it needs the gate to read the hub, which is network",
-    "the gate does not have today. Filed as the follow-up in R-287.",
-    "",
-    "NEVER retire a git TAG to satisfy this. felhom-host-install.sh fetches an agent's config",
-    "files from raw/tag/v<version>/configs/, so deleting a tag retires the ability to install that",
-    "version at all — a strictly worse act than an un-downloadable binary."
   ],
-  "generic_versions_kept": 10,
+  "generic_versions_kept": 5,
   "readers": [
-    "scripts/check-published-versions.py — bounds its assertion to the newest N versions"
+    "scripts/check-published-versions.py — asserts the newest N versions plus every pinned version"
   ],
-  "recorded": "2026-08-09",
+  "recorded": "2026-10-25",
-  "recorded_by": "CC 2026-08-09; the number's source (the R-267 newest-10 prune, established 2026-08-10 by R-287) recorded 2026-10-05 (R-291)"
+  "recorded_by": "CC 2026-10-25, D50 = A (operator); earlier history: R-287 (2026-08-09), R-291 (2026-10-05)",
+  "pinned": {
+    "hub_min_agent": "0.158.1",
+    "day0_agent": "0.165.0"
+  }
 }

scripts/test_gate_decoys.py

@@ -179,13 +179,24 @@ def published_cases(base):
     TAGS = ["v0.150.%d" % i for i in range(3)]          # inside any retention window >= 3
     VERS = [t[1:] for t in TAGS]
 
+    # D50 = A: the gate also asserts the PINNED versions. The decoys run against their own policy file (the real pins
+    # name versions the fake Gitea does not hold); PINS is what each case writes there.
+    pol_dir = tempfile.mkdtemp(prefix="retention-decoy-")
+    pol = os.path.join(pol_dir, "retention-policy.json")
+    PINS = {}
+
     def case(name, setup, expect_rc, must=()):
         FAKE.reset()
         FAKE.tags = list(TAGS)
         FAKE.packages = set(VERS)
         FAKE.raw = set(VERS)
+        PINS.clear()
         setup()
-        rc, out = run([sys.executable, gate], ROOT, child_env(base))
+        with io.open(pol, "w", encoding="utf-8") as fh:
+            json.dump({"generic_versions_kept": keep, "pinned": dict(PINS)}, fh)
+        env = child_env(base)
+        env["FELHOM_RETENTION_POLICY"] = pol
+        rc, out = run([sys.executable, gate], ROOT, env)
         if not FAKE.hits:
             fails.append("published/%s: the gate never asked the fake Gitea - the seam is not wired" % name)
         report("published: " + name, rc, out, expect_rc, must)
@@ -220,6 +231,25 @@ def published_cases(base):
     case("BY DESIGN: a version older than the retention window is not asserted", retired, 0,
          ("NOT ASSERTED", "0.9.0"))
 
+    def pinned_old_ok():
+        FAKE.tags = ["v0.8.0"] + ["v0.10.%d" % i for i in range(keep)]
+        FAKE.packages = set(t[1:] for t in FAKE.tags)
+        FAKE.raw = set(t[1:] for t in FAKE.tags)
+        PINS["hub_min_agent"] = "0.8.0"
+    case("GENUINE (D50): a pinned version older than the window is asserted and downloads", pinned_old_ok, 0,
+         ("verifying: 0.8.0", "ALL RELEASED VERSIONS INSTALLABLE"))
+
+    def pinned_old_gone():
+        pinned_old_ok()
+        FAKE.packages.discard("0.8.0")
+    case("FACT (D50): a pinned version older than the window whose package 404s", pinned_old_gone, 1,
+         ("FAIL v0.8.0", "binary NOT downloadable"))
+
+    def pinned_no_tag():
+        PINS["day0_agent"] = "0.150.9"
+    case("FACT (D50): a pin that names a version with no tag", pinned_no_tag, 1,
+         ("FAIL pinned day0_agent=0.150.9", "no git tag"))
+
     def five_hundred():
         FAKE.tags_status = 500
     case("INCONCLUSIVE: the tags api answers 500", five_hundred, 2, ("INCONCLUSIVE",))