CI workflow: the failure mail links the public run page, by run id (R-1135)

The alarm step built its link from GITHUB_SERVER_URL, which on this runner
is the in-cluster Gitea address, and from the run NUMBER, which Gitea 1.26
reads as a run id first (felhom.eu /actions/runs/1116 opened run id 1116 =
run number 613). Now https://gitea.dooplex.hu + GITHUB_RUN_ID, and the link
is printed into the job log as ALARM-LINK. Same change in all four repos.

.gitea/workflows/gates.yml

@@ -108,8 +108,16 @@ jobs:
 
           repo = os.environ.get("GITHUB_REPOSITORY", "?")
           sha  = os.environ.get("GITHUB_SHA", "?")
-          run  = os.environ.get("GITHUB_RUN_NUMBER", "?")
+          # R-1135: the link is for a PERSON, so it is the public address and the run ID.
-          srv  = os.environ.get("GITHUB_SERVER_URL", "https://gitea.dooplex.hu")
+          # NOT GITHUB_SERVER_URL: the runner registers at the in-cluster address, so that variable
+          # reads http://gitea.gitea-system.svc.cluster.local:3000, which nobody outside the cluster
+          # can open (forced-test mail, job 1997). NOT GITHUB_RUN_NUMBER: Gitea 1.26 reads
+          # /actions/runs/<n> as a run id first, so a run number that is also an older run's id
+          # opens THAT run (measured 2026-10-18: felhom.eu /actions/runs/1116 opened the run whose
+          # id is 1116 = run number 613). The link is printed too, so the job log carries it.
+          run  = os.environ.get("GITHUB_RUN_ID", "?")
+          srv  = "https://gitea.dooplex.hu"
+          print("ALARM-LINK %s/%s/actions/runs/%s" % (srv, repo, run))
 
           body = json.dumps({
               "from": "Felhom CI <monitoring@felhom.eu>",

CHANGELOG.md

@@ -1,3 +1,126 @@
+## Unreleased (2026-10-18) — CI workflow: the failure mail links the public run page, by run id (R-1135); no code change
+
+`.gitea/workflows/gates.yml`, the alarm step: the mail's run link is now `https://gitea.dooplex.hu/<repo>/actions/runs/<run id>`.
+It used `GITHUB_SERVER_URL`, which on this runner is the in-cluster address (`http://gitea.gitea-system.svc.cluster.local:3000`,
+the address the runner registers at), so nobody outside the cluster could open it. It also used the run NUMBER, and
+Gitea 1.26 reads `/actions/runs/<n>` as a run id first: a number that is also an older run's id opened that other run
+(measured 2026-10-18: `felhom.eu/actions/runs/1116` opened run id 1116, run number 613). It now uses `GITHUB_RUN_ID` and
+prints the link (`ALARM-LINK …`) into the job log. Same change in all four repos' workflows.
+
+## Unreleased — three fixes from the 2026-10-12 security review and the storage build (R-1093, R-978, R-973) (2026-10-18)
+
+Committed, not released — the release and its vouch are the operator's.
+
+- **R-1093 — an EMPTY partition no longer asks the typed „data will be erased" confirmation.** `lsblk` prints the
+  parent disk's table type (PTTYPE) on every partition, so `SudoHostOps.InspectDevice` read a blank partition as
+  „device has a partition table". Now lsblk's PTTYPE is set aside only when `blkid -p` on the partition itself proved
+  it read the device and that it is a partition of a table of exactly that type (`PART_ENTRY_NUMBER` +
+  `PART_ENTRY_SCHEME` equal to lsblk's PTTYPE). The partition's own signatures still come from `blkid -p`: a
+  filesystem or a nested table inside it stays data-bearing; a blkid that printed nothing, a whole disk, or a scheme
+  that differs keeps the old, safe reading. **No new command, no sudoers or config-bundle change** (the two probes are
+  the same allow-listed argv). Test `TestInspect_R1093_PartitionVerdicts` (8 rows), red-proved: with the old
+  `if pttype != ""` both empty-partition rows fail `DataBearing=true reason="device has a partition table"`.
+  **Hardened after the same night's review (fail-closed on every path):** `PART_ENTRY_*` is read from the PARENT's
+  table, so lsblk's PTTYPE is set aside only when blkid also **exited 0** — exit 2, an I/O error or a missing binary
+  keep „device has a partition table"; and an `lsblk` that exits 0 with output that does not parse into a device (or
+  lists none) now leaves the device UNPROBED („could not be reliably inspected") instead of „blank" — that hole
+  predates R-1093. Six more rows in the same test, each red-proved (`blkidErr` check removed → „device is blank";
+  `&& parsed` removed → `Probed=true … „device is blank"`).
+- **R-978 — an abandoned second escrow ceremony no longer makes the customer's recovery code useless.** The root
+  ceremony (the FELHOM_ESCROW argv, unchanged: `--upload --output=json`) used to upload the new package before the new
+  code was claimed, and the hub overwrites the current package when the sealed secrets are unchanged — so a customer
+  who started the ceremony again and walked away (or an agent restart in the 10-minute window) was left with a
+  written-down code that opened nothing. Now json mode hands the package back (`CeremonyOutput.PendingUpload`, output
+  **version 2**; text mode still uploads at once) and the daemon uploads it at `POST /escrow/ceremony/claim`, **before**
+  the code is handed out (`EscrowCeremonyConfig.Upload` → `putEscrowBody`, the same PUT and per-host key). A failed
+  upload answers 502 and keeps the code claimable; the TTL and a new start wait while the claim's upload is in flight;
+  no uploader → 503; an output that uploaded by itself or carries no package fails the ceremony. Status `uploaded`
+  now reads false until the claim. The controller is unchanged (it never read `uploaded`; its claim already assumed
+  the package is on the hub when the claim returns). **No sudoers or config-bundle change.** Tests (localapi)
+  `TestEscrowCeremony_R978_*` (abandon by TTL / by a new start / by a restart → the hub's package still opens with
+  code 1; claim stores before handing out; upload failure; in-flight TTL; no uploader; no package) and (cmd)
+  `TestEscrowOptsFromCLI_JSONModeDefersUpload`, `TestDeliverEscrowPackage_JSONModeSendsNothing`,
+  `TestR978_CeremonyUploadIsWiredIntoTheDaemon`. Red-proved: upload-at-create restored → „code 1 no longer opens it
+  (uploads=2)"; the in-flight TTL guard removed → „the TTL voided the code while its package was uploading";
+  `deferUpload: false` → „json mode must defer the upload"; the `Upload:` line removed → „has no Upload". **Residual:**
+  the customer's typed-back check after the claim is browser-only, so „confirmed" here means „claimed" — a customer
+  who opens the code and closes the page without writing it down has replaced the package.
+- **R-973 — the local-API whole-guest backup route has a cadence cap.** Every successful run prunes its tier, so a
+  compromised controller calling `POST /backup` in a loop pushed out every pre-compromise restore point (local
+  keep-last 3; off-site via ep0's keep-last 2). Now (`internal/localapi/backup_cap.go`): a tier whose cadence is ≤ 24 h
+  allows **3 successful runs per rolling 24 h** (the `07` §6 day: the night run, a „Mentés most" press, and the next
+  night, which since R-899 still runs and can land inside the same 24 h — so a press never costs the night); a longer
+  tier allows **1 per max(cadence − 24 h, cadence/2)** — 6 days for the 7-day off-site tier, which the controller only
+  runs when due and the button never targets. Only successes count (a failed run prunes nothing). A refusal answers
+  **429** with the next allowed time, is logged at WARN, is saved, and for 24 h raises the CRITICAL host-report status
+  `localapi:backup-cadence-cap` — the hub's existing operator-only `agent_capability_degraded` event, **no hub change**.
+  **The cap survives an agent restart:** `BackupSuccessState` (the R-894 file) now also keeps the newest 8 success
+  times and the last refusal per tier and guest (additive JSON fields; a pre-R-973 file counts its one known
+  success). Tests `TestBackupCap_*` (an off-site flood keeps a pre-compromise copy for 5+ days; a local flood runs 3
+  a day; the scheduled day is never refused; restart; failures; the operator signal) + `TestBackupCadenceCap`,
+  `TestBackupSuccessState_R973_*`, `TestBackupCapStatus_*`, `TestR973_CapRefusalIsWiredIntoTheReport`. Red-proved:
+  the cap disabled → „the off-site tier lost every pre-compromise copy: 10 runs"; the disk history ignored → „after a
+  restart the 4th run in a day was allowed"; the wiring line removed → „never binds backupCapRefusal". **Limits, stated:**
+  with keep-last 3 the local tier's three points can still be replaced within one day (3 per day is the schedule's own
+  pace plus one press); the hub's event text reads „agent privileged capability degraded" (the type is generic). No
+  sudoers or bundle change.
+
+## v0.159.2 — the orphan thin check runs only after a failed backup (R-1012 follow-up) (2026-10-18)
+
+**Released 2026-10-18 (committed 2026-10-17): sha256 `4d98868e922d6c10665a6c0da9323e8627ea9c589094e81d27d741e625d9d8f9`, bundle `477704ea2980948433ffc34bb59f008cc11410fd9683d386bfd3e2f085d55bc9`; the vouch is the operator's.** No `configs/` change since v0.159.1. The independent review of 2026-10-18 found nothing in it. v0.159.0/v0.159.1 ran the R-1012 orphan check
+before EVERY snapshot-mode whole-guest backup, and that check takes a short metadata snapshot of the live thin pool.
+Now `BackupRunner` runs it only when the snapshot step has failed (the vzdump start or its task) — the moment the
+answer is needed. The finding is still appended in plain words to the failed backup's error, exactly as before; a
+successful backup never runs the check and keeps a clean record; stop mode and an unwired runner are unchanged. The
+`thin-orphan-check` capability text says „after a failed whole-guest backup". Tests
+`TestBackup_SnapshotFailure_NamesTheOrphan` (now: exactly 1 call, after the failure) and
+`TestBackup_ThinCheck_SilentWhenNotApplicable` (now: 0 calls on a successful backup), red-proved against the v0.159.1
+`runner.go`: failure legs „thin check ran 2 times, want 1", success leg `calls=1`. No root file, sudoers or bundle
+change. (Installer side, felhom.eu, same night: `felhom-host-install.sh --rotate-recovery-only`, R-1141.)
+**Proven live 2026-10-18 (demo-hp):** a green snapshot backup with 0 `felhom-thin-check` calls (the 0.159.1 night backup
+had 1). The failed leg stays covered by the fixture test.
+
+## v0.159.1 — the signed decommission prunes the guest-bind record (found by v0.159.0's own live proof) (2026-10-17)
+
+**v0.159.0 is superseded — do not vouch it.** Everything of v0.159.0 below, plus: `signedjobs.DecommissionExecutor`
+drops the drive from every guest's bind record (`SetGuestBinds`, wired to the agent's `GuestBindStore`), as the
+self-serve decommission always did. Without it a drive decommissioned by an operator-signed job and then forgotten
+(R-1124) left a record the 20-second guest-bind re-assert warned about every tick („reconcile: skipping non-enrolled
+drive") — seen on demo-hp minutes after the R-1124 proof. Test `TestDecommissionExecutor_PrunesTheGuestBindRecord`,
+red-proved. Released: tag `v0.159.1`, sha256 `de8f6ad9…aff9`, bundle `2af3d27d…7cd6` (the same root files as v0.159.0).
+**Not vouched.** Delivery to a box on the v0.158.1 bundle: `agent_update 0.159.1` → `agent_config_update 0.159.0-step1`
+(sha `1d65edb3…d6cf`, published) → `agent_config_update 0.159.1` (the bundle adds `felhom-thin-check`, R16).
+
+## v0.159.0 — forget a decommissioned drive that is gone (R-1124); name an orphan thin device in a backup failure (R-1012); Go 1.27.2 (R-972) (2026-10-17)
+
+**Superseded by v0.159.1 — do not vouch.** Task: „agent release, the catalog allow-list, break-glass rotation, off-site
+history check" (felhom.eu `REPORT-agent-allowlist-rotation-2026-10-17.md`).
+
+- **Go 1.27.2 + x/crypto v0.58.0** (R-972, `37b99d7`): the first release that carries them (v0.158.1 was built with
+  go1.26.0).
+- **R-1124 — a decommissioned drive that is gone is forgotten.** `SudoHostOps.ForgetRetiredDrives` (agent start and the
+  20-second storage tick): for a Felhom drive unit whose intent is `decommissioned`, whose `/dev/disk/by-uuid` link is
+  gone and which is not mounted, `systemctl disable` + the new root verb `felhom-priv-apply unit-forget <unit>` +
+  `daemon-reload`. The unit is gone, so a later boot no longer waits for the drive and the agent stops listing it (no
+  „absent by UUID" line every 20 s). A decommissioned drive still plugged in KEEPS its unit and raw mount, so the
+  one-click re-enroll (H3) still works. `unit-forget` re-checks as root: a felhom by-UUID drive unit (not a network
+  unit), disabled, its `Where` not mounted, its by-UUID link absent — otherwise refused. Sudoers `FELHOM_MOUNT` gains the
+  one verb. Tests `TestForgetRetiredDrives_*` and `configs/test_felhom_priv_apply.py` `UnitForget`, red-proved.
+  **Live on demo-hp:** a loop-backed test drive enrolled through the local API, decommissioned by a signed job (kept
+  its unit while present), unplugged → 8 s later the unit and its wants link were removed and `GET /disks` dropped it.
+- **R-1012 — an orphan thin device is NAMED in the backup failure, never removed.** New READ-ONLY root helper
+  `felhom-thin-check <vg> <pool>` (config bundle): LVM's thin ids before and after, the pool's ids through a metadata
+  snapshot it reserves and releases (refuses — exit 3 — when someone else holds one; releases on every exit; sends no
+  other pool message). `internal/thincheck`: an orphan = in the pool and in neither LVM list, on two reads 3 s apart;
+  fails open. `BackupRunner.SetThinCheck`: before a snapshot-mode backup (a WARN when found) and again after a failure —
+  the finding is appended in plain words to the error the controller mails to the operator, pointing at the new
+  runbook `whole-guest-backup-orphan-thin.md`. Sudoers `FELHOM_THINCHECK`, capability `thin-orphan-check`. Tests
+  `internal/thincheck`, `TestBackup_SnapshotFailure_NamesTheOrphan`, `configs/test_felhom_thin_check.py`, red-proved.
+  **Live on demo-hp:** on a scratch thin pool (a loop file on `/mnt/hdd_1`, never `pve/data`) the helper listed a made
+  orphan (device 7) and refused a snapshot held by someone else; on `pve/data` it read 1, 2 = LVM's and released; a
+  normal whole-guest backup of 9201 ran the check first and finished green.
+- Released: tag `v0.159.0`, sha256 `3cfd0a25…edfed5`, bundle `40663c4f…6d681`; step bundle `0.159.0-step1`.
+
 ## CI only — a manual proof that the failure mail can be sent (R-1129, 2026-10-17)
 
 No agent release; no binary changes.

CONTEXT.md

@@ -1,5 +1,22 @@
 # CONTEXT — felhom-agent working state
 
+> **2026-10-18 (night) — UNRELEASED on main: R-1093, R-978, R-973** (release + vouch: the operator's). R-1093 an empty
+> partition is blank (lsblk's PTTYPE on a partition is the parent's; set aside only on a clean blkid exit 0 with a
+> matching `PART_ENTRY_SCHEME`; an unparsable lsblk is unprobed). R-978 the escrow package reaches the hub only at the
+> claim (ceremony output v2, `PendingUpload`; `EscrowCeremonyConfig.Upload`). R-973 the `POST /backup` cadence cap.
+> **Decision (decided by CC unattended — operator may reverse): the cap's numbers.** Q: how many whole-guest runs may
+> the guest ask for? Options: (a) 2 per 24 h on the daily tier — keeps one local point older than a day, but refuses
+> the NIGHT after a press (R-899's night can fall inside the same 24 h); (b) 3 per 24 h + 1 per 6 days on the weekly
+> tier — the schedule never refused, a second press in a day refused, the local tier still replaceable in a day;
+> (c) none — status quo. Chose (b): the off-site tier is the one `07` §8 row 10 relies on and it is fully protected
+> for ~6 days; the local half's real fix is a time-bucketed prune (keep-daily), which is a retention change for the
+> operator. No sudoers or bundle change in any of the three.
+
+> **2026-10-17 — v0.159.1 RELEASED (0.159.0 superseded, do not vouch).** Go 1.27.2; R-1124 a decommissioned drive that is
+> gone is forgotten (`ForgetRetiredDrives`, root verb `unit-forget`); R-1012 the read-only `felhom-thin-check` names an
+> orphan thin device in a failed backup's text; the signed decommission prunes the guest-bind record. Bundle adds a path →
+> step bundle `0.159.0-step1`. Proven on demo-hp; delivered by signed jobs on the operator's yes. Vouch: the operator's.
+
 > **2026-10-16 — v0.158.1 RELEASED and VOUCHED (storage identity build B; 0.158.0 superseded, do not vouch).** The internal
 > data place (root helper `felhom-internal-place`, the record in the state dir, mounted at /mnt/belso like a drive), the
 > signed `add_internal_place`, partitions in `/disks/candidates` (`formattable` only while nothing on the disk is mounted),

REPORT.md

@@ -1,12 +1,12 @@
-# REPORT — the agent on Go 1.27.2 (R-972, 2026-10-16) — NOT released
+# REPORT — agent v0.159.2 released (2026-10-18)
 
-Written in ASD-STE100 Simplified Technical English. Part of burndown 2 (felhom.eu `REPORT-burndown2-2026-10-16.md`).
+**In plain words.** The pool check that names an orphan snapshot now runs only after a whole-box backup has failed — not
+before every backup. Released today; it runs on demo-hp, Tester 1 and demo-felhom. The vouch is the operator's.
 
-- **What changed:** `go.mod` `go 1.27.2` and golang.org/x/crypto v0.58.0. No code change.
+- sha256 `4d98868e922d6c10665a6c0da9323e8627ea9c589094e81d27d741e625d9d8f9`, bundle
-- **Proof:** `go build ./...`, `go vet ./...`, `go test ./...` rc 0 on Go 1.27.2; `agent_gates.py --fast` all OK.
+  `477704ea2980948433ffc34bb59f008cc11410fd9683d386bfd3e2f085d55bc9`, tag `v0.159.2`; no `configs/` change since 0.159.1.
-- **Not done, on purpose:** no release and no deploy — agent releases wait for the operator's vouch. R-972 stays
+- Live proof on demo-hp: a whole-guest backup of 9201 (snapshot mode, completed) wrote **0** `felhom-thin-check` sudo
-  NARROWED for the agent: the next vouched release carries the new Go.
+  lines; the night's backup on 0.159.1 wrote 1. The failure leg is covered by `TestBackup_SnapshotFailure_NamesTheOrphan`.
-
+- Delivered on the operator's yes by signed jobs (agent_update + agent_config_update): all three hosts `update
-## Observations
+  committed`, capability probe 75/75. Evidence: felhom.eu `documentation/audits/day-2026-10-18/P4/`.
-
+- The independent review of 2026-10-18 found nothing in this change.
-1. The release script needs no change: it builds with the toolchain `go.mod` names. NOT-A-FINDING: GOTOOLCHAIN=auto on DooPlex fetched go1.27.2 for this build and will for a release build.

REUSE.md

@@ -114,8 +114,8 @@
 | `Engine.RunSignedJob` | internal/reconcile/job.go | `RunSignedJob(ctx, intent, signed, exec) JobResult` | executing a gated destructive job | Idempotency by nonce; journaled |
 | `escrow.Create` | internal/escrow/escrow.go | `Create(ctx, CreateOptions) (CreateResult, R, error)` | PBS-key escrow (zero-knowledge) | Recovery code returned SEPARATELY from the result (anti-log); self-verifies recoverability |
 | `escrow.GenerateRecoveryCode` / `joinSafe` / `RecoveryCodeSep` | internal/escrow/wordlist.go | `GenerateRecoveryCode() (string, error)` | minting the customer recovery code R | Draws from the EFF large list **filtered of every word containing `RecoveryCodeSep`** (4 entries: drop-down, felt-tip, t-shirt, yo-yo) so a code always segments back into exactly 10 words — a hyphenated word made codes ambiguous to transcribe AND flaked the test ~1/5 (v0.93.0). Generation-only: **already-issued codes stay valid**, R is verified as a whole passphrase and never re-split. Never count words by splitting the joined string — count what the generator drew |
-| `escrow.CeremonyBinary` / `CeremonyArgs()` / `CeremonyOutput` | internal/escrow/ceremony.go | the ONE fixed sudo self-invocation argv + the `--output=json` wire object (v1) | controller-driven ceremony (v0.88.0) | SINGLE SOURCE shared by the localapi exec, the capability manifest entry, and (byte-identically) the FELHOM_ESCROW sudoers line — `TestEscrowCeremonyArgvPinned` + `TestManifestCoveredBySudoers` lock all three. Never flag-helpers, never `--`→`-` (spike §2.2) |
+| `escrow.CeremonyBinary` / `CeremonyArgs()` / `CeremonyOutput` | internal/escrow/ceremony.go | the ONE fixed sudo self-invocation argv + the `--output=json` wire object (v2, R-978: json mode returns the package in `PendingUpload` instead of uploading) | controller-driven ceremony (v0.88.0) | SINGLE SOURCE shared by the localapi exec, the capability manifest entry, and (byte-identically) the FELHOM_ESCROW sudoers line — `TestEscrowCeremonyArgvPinned` + `TestManifestCoveredBySudoers` lock all three. Never flag-helpers, never `--`→`-` (spike §2.2) |
-| localapi escrow ceremony job | internal/localapi/escrow_ceremony.go | `POST /escrow/ceremony` + status + ONE-SHOT claim + preflight | the wizard's agent half | R lives ONLY in `Server.escrowR` (NEVER the job struct — snapshots must be structurally R-free); zeroed on claim/supersede/10-min TTL (`unclaimed_void`); in-memory BY DESIGN (restart loses R safely; re-run supersedes); subprocess stdout is SECRET-BEARING → parsed then zeroed, never logged |
+| localapi escrow ceremony job | internal/localapi/escrow_ceremony.go | `POST /escrow/ceremony` + status + ONE-SHOT claim + preflight | the wizard's agent half | R lives ONLY in `Server.escrowR` (NEVER the job struct — snapshots must be structurally R-free); zeroed on claim/supersede/10-min TTL (`unclaimed_void`); in-memory BY DESIGN (restart loses R safely; re-run supersedes); **R-978: the new package reaches the hub only at the claim, before R is handed out (`EscrowCeremonyConfig.Upload`); TTL and a new start wait while that upload is in flight — an abandoned ceremony leaves the previous package current**; subprocess stdout is SECRET-BEARING → parsed then zeroed, never logged |
 | `poke.Listener` + `poke.Port` | internal/poke/poke.go | `NewListener(resolve, trigger, port, logger)`; `poke.Port = 51822` | agent-plane immediate-sync (Direction-2a, v0.89.0) | Binds a contentless UDP socket EXCLUSIVELY to the box's WG /32 (`wgtunnel.LoadAssignedAddr`), fires the hub-loop out-of-band trigger. **Port 51822 is a SHARED cross-repo contract** — the hub poke sender + the ep0 `felhom-poke` forced-command target the SAME number; change one → change all three. Contentless (payload ignored), leading-edge debounced (`DebounceWindow`), WG-confined (kernel EKEYREJECTED refuses non-peer /32s). Wired only when `wg_tunnel.enabled` |
 | `wgtunnel.LoadAssignedAddr` | internal/wgtunnel/manager.go | `LoadAssignedAddr(stateDir) (netip.Addr, bool)` | the box's own WG /32 without a Manager | Reads `registered.json`; ok=false until registered; strips the /32 → bare addr (the poke bind target) |
 | `fasttick.Loop` + `fasttick.SourceFunc` | internal/fasttick/fasttick.go | `New(out chan<- struct{}, interval, logger, sources...)`; `Source.Unconverged() (bool, reason)` | agent-plane immediacy SECONDARY (v0.90.0, R-28): pulse the SAME out-of-band trigger every 30 s while ANY source is unconverged, self-disarm on convergence | STATE-BASED (no timer, nothing to journal). Every source MUST be a CACHED read (no exec/network per tick) — `desiredProvider.Generation()`, `reconcile.Engine.LastResult()` (Planned−Pending>0), `pbsdrLoop.PBSDRStatus().State=="waiting_secret"` ONLY, `wgtunnel.Manager.TunnelConvergence()`. The LOUD pbsdr states + destructive `pending_signature` are DELIBERATELY EXCLUDED (a stuck-loud box must not hammer). Pulses the cap-1 channel non-blocking (coalesces with poke/watchdog); NEVER touch `MinPollSeconds`/`clampInterval`/the ticker |
@@ -174,6 +174,7 @@
 | `capability` store-grant probe (`storeGrantStatuses` / `storeGrantVerdict` / `Client.Permissions`) | cmd/felhom-agent/main.go, internal/proxmox/query.go | *"may the agent READ this backup tier?"*, one `capability.Status` per configured tier | R-185. **Never infer permission from an empty content listing** — `{"data":[]}` is what a FORBIDDEN tier and a NEWBORN tier both return, and that ambiguity hid an unreadable host tier on both demo boxes. Ask `/access/permissions` **as the agent's own token** (root always says yes). **The ungranted answer is not empty and not a 403** — it carries the privileges inherited from the box-wide `/` grant, so test for **`Datastore.AllocateSpace`** specifically; path-presence or `Datastore.Audit` reports a blinded storage healthy. Probed set comes from `BackupTiers()`, never a fixed list. Critical except the `local` fallback. Composes AROUND the sudo prober (the `poolReadStatus` precedent); `Status`'s wire shape is untouched so the hub alert is free. Unreachable PVE ⇒ degraded, never ok. |
 | `backup.RestoreTestState` | internal/backup/restoretest_state.go | `RecordSuccess(target,archive,tier,verified,t)` / `ProvenArchive(target)` / `ProvenRestoreTests(ctx)` / `LastSuccess(target)` / `OldestFirst(targets)` | Per-tier restore-test PROOF state, persisted (atomic tmp+rename) — **which archive** was proven, and when (R-86) | **Credit ONLY on success** — a permanently failing tier must keep sorting first, or it looks freshly proven and stops being retried. Ties break on target id: without it, two tiers proven in the same second rotate by Go's randomised map order. **This one NEEDS persistence unlike R-84** — R-84 had ground truth to consult (the archive is still on the storage); a restore-test destroys its scratch and leaves no artifact. **R-86: the ARCHIVE is the state, the time is metadata** — a time alone cannot answer "have we proven THIS archive", which is the due-check's whole question. A pre-R-86 file (bare RFC3339 per target) keeps its time and yields NO proven archive, so each tier is due once after the upgrade; reading a legacy time as proof of the current archive would invent a guarantee. **R-189: it is also the REPORTABLE half of the restore-test signal.** The in-memory `backup.Store` holds only this process's latest run, and under per-archive due-ness the agent will not re-test a proven archive — so a proof lost to a restart is not repeated for a whole archive generation (observed live: a passing 14.5 GB offsite restore reached no host-report). `ProvenRestoreTests` renders the stored proofs as `hub.RestoreTest` entries and the collector merges them; a record missing the archive or the tier is NOT emitted, because an unproven tier reading as proven is worse than the defect. **Only successes are stored, deliberately:** a success suppresses future work, a failure causes it. |
 | `backup.BackupSuccessState` | internal/backup/backup_state.go | `RecordBackupSuccess(target, b)` / `LastKnownSuccess(target, vmid)` | Newest SUCCESSFUL backup per tier+guest, persisted (atomic tmp+rename) — the due-check's fallback when the tier's storage cannot be read after a restart (R-894) | **Read ONLY when the storage cannot be read** — a storage that answers is the ground truth (R-84), and an archive absent there must make the tier due even when this file remembers one. A saved copy older than the cadence still reads due. Only successes are written. |
+| localapi backup cadence cap (R-973) | internal/localapi/backup_cap.go | `backupCadenceCap(cadence)`, `Server.capRefuses`, `Server.BackupCapRefusal()` | POST /backup refuses (429) a run the `07` §6 schedule would never ask for: ≤24 h tier → 3 successes per rolling 24 h; longer tier → 1 per max(cadence−24 h, cadence/2) | Counts SUCCESSES only, from this process AND `BackupSuccessState.RecentBackupSuccesses` (so a restart does not reset it); refusals saved (`RecordBackupRefusal`) and raised for 24 h as the CRITICAL host-report status `localapi:backup-cadence-cap` (`backupCapStatus` in main) → the hub's existing `agent_capability_degraded`. Tests `TestBackupCap_*`. |
 | `hub.ProvenRestoreTestReporter` + `Collector.SetProvenRestoreTests` | internal/hub/collect.go | the DURABLE restore-test source, merged with the in-memory one | R-189. Merge rule: **one entry per tier, newest by `TestedAt` wins** — a fresh failure beats a stored success (the failure is the news, and it lives nowhere else), a stored success beats a stale in-memory entry after a restart, and a tier never appears twice (the hub would read two tests). An unparseable timestamp counts as OLDER, so a malformed entry cannot displace a good one. **The wiring is pinned by an AST test** — the method this replaced (`RestoreTestState.Snapshot`) carried a doc comment naming a host-report gauge and had no caller for weeks. |
 | `backup.SpecBuilder` / `backup.TierPicker` / `(*BackupRunner).PickSettledRestoreCandidateOn` | internal/backup/schedule.go, runner.go | `func(ctx,archive) RestoreTestSpec`; `func(ctx,target,notAfter) (archive,landed,error)` | The per-run restore-test spec + per-tier **settled** candidate lookup (R-85, widened by R-86) | The spec is built **PER RUN**, never frozen at construction — the pre-R-85 immediately-invoked value made the offsite tier unschedulable AND went stale on any config change. `SourceTier` comes from **the archive**, never the configured target (the v0.100.0 rule). A tier with no archive returns `("", zero, nil)` — **`""` is NOT an error**, or every fresh box looks broken for its first week. **R-86: `notAfter` is the settle cutoff** (zero = no cutoff, which is what keeps `PickRestoreCandidateOn` a one-line call into it), and the picker now skips entries failing `archivePlausiblyComplete` — under per-archive due-ness an incomplete phantom would be picked forever, fail forever, never earn proof, and make the tier due at EVERY evaluation. |
 | `localapi.BackupTier` + `normalizeBackupTiers` / `config.BackupConfig.BackupTiers` | internal/localapi/backup_tiers.go, internal/config/config.go | `normalizeBackupTiers(tiers, legacy, cadence) []BackupTier`; `BackupTiers() ([]BackupTier, []string)` | THE R-82 multi-tier resolution — one runner per tier, primary first | **The untargeted local-API contract is FROZEN**: no `?target=` ⇒ primary tier ⇒ pre-R-82 response BYTES (Target is `omitempty` and stays empty). Never default a missing cadence — reject it and log the warning at ERROR. Never share one retention knob between tiers. Jobs are keyed by (vmid,target). |

cmd/felhom-agent/backup_cap_r973_test.go

@@ -0,0 +1,67 @@
+package main
+
+import (
+	"go/ast"
+	"strings"
+	"testing"
+	"time"
+
+	"gitea.dooplex.hu/admin/felhom-agent/internal/capability"
+)
+
+// R-973: a cadence-cap refusal becomes a CRITICAL degraded status — what the hub's existing
+// agent_capability_degraded event (operator-only) keys on — and clears when the refusal ages out.
+func TestBackupCapStatus_RefusalIsCriticalDegraded(t *testing.T) {
+	at := time.Date(2026, 10, 18, 3, 0, 0, 0, time.UTC)
+	s := backupCapStatus(func() (string, int, time.Time, bool) { return "felhom-pbs", 9201, at, true })
+	if !s.Critical || s.Status != capability.StatusDegraded || !strings.Contains(s.Reason, "9201") || !strings.Contains(s.Reason, "felhom-pbs") {
+		t.Fatalf("a refusal must raise a critical degraded status naming the guest and the tier: %+v", s)
+	}
+	if s := backupCapStatus(func() (string, int, time.Time, bool) { return "", 0, time.Time{}, false }); s.Status != capability.StatusOK || !s.Critical {
+		t.Fatalf("no refusal → ok (and still critical, so a later refusal alerts): %+v", s)
+	}
+	if s := backupCapStatus(nil); s.Status != capability.StatusOK {
+		t.Fatalf("no local API → ok: %+v", s)
+	}
+}
+
+// Wiring (the built-but-never-wired class): runDaemon's capability probe appends backupCapStatus, and the
+// late-bound source is set from the local API server.
+//
+// COMPANION RED-PROOF (observed 2026-10-18): delete `backupCapRefusal = localSrv.BackupCapRefusal` →
+// "runDaemon never binds backupCapRefusal to the local API server". Restored.
+func TestR973_CapRefusalIsWiredIntoTheReport(t *testing.T) {
+	_, f := parseMain(t)
+	var appended, bound bool
+	for _, d := range f.Decls {
+		fd, ok := d.(*ast.FuncDecl)
+		if !ok || fd.Name == nil || fd.Name.Name != "runDaemon" || fd.Body == nil {
+			continue
+		}
+		ast.Inspect(fd.Body, func(n ast.Node) bool {
+			switch x := n.(type) {
+			case *ast.CallExpr:
+				if id, ok := x.Fun.(*ast.Ident); ok && id.Name == "backupCapStatus" && len(x.Args) == 1 {
+					if a, ok := x.Args[0].(*ast.Ident); ok && a.Name == "backupCapRefusal" {
+						appended = true
+					}
+				}
+			case *ast.AssignStmt:
+				if len(x.Lhs) == 1 && len(x.Rhs) == 1 {
+					if l, ok := x.Lhs[0].(*ast.Ident); ok && l.Name == "backupCapRefusal" {
+						if sel, ok := x.Rhs[0].(*ast.SelectorExpr); ok && sel.Sel.Name == "BackupCapRefusal" {
+							bound = true
+						}
+					}
+				}
+			}
+			return true
+		})
+	}
+	if !appended {
+		t.Fatal("runDaemon's capability probe does not call backupCapStatus(backupCapRefusal)")
+	}
+	if !bound {
+		t.Fatal("runDaemon never binds backupCapRefusal to the local API server")
+	}
+}

cmd/felhom-agent/escrow_r978_test.go

@@ -0,0 +1,112 @@
+package main
+
+import (
+	"context"
+	"encoding/json"
+	"go/ast"
+	"testing"
+
+	"gitea.dooplex.hu/admin/felhom-agent/internal/config"
+	"gitea.dooplex.hu/admin/felhom-agent/internal/escrow"
+)
+
+// R-978, the root half: the controller-driven ceremony (json mode, the FELHOM_ESCROW argv with
+// `--upload`) must hand its package back instead of sending it — the daemon sends it at the claim.
+//
+// COMPANION RED-PROOF (observed 2026-10-18): set `deferUpload: false` in escrowOptsFromCLI → this fails
+// with "json mode must defer the upload (R-978): got deferUpload=false", and the delivery test's json leg
+// then sends the package at create time ("put called 1 times"). Restored.
+func TestEscrowOptsFromCLI_JSONModeDefersUpload(t *testing.T) {
+	// The FELHOM_ESCROW argv: --selftest=escrow-create --upload --output=json.
+	if o := escrowOptsFromCLI("", false, false, true, "", true); !o.deferUpload || !o.upload {
+		t.Fatalf("json mode must defer the upload (R-978): got deferUpload=%v upload=%v", o.deferUpload, o.upload)
+	}
+	// Text mode (an operator at a terminal) keeps the immediate upload; no --upload means no upload at all.
+	if o := escrowOptsFromCLI("", false, false, true, "", false); o.deferUpload {
+		t.Fatal("text mode must upload at once")
+	}
+	if o := escrowOptsFromCLI("", false, false, false, "", true); o.deferUpload {
+		t.Fatal("without --upload there is nothing to defer")
+	}
+}
+
+func TestDeliverEscrowPackage_JSONModeSendsNothing(t *testing.T) {
+	cfg := config.Config{}
+	cfg.Hub.URL, cfg.Hub.HostID, cfg.Hub.APIKey = "https://hub.invalid", "host-1", "k"
+	res := escrow.CreateResult{Blob: []byte("ciphertext"), KeyFingerprint: "f2:87", Posture: escrow.PostureZeroKnowledge, IdentityBlob: []byte("id")}
+
+	for _, jsonMode := range []bool{true, false} {
+		calls := 0
+		var sent []byte
+		put := func(_ context.Context, _ config.Config, body []byte) error { calls++; sent = body; return nil }
+		o := escrowOptsFromCLI("", false, false, true, "", jsonMode)
+		pending, uploaded, err := deliverEscrowPackage(context.Background(), cfg, res, "abc", o.deferUpload, put)
+		if err != nil {
+			t.Fatal(err)
+		}
+		if jsonMode {
+			if calls != 0 || uploaded {
+				t.Fatalf("json mode: put called %d times (uploaded=%v), want 0 — the package must wait for the claim", calls, uploaded)
+			}
+			var m map[string]any
+			if json.Unmarshal(pending, &m) != nil || m["blob_b64"] == "" || m["identity_blob_b64"] == "" || m["restic_pw_sha256"] != "abc" {
+				t.Fatalf("json mode: the pending package is not the hub upload body: %s", pending)
+			}
+		} else if calls != 1 || !uploaded || pending != nil || len(sent) == 0 {
+			t.Fatalf("text mode: put called %d times, uploaded=%v, pending=%q — want one immediate upload", calls, uploaded, pending)
+		}
+	}
+
+	// A hub that is not configured fails in both modes (as the pre-R-978 upload did).
+	if _, _, err := deliverEscrowPackage(context.Background(), config.Config{}, res, "", true, nil); err == nil {
+		t.Fatal("an unconfigured hub must fail the ceremony even when the upload is deferred")
+	}
+}
+
+// R-978 wiring (the built-but-never-wired class): runDaemon's localapi.EscrowCeremonyConfig carries an
+// Upload that calls putEscrowBody. Without it every claim of a new package answers 503 and no customer
+// can finish a ceremony.
+//
+// COMPANION RED-PROOF (observed 2026-10-18): delete the `Upload:` line in runDaemon → this fails with
+// "runDaemon's EscrowCeremonyConfig has no Upload calling putEscrowBody". Restored.
+func TestR978_CeremonyUploadIsWiredIntoTheDaemon(t *testing.T) {
+	_, f := parseMain(t)
+	if !callsWithin(f, "main")["runDaemon"] {
+		t.Fatal("main → runDaemon is broken — the path this test asserts is not the live one")
+	}
+	wired := false
+	for _, d := range f.Decls {
+		fd, ok := d.(*ast.FuncDecl)
+		if !ok || fd.Name == nil || fd.Name.Name != "runDaemon" || fd.Body == nil {
+			continue
+		}
+		ast.Inspect(fd.Body, func(n ast.Node) bool {
+			cl, ok := n.(*ast.CompositeLit)
+			if !ok {
+				return true
+			}
+			sel, ok := cl.Type.(*ast.SelectorExpr)
+			if !ok || sel.Sel.Name != "EscrowCeremonyConfig" {
+				return true
+			}
+			for _, el := range cl.Elts {
+				if kv, ok := el.(*ast.KeyValueExpr); ok {
+					if k, ok := kv.Key.(*ast.Ident); ok && k.Name == "Upload" {
+						ast.Inspect(kv.Value, func(n ast.Node) bool {
+							if ce, ok := n.(*ast.CallExpr); ok {
+								if id, ok := ce.Fun.(*ast.Ident); ok && id.Name == "putEscrowBody" {
+									wired = true
+								}
+							}
+							return true
+						})
+					}
+				}
+			}
+			return true
+		})
+	}
+	if !wired {
+		t.Fatal("runDaemon's EscrowCeremonyConfig has no Upload calling putEscrowBody")
+	}
+}

cmd/felhom-agent/main.go

@@ -194,7 +194,7 @@ func main() {
 	flag.StringVar(&pbsStorage, "storage", "", "for --selftest=escrow-create: the pbs storage whose key to escrow (default: escrow.pbs_storage_id)")
 	flag.BoolVar(&paperkey, "paperkey", false, "for --selftest=escrow-create: ALSO emit the raw-key paperkey (opt-in (a); single-factor, unrevocable)")
 	flag.BoolVar(&offline, "offline", false, "for --selftest=escrow-create: ALSO emit the R-wrapped offline copy to print (opt-in (b))")
-	flag.BoolVar(&upload, "upload", false, "for --selftest=escrow-create: upload the opaque blob to the hub")
+	flag.BoolVar(&upload, "upload", false, "for --selftest=escrow-create: upload the opaque blob to the hub (with --output=json: hand the package back in pending_upload — the agent daemon uploads it when the code is claimed, R-978)")
 	flag.StringVar(&blobPath, "blob", "", "for --selftest=escrow-consume: path to the R-wrapped escrow blob file")
 	flag.StringVar(&expectedFP, "fingerprint", "", "for --selftest=escrow-consume: the EXPECTED key fingerprint (the gate target)")
 	flag.StringVar(&keyDest, "keydest", "", "for --selftest=escrow-consume: where to install the recovered key (0600)")
@@ -404,6 +404,31 @@ func (r *gateRemounter) Remount(ctx context.Context, t storage.KnownTarget) {
 // poolReadStatus probes the felhom-pool membership read (GET /pools/felhom) the stale-lock reaper's
 // ownership scoping depends on (v0.62.0, audit A1). Needs `Pool.Audit` at `/pool/felhom` — granted
 // by host-install v1.9.0; on an older ACL this reports degraded and the reaper fail-safes (skips).
+// backupCapStatusName is the host-report status the R-973 cadence cap raises.
+const backupCapStatusName = "localapi:backup-cadence-cap"
+
+// backupCapStatus turns the local-API backup cadence cap's newest refusal (R-973) into a CRITICAL
+// capability status: degraded for 24 h after a refusal, so the hub raises its operator-only
+// agent_capability_degraded event (no hub change). The Reason names the guest, the tier and the time, and
+// says what to do. nil source (local API off) → ok.
+func backupCapStatus(refusal func() (string, int, time.Time, bool)) capability.Status {
+	s := capability.Status{
+		Name:     backupCapStatusName,
+		Feature:  "whole-guest restore points: a guest asked for more whole-guest backups than its schedule allows, and the extra ones were refused (R-973)",
+		Critical: true,
+		Status:   capability.StatusOK,
+	}
+	if refusal == nil {
+		return s
+	}
+	if target, vmid, at, ok := refusal(); ok {
+		s.Status = capability.StatusDegraded
+		s.Reason = fmt.Sprintf("refused a whole-guest backup of guest %d on target %q at %s — more than the schedule allows; the guest's controller may be misbehaving or compromised: check the guest",
+			vmid, target, at.UTC().Format(time.RFC3339))
+	}
+	return s
+}
+
 func poolReadStatus(ctx context.Context, px *proxmox.Client) capability.Status {
 	s := capability.Status{
 		Name:     "pve:pool-read",
@@ -830,8 +855,13 @@ func runDaemon(cfg config.Config, logger *slog.Logger, logRing *applog.Ring) int
 		run: (&proxmox.ExecRunner{Mode: proxmox.RunnerMode(cfg.Privileged.Mode)}).Run,
 		log: logger,
 	}
+	// R-973: the local-API backup cadence cap's refusals ride the report as a CRITICAL status, so the hub's
+	// existing agent_capability_degraded event (operator-only) tells the operator. Late-bound like
+	// drConfigured: the local API server is built further down, before the loop starts.
+	var backupCapRefusal func() (string, int, time.Time, bool)
 	probeAll := func(ctx context.Context) []capability.Status {
 		out := append(capProber.Probe(ctx), poolReadStatus(ctx, px))
+		out = append(out, backupCapStatus(backupCapRefusal))
 		return append(out, storeGrantStatuses(ctx, px, cfg, grantRepairer)...)
 	}
 	// (The startup self-check log runs AFTER the pbsdr manager is wired below, so its snapshot
@@ -1197,6 +1227,9 @@ func runDaemon(cfg config.Config, logger *slog.Logger, logRing *applog.Ring) int
 			return cfg.Escrow.PBSStorageID
 		},
 		HubConfigured: cfg.Hub.URL != "" && cfg.Hub.HostID != "" && cfg.Hub.APIKey != "",
+		// R-978: the ceremony hands its package back; the claim uploads it with the same PUT the root
+		// subprocess used, under the same per-host key, before the code is handed out.
+		Upload: func(ctx context.Context, pkg []byte) error { return putEscrowBody(ctx, cfg, pkg) },
 		DRConfigured: func() bool {
 			if drConfigured != nil {
 				return drConfigured()
@@ -1206,6 +1239,7 @@ func runDaemon(cfg config.Config, logger *slog.Logger, logRing *applog.Ring) int
 	}
 	localSrv := buildLocalAPIServer(cfg, px, backupStore, heavyOps, observer, driveKnown, hostOps, gate, collector, client, intentRec, guestBindStore, formatJobStore, logRing, escrowCeremonyCfg, logger, &localTokens)
 	if localSrv != nil {
+		backupCapRefusal = localSrv.BackupCapRefusal // R-973
 		internalPlaceExec.SetEnsure(localSrv.EnsureInternalPlace)
 		// R-998 (doc 14): POST /host/poweroff — guest shutdown, then host power-off; guards in hostpoweroff.go.
 		pwMode := proxmox.RunnerMode(cfg.Privileged.Mode)
@@ -2826,6 +2860,20 @@ type escrowCeremonyOpts struct {
 	offline            bool
 	upload             bool
 	identityBundlePath string
+	// deferUpload (R-978): with upload, build the hub upload body and hand it back in
+	// CeremonyOutput.PendingUpload instead of sending it. Set for json mode — the controller-driven
+	// ceremony, whose daemon uploads the package only when the customer claims the new code.
+	deferUpload bool
+}
+
+// escrowOptsFromCLI maps the escrow-create flags onto the ceremony options. json mode defers the upload
+// (R-978, pinned by TestEscrowOptsFromCLI_JSONModeDefersUpload); text mode — an operator at a terminal,
+// who sees R at once — keeps the immediate upload.
+func escrowOptsFromCLI(storage string, paperkey, offline, upload bool, identityBundlePath string, jsonMode bool) escrowCeremonyOpts {
+	return escrowCeremonyOpts{
+		storage: storage, paperkey: paperkey, offline: offline, upload: upload,
+		identityBundlePath: identityBundlePath, deferUpload: upload && jsonMode,
+	}
 }
 
 // escrowCeremonyOutcome is the shared core's result. R is the ONLY secret; Sum mirrors the
@@ -2976,16 +3024,36 @@ func escrowCeremony(ctx context.Context, cfg config.Config, logger *slog.Logger,
 		ResticPwSealed:    resticStaged,
 	}
 	if opts.upload {
-		if err := uploadEscrowBlob(ctx, cfg, res, resticPwSHA256); err != nil {
+		pending, uploaded, err := deliverEscrowPackage(ctx, cfg, res, resticPwSHA256, opts.deferUpload, putEscrowBody)
+		if err != nil {
 			// R is minted and the blob self-verified — only the hub leg failed. kind "upload" lets
 			// the text shell keep the pre-extraction order (R surfaced, THEN the failure).
 			return out, &escrowCeremonyErr{kind: "upload", err: err}
 		}
-		out.Sum.Uploaded = true
+		out.Sum.PendingUpload, out.Sum.Uploaded = pending, uploaded
 	}
 	return out, nil
 }
 
+// deliverEscrowPackage builds the hub upload body and either sends it now (text mode) or returns it for
+// the caller to send later (deferUpload — json mode, R-978: the daemon sends it when the customer claims
+// the code). A hub that is not configured fails in both modes, as before: a package that can never be
+// delivered is not a ceremony. put is the sender seam (putEscrowBody in production).
+func deliverEscrowPackage(ctx context.Context, cfg config.Config, res escrow.CreateResult, resticPwSHA256 string, deferUpload bool,
+	put func(ctx context.Context, cfg config.Config, body []byte) error) (pending []byte, uploaded bool, err error) {
+	if cfg.Hub.URL == "" || cfg.Hub.HostID == "" || cfg.Hub.APIKey == "" {
+		return nil, false, fmt.Errorf("hub not configured (url/host_id/api_key)")
+	}
+	body := escrowUploadBody(res, resticPwSHA256)
+	if deferUpload {
+		return body, false, nil
+	}
+	if err := put(ctx, cfg, body); err != nil {
+		return nil, false, err
+	}
+	return nil, true, nil
+}
+
 // printEscrowTextBanner prints the text-mode header line (shared by the success and the
 // post-banner failure paths so the pre-extraction output stays byte-identical — including the
 // posture: the historical banner always printed the DEFAULT posture, not the result's).
@@ -3049,10 +3117,7 @@ func runSelftestEscrowCreate(ctx context.Context, cfg config.Config, logger *slo
 		return 2
 	}
 
-	out, cerr := escrowCeremony(ctx, cfg, logger, escrowCeremonyOpts{
+	out, cerr := escrowCeremony(ctx, cfg, logger, escrowOptsFromCLI(storage, paperkey, offline, upload, identityBundlePath, jsonMode))
-		storage: storage, paperkey: paperkey, offline: offline, upload: upload,
-		identityBundlePath: identityBundlePath,
-	})
 	if cerr != nil {
 		switch cerr.kind {
 		case "usage":
@@ -3090,6 +3155,9 @@ func runSelftestEscrowCreate(ctx context.Context, cfg config.Config, logger *slo
 		if out.Sum.Uploaded {
 			fmt.Fprintln(os.Stderr, "  uploaded the opaque blob(s) to the hub (host record); the hub cannot open them")
 		}
+		if len(out.Sum.PendingUpload) > 0 {
+			fmt.Fprintln(os.Stderr, "  package prepared, NOT uploaded: the agent uploads it when the new code is claimed (R-978)")
+		}
 		wire := out.Sum
 		wire.RecoveryCode = out.R
 		if err := json.NewEncoder(os.Stdout).Encode(wire); err != nil {
@@ -3243,13 +3311,10 @@ type escrowUploadRequest struct {
 	ResticPwSHA256 string `json:"restic_pw_sha256,omitempty"`
 }
 
-// uploadEscrowBlob PUTs the opaque blob (and, for 10D, the identity blob) to
+// escrowUploadBody is the hub upload body for one ceremony's result: the opaque blob (and, for 10D, the
-// the hub, authed with the per-host key. The hub stores ciphertext + non-secret fields; no usable
+// identity blob) plus non-secret fields. Ciphertext only — useless without R; no usable secret leaves
-// secret leaves the agent.
+// the agent. putEscrowBody sends it, authed with the per-host key.
-func uploadEscrowBlob(ctx context.Context, cfg config.Config, res escrow.CreateResult, resticPwSHA256 string) error {
+func escrowUploadBody(res escrow.CreateResult, resticPwSHA256 string) []byte {
-	if cfg.Hub.URL == "" || cfg.Hub.HostID == "" || cfg.Hub.APIKey == "" {
-		return fmt.Errorf("hub not configured (url/host_id/api_key)")
-	}
 	upReq := escrowUploadRequest{
 		BlobB64:        base64.StdEncoding.EncodeToString(res.Blob),
 		KeyFingerprint: res.KeyFingerprint,
@@ -3261,6 +3326,15 @@ func uploadEscrowBlob(ctx context.Context, cfg config.Config, res escrow.CreateR
 		upReq.IdentityBlobB64 = base64.StdEncoding.EncodeToString(res.IdentityBlob)
 	}
 	body, _ := json.Marshal(upReq)
+	return body
+}
+
+// putEscrowBody PUTs an escrow upload body to this host's hub record (the per-host key). Used by the
+// text-mode ceremony at once, and by the daemon when the customer claims the code (R-978).
+func putEscrowBody(ctx context.Context, cfg config.Config, body []byte) error {
+	if cfg.Hub.URL == "" || cfg.Hub.HostID == "" || cfg.Hub.APIKey == "" {
+		return fmt.Errorf("hub not configured (url/host_id/api_key)")
+	}
 	url := strings.TrimRight(cfg.Hub.URL, "/") + "/api/v1/hosts/" + cfg.Hub.HostID + "/escrow"
 	req, err := http.NewRequestWithContext(ctx, http.MethodPut, url, bytes.NewReader(body))
 	if err != nil {

internal/backup/backup_state.go

@@ -39,13 +39,25 @@ type savedSuccess struct {
 	target string
 	vmid   int
 	at     time.Time
+	// recent (R-973) holds the start times of the newest successes (at most maxRecentSuccesses, newest
+	// last) — what the local-API backup cadence cap counts, so an agent restart does not reset the cap.
+	recent []time.Time
+	// refusedAt (R-973) is the newest time the cap refused a backup of this guest on this tier.
+	refusedAt time.Time
 }
 
-// backupSuccessJSON is one entry on disk.
+// maxRecentSuccesses bounds the per-tier success history kept for the cadence cap (R-973). The cap
+// allows at most 3 runs in a window, so 8 is ample.
+const maxRecentSuccesses = 8
+
+// backupSuccessJSON is one entry on disk. Recent and LastRefusedAt are additive (R-973): a file written
+// before them reads as "no history", and an older agent ignores them.
 type backupSuccessJSON struct {
-	Target    string `json:"target"`
+	Target        string   `json:"target"`
-	VMID      int    `json:"vmid"`
+	VMID          int      `json:"vmid"`
-	StartedAt string `json:"started_at"`
+	StartedAt     string   `json:"started_at"`
+	Recent        []string `json:"recent,omitempty"`
+	LastRefusedAt string   `json:"last_refused_at,omitempty"`
 }
 
 func backupStateKey(target string, vmid int) string { return target + "/" + strconv.Itoa(vmid) }
@@ -67,7 +79,19 @@ func NewBackupSuccessState(path string) *BackupSuccessState {
 		if perr != nil {
 			continue // one unreadable entry must not lose the others
 		}
-		s.last[backupStateKey(e.Target, e.VMID)] = savedSuccess{target: e.Target, vmid: e.VMID, at: t.UTC()}
+		ss := savedSuccess{target: e.Target, vmid: e.VMID, at: t.UTC()}
+		for _, r := range e.Recent {
+			if rt, rerr := time.Parse(time.RFC3339, r); rerr == nil {
+				ss.recent = append(ss.recent, rt.UTC())
+			}
+		}
+		if len(ss.recent) == 0 {
+			ss.recent = []time.Time{ss.at} // a pre-R-973 file: its one known success counts
+		}
+		if rt, rerr := time.Parse(time.RFC3339, e.LastRefusedAt); rerr == nil {
+			ss.refusedAt = rt.UTC()
+		}
+		s.last[backupStateKey(e.Target, e.VMID)] = ss
 	}
 	return s
 }
@@ -85,13 +109,72 @@ func (s *BackupSuccessState) RecordBackupSuccess(target string, b hub.Backup) er
 	s.mu.Lock()
 	defer s.mu.Unlock()
 	k := backupStateKey(target, b.VMID)
-	if old, ok := s.last[k]; ok && !t.After(old.at) {
+	old, had := s.last[k]
+	if had && !t.After(old.at) {
 		return nil
 	}
-	s.last[k] = savedSuccess{target: target, vmid: b.VMID, at: t.UTC()}
+	ss := savedSuccess{target: target, vmid: b.VMID, at: t.UTC(), refusedAt: old.refusedAt}
+	ss.recent = append(append([]time.Time(nil), old.recent...), t.UTC())
+	if len(ss.recent) > maxRecentSuccesses {
+		ss.recent = ss.recent[len(ss.recent)-maxRecentSuccesses:]
+	}
+	s.last[k] = ss
 	return s.saveLocked()
 }
 
+// RecentBackupSuccesses returns the saved success start times for this tier and guest that are not
+// older than since (R-973 — the cadence cap's restart-proof count).
+func (s *BackupSuccessState) RecentBackupSuccesses(target string, vmid int, since time.Time) []time.Time {
+	if s == nil {
+		return nil
+	}
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	var out []time.Time
+	for _, t := range s.last[backupStateKey(target, vmid)].recent {
+		if !t.Before(since) {
+			out = append(out, t)
+		}
+	}
+	return out
+}
+
+// RecordBackupRefusal saves that the cadence cap refused a backup of this guest on this tier at at
+// (R-973), so the operator signal it raises survives an agent restart.
+func (s *BackupSuccessState) RecordBackupRefusal(target string, vmid int, at time.Time) error {
+	if s == nil {
+		return nil
+	}
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	k := backupStateKey(target, vmid)
+	ss, ok := s.last[k]
+	if !ok {
+		return nil // nothing to cap against without a success; the cap never refuses then
+	}
+	if !at.After(ss.refusedAt) {
+		return nil
+	}
+	ss.refusedAt = at.UTC()
+	s.last[k] = ss
+	return s.saveLocked()
+}
+
+// LastBackupRefusal returns the newest saved cadence-cap refusal across every tier and guest.
+func (s *BackupSuccessState) LastBackupRefusal() (target string, vmid int, at time.Time, ok bool) {
+	if s == nil {
+		return "", 0, time.Time{}, false
+	}
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	for _, e := range s.last {
+		if !e.refusedAt.IsZero() && e.refusedAt.After(at) {
+			target, vmid, at, ok = e.target, e.vmid, e.refusedAt, true
+		}
+	}
+	return target, vmid, at, ok
+}
+
 // LastKnownSuccess returns the newest saved success for this tier and guest (ok=false = none on file).
 func (s *BackupSuccessState) LastKnownSuccess(target string, vmid int) (time.Time, bool) {
 	if s == nil {
@@ -129,7 +212,14 @@ func (s *BackupSuccessState) KnownBackupSuccesses() []hub.Backup {
 func (s *BackupSuccessState) saveLocked() error {
 	entries := make([]backupSuccessJSON, 0, len(s.last))
 	for _, e := range s.last {
-		entries = append(entries, backupSuccessJSON{Target: e.target, VMID: e.vmid, StartedAt: e.at.Format(time.RFC3339)})
+		j := backupSuccessJSON{Target: e.target, VMID: e.vmid, StartedAt: e.at.Format(time.RFC3339)}
+		for _, r := range e.recent {
+			j.Recent = append(j.Recent, r.Format(time.RFC3339))
+		}
+		if !e.refusedAt.IsZero() {
+			j.LastRefusedAt = e.refusedAt.Format(time.RFC3339)
+		}
+		entries = append(entries, j)
 	}
 	// Deterministic file content (Go's map order is random).
 	sort.Slice(entries, func(i, j int) bool {

internal/backup/backup_state_test.go

@@ -74,3 +74,40 @@ func TestBackupSuccessState_KnownBackupSuccessesSurviveAReopen(t *testing.T) {
 		t.Fatal("uncovered_volumes must marshal as [], not null")
 	}
 }
+
+// R-973: the cadence cap's history and refusals survive a reopen; the history is bounded; a file
+// written before R-973 counts its one known success (never "no history" for a tier that has a copy).
+func TestBackupSuccessState_R973_RecentHistoryAndRefusalSurviveAReopen(t *testing.T) {
+	path := filepath.Join(t.TempDir(), "backup-success-state.json")
+	s := NewBackupSuccessState(path)
+	base := time.Date(2026, 10, 18, 2, 30, 0, 0, time.UTC)
+	for i := 0; i < 10; i++ {
+		at := base.Add(time.Duration(i) * time.Hour)
+		if err := s.RecordBackupSuccess("local", hub.Backup{VMID: 9201, Success: true, StartedAt: at.Format(time.RFC3339)}); err != nil {
+			t.Fatal(err)
+		}
+	}
+	refused := base.Add(11 * time.Hour)
+	if err := s.RecordBackupRefusal("local", 9201, refused); err != nil {
+		t.Fatal(err)
+	}
+	r := NewBackupSuccessState(path)
+	all := r.RecentBackupSuccesses("local", 9201, time.Time{})
+	if len(all) != maxRecentSuccesses || !all[len(all)-1].Equal(base.Add(9*time.Hour)) {
+		t.Fatalf("recent history after a reopen: %v (want the newest %d, ending at +9h)", all, maxRecentSuccesses)
+	}
+	if got := r.RecentBackupSuccesses("local", 9201, base.Add(7*time.Hour)); len(got) != 3 {
+		t.Fatalf("successes since +7h: %d, want 3", len(got))
+	}
+	if tg, vm, at, ok := r.LastBackupRefusal(); !ok || tg != "local" || vm != 9201 || !at.Equal(refused) {
+		t.Fatalf("refusal after a reopen: %q %d %v %v", tg, vm, at, ok)
+	}
+
+	old := filepath.Join(t.TempDir(), "old.json")
+	if err := os.WriteFile(old, []byte(`[{"target":"felhom-pbs","vmid":9201,"started_at":"2026-10-17T03:00:00Z"}]`), 0o600); err != nil {
+		t.Fatal(err)
+	}
+	if got := NewBackupSuccessState(old).RecentBackupSuccesses("felhom-pbs", 9201, time.Time{}); len(got) != 1 {
+		t.Fatalf("a pre-R-973 file must count its one success: %v", got)
+	}
+}

internal/backup/r1012_thincheck_test.go

@@ -10,8 +10,9 @@ import (
 )
 
 // R-1012: a snapshot backup that fails carries the orphan finding in its error — the text the controller mails to
-// the operator (`whole_guest_backup_failed`). The check runs once before the run and once, FRESH, after the failure.
+// the operator (`whole_guest_backup_failed`). The check runs ONCE, after the failure — never before the run.
-// RED-PROOF (P1/redproof-thincheck.txt): with explainFailure's append removed this test fails.
+// RED-PROOF (P1/redproof-thincheck.txt): with explainFailure's append removed this test fails. RED-PROOF (2026-10-17
+// night, felhom-agent REPORT.md): with the v0.159.0 pre-run check restored, calls = 2 and this test fails.
 func TestBackup_SnapshotFailure_NamesTheOrphan(t *testing.T) {
 	const why = "probable cause: thin pool pve/data holds device 4 that LVM does not know"
 	for name, api := range map[string]*fakeBackupAPI{
@@ -29,15 +30,17 @@ func TestBackup_SnapshotFailure_NamesTheOrphan(t *testing.T) {
 			if !strings.Contains(rec.Error, why) {
 				t.Fatalf("the failure does not name the orphan: %q", rec.Error)
 			}
-			if calls != 2 {
+			if calls != 1 {
-				t.Fatalf("thin check ran %d times, want 2 (before the run and after the failure)", calls)
+				t.Fatalf("thin check ran %d times, want 1 (after the failure only)", calls)
 			}
 		})
 	}
 }
 
-// A successful backup keeps a clean record (the pre-check only logs); a stop-mode backup takes no snapshot, so the
+// A successful backup never runs the check (it takes a metadata snapshot of the live pool, and nobody needs the answer)
-// check never runs; an unwired runner behaves exactly as before.
+// and keeps a clean record; a stop-mode backup takes no snapshot, so the check never runs; an unwired runner behaves
+// exactly as before. RED-PROOF (2026-10-17 night): with the v0.159.0 pre-run check restored, the success leg fails
+// with calls=1.
 func TestBackup_ThinCheck_SilentWhenNotApplicable(t *testing.T) {
 	ok := &fakeBackupAPI{vzdumpUPID: "UPID:x",
 		content: []proxmox.StorageContent{{VolID: "local:backup/vzdump-lxc-9201-x", Content: "backup", VMID: 9201, Size: 100, CTime: 1}}}
@@ -45,8 +48,8 @@ func TestBackup_ThinCheck_SilentWhenNotApplicable(t *testing.T) {
 	calls := 0
 	r.SetThinCheck(func(context.Context, int) string { calls++; return "probable cause: x" })
 	rec, err := r.Backup(context.Background(), 9201)
-	if err != nil || !rec.Success || rec.Error != "" || calls != 1 {
+	if err != nil || !rec.Success || rec.Error != "" || calls != 0 {
-		t.Fatalf("success: err=%v rec=%+v calls=%d (want one pre-check, clean record)", err, rec, calls)
+		t.Fatalf("success: err=%v rec=%+v calls=%d (want no check on a successful backup, clean record)", err, rec, calls)
 	}
 
 	stop := NewBackupRunner(&fakeBackupAPI{vzdumpUPID: "UPID:x", waitErr: errors.New("boom")}, "local", proxmox.ModeStop, "", "", quiet())

internal/backup/runner.go

@@ -62,8 +62,9 @@ type BackupRunner struct {
 	// this target, so a leftover from a power cut cannot hold the space this backup needs. nil = nothing.
 	beforeBackup func(ctx context.Context, storage string, vmid int)
 	// thinCheck (R-1012) names an orphan thin device in the guest's pools in plain words ("" = none / not checked). It
-	// runs before a snapshot-mode backup (a WARN in the log) and again when the vzdump fails, so the failure the
+	// runs ONLY when a snapshot-mode vzdump has failed (start or task), so the failure the controller mails to the
-	// controller mails to the operator says why. It never removes anything. nil = not wired.
+	// operator says why; a successful backup never runs it (it takes a metadata snapshot of the live pool). It never
+	// removes anything. nil = not wired.
 	thinCheck func(ctx context.Context, vmid int) string
 	logger    *slog.Logger
 	now       func() time.Time
@@ -117,8 +118,8 @@ func (r *BackupRunner) SetBeforeBackup(fn func(ctx context.Context, storage stri
 // SetThinCheck wires the R-1012 orphan check (see BackupRunner.thinCheck).
 func (r *BackupRunner) SetThinCheck(fn func(ctx context.Context, vmid int) string) { r.thinCheck = fn }
 
-// explainFailure appends the orphan finding to a failed snapshot-mode backup's error. A FRESH check (not the one from
+// explainFailure appends the orphan finding to a failed snapshot-mode backup's error — the only place the check runs, so
-// before the run): the failure itself may be what made the orphan. Bounded, so a stuck helper cannot hold the job.
+// it sees the pool as the failure left it. Bounded, so a stuck helper cannot hold the job.
 func (r *BackupRunner) explainFailure(ctx context.Context, vmid int, rec *hub.Backup) {
 	if r.thinCheck == nil || r.mode != proxmox.ModeSnapshot {
 		return
@@ -221,11 +222,8 @@ func (r *BackupRunner) backup(ctx context.Context, vmid int, onSnapshot func())
 		return rec, fmt.Errorf("backup: %s", rec.Error)
 	}
 
-	if r.thinCheck != nil && r.mode == proxmox.ModeSnapshot {
+	// R-1012: no orphan check here. It takes a metadata snapshot of the live pool, and its answer is only needed when the
-		if why := r.thinCheck(ctx, vmid); why != "" {
+	// snapshot step has failed — explainFailure runs it then (pinned by TestBackup_ThinCheck_SilentWhenNotApplicable).
-			r.logger.Warn("backup: an orphan thin device may make this snapshot backup fail (R-1012)", "vmid", vmid, "finding", why)
-		}
-	}
 	upid, err := r.api.Vzdump(ctx, proxmox.VzdumpOptions{
 		VMID: vmid, Storage: r.target, Mode: r.mode, Notes: r.notes,
 		PruneBackups: r.localPruneSpec(ctx), // local target → keep-last=N; PBS/unknown → "" (no prune)

internal/capability/manifest.go

@@ -91,7 +91,7 @@ var manifest = []Capability{
 	{"disk-mkfs-xfs", "guarded format (xfs)", "/usr/local/sbin/felhom-mkfs-guarded", []string{"/dev/sda", "xfs"}, false, ""},
 	{"disk-smart", "disk SMART health read", "/usr/sbin/smartctl", []string{"-a", "-j", "/dev/sda"}, false, ""},
 	{"disk-lvs", "thin-pool usage read", "/usr/sbin/lvs", []string{"--reportformat", "json", "--units", "b", "-o", "lv_name,data_percent,metadata_percent", "--", "pve/data"}, false, ""},
-	{"thin-orphan-check", "orphan thin device check around a whole-guest backup (read-only, R-1012)", "/usr/local/sbin/felhom-thin-check", []string{"pve", "data"}, false, ""},
+	{"thin-orphan-check", "orphan thin device check after a failed whole-guest backup (read-only, R-1012)", "/usr/local/sbin/felhom-thin-check", []string{"pve", "data"}, false, ""},
 
 	// ---- Storage mount units (watchdog re-mount) ----
 	{"mount-unit-install", "fs-UUID mount unit install (root content check, R-861)", "/usr/local/sbin/felhom-priv-apply", []string{"unit", "mnt-felhom\\x2dx.mount"}, false, ""},

internal/escrow/ceremony.go

@@ -1,5 +1,7 @@
 package escrow
 
+import "encoding/json"
+
 // Controller-driven ceremony contract (v0.88.0, TASK 2026-07-13; mechanics validated by
 // felhom.eu/documentation/audits/SPIKE-controller-escrow-2026-07-13.md). The agent's local API
 // re-invokes the agent binary as root via `sudo -n` with ONE fixed argument vector; sudoers
@@ -43,7 +45,15 @@ type CeremonyOutput struct {
 	IdentityBlobBytes int     `json:"identity_blob_bytes"`
 	ResticPwSealed    bool    `json:"restic_pw_sealed"`
 	Uploaded          bool    `json:"uploaded"`
+	// PendingUpload (version 2, R-978) is the hub upload body — the R-wrapped blobs (ciphertext), the key
+	// fingerprint and the sealed password's hash — that json mode hands back INSTEAD of uploading. The agent
+	// daemon uploads it only when the customer claims the new recovery code, so a ceremony that is started
+	// and abandoned never replaces the package the customer's previous code opens. Not a secret without R,
+	// but never logged. Uploaded is false whenever this is set.
+	PendingUpload json.RawMessage `json:"pending_upload,omitempty"`
 }
 
-// CeremonyOutputVersion is the current CeremonyOutput.Version value.
+// CeremonyOutputVersion is the current CeremonyOutput.Version value. 2 (R-978): json mode no longer
-const CeremonyOutputVersion = 1
+// uploads; it returns PendingUpload. The daemon refuses any other version (a binary/daemon mismatch
+// fails the ceremony instead of losing a package).
+const CeremonyOutputVersion = 2

internal/localapi/backup_cap.go

@@ -0,0 +1,146 @@
+package localapi
+
+import (
+	"fmt"
+	"net/http"
+	"sort"
+	"time"
+
+	"gitea.dooplex.hu/admin/felhom-agent/internal/hub"
+)
+
+// The whole-guest backup cadence cap (R-973, security review 2026-10-12).
+//
+// POST /backup is reachable from inside the guest, and every successful run prunes the tier
+// (`--prune-backups keep-last=N` on a local tier; ep0's server-side keep-last 2 on the off-site tier). So
+// a compromised controller calling it in a loop pushed every pre-compromise whole-guest restore point out
+// within minutes — the threat `07` §8 row 10 counts the off-site tier as the answer to. The agent now
+// refuses a run that the schedule in `07` §6 would never ask for:
+//
+//   - a tier whose cadence is at most 24 h (the local tier): at most 3 successful runs in any rolling
+//     24 h. The schedule is one night run; a „Mentés most" press makes a LOCAL copy and, since R-899, the
+//     night still takes its own after it, which can land inside the same 24 h as the previous night (the
+//     window opens anywhere in [W+2h, W+6h)). So a normal day needs at most 3: two nights plus one press.
+//     A second press inside 24 h is refused — the margin is spent on the night, which must never be refused
+//     because of the household's presses.
+//   - a longer tier (the 7-day off-site tier): at most 1 successful run per max(cadence − 24 h, cadence/2)
+//     — 6 days for 7 days. The controller asks /backup/due first and runs only when the newest copy is older
+//     than the cadence, and the button never targets this tier (`07` §6, controller v0.301.0), so a
+//     legitimate run never comes sooner; 24 h of slack absorbs clock and archive-time differences.
+//
+// Only SUCCESSES count — a failed run prunes nothing and must stay retryable. The count survives an agent
+// restart: it reads the success history saved on disk (LastKnownBackupStore, the R-894 file) as well as
+// this process's own record. A refusal answers 429, is logged at WARN, is saved, and raises the operator
+// signal BackupCapRefusal feeds (a critical capability in the host report → the hub's existing
+// `agent_capability_degraded` event, operator-only). Pinned by TestBackupCap_* (backup_cap_test.go).
+
+// backupCapRefusalHold is how long one refusal keeps the operator signal raised.
+const backupCapRefusalHold = 24 * time.Hour
+
+// backupCadenceCap returns the rolling window and the most successful runs allowed in it for a tier.
+func backupCadenceCap(cadence time.Duration) (window time.Duration, max int) {
+	if cadence <= 0 {
+		cadence = defaultBackupCadence
+	}
+	if cadence <= 24*time.Hour {
+		return 24 * time.Hour, 3
+	}
+	w := cadence - 24*time.Hour
+	if w < cadence/2 {
+		w = cadence / 2
+	}
+	return w, 1
+}
+
+// recentCapRuns returns the start times of the successful runs on this tier for this guest inside the
+// window ending now — this process's record and the saved one, each counted once. Caller holds no lock.
+func (s *Server) recentCapRuns(key backupJobKey, since time.Time) []time.Time {
+	seen := map[int64]bool{}
+	var out []time.Time
+	add := func(t time.Time) {
+		if t.Before(since) || seen[t.Unix()] {
+			return
+		}
+		seen[t.Unix()] = true
+		out = append(out, t)
+	}
+	s.capMu.Lock()
+	for _, t := range s.capRuns[key] {
+		add(t)
+	}
+	s.capMu.Unlock()
+	if s.lastKnown != nil {
+		for _, t := range s.lastKnown.RecentBackupSuccesses(key.target, key.vmid, since) {
+			add(t)
+		}
+	}
+	sort.Slice(out, func(i, j int) bool { return out[i].Before(out[j]) })
+	return out
+}
+
+// recordCapRun notes a successful run for the cap (the saved half is written by RecordBackupSuccess).
+func (s *Server) recordCapRun(key backupJobKey, b hub.Backup) {
+	t, err := time.Parse(time.RFC3339, b.StartedAt)
+	if err != nil {
+		t = s.now()
+	}
+	s.capMu.Lock()
+	defer s.capMu.Unlock()
+	if s.capRuns == nil {
+		s.capRuns = map[backupJobKey][]time.Time{}
+	}
+	runs := append(s.capRuns[key], t.UTC())
+	if len(runs) > 8 {
+		runs = runs[len(runs)-8:]
+	}
+	s.capRuns[key] = runs
+}
+
+// capRefuses applies the cap to a new run request. It writes the 429 and returns true when refused.
+func (s *Server) capRefuses(w http.ResponseWriter, key backupJobKey, tier BackupTier) bool {
+	window, max := backupCadenceCap(tier.Cadence)
+	now := s.now()
+	runs := s.recentCapRuns(key, now.Add(-window))
+	if len(runs) < max {
+		return false
+	}
+	next := runs[len(runs)-max].Add(window)
+	s.capMu.Lock()
+	s.capRefusal = capRefusal{target: key.target, vmid: key.vmid, at: now.UTC()}
+	s.capMu.Unlock()
+	if s.lastKnown != nil {
+		if err := s.lastKnown.RecordBackupRefusal(key.target, key.vmid, now); err != nil {
+			s.logger.Warn("local-api: could not save the backup refusal on disk (R-973)", "vmid", key.vmid, "target", key.target, "err", err)
+		}
+	}
+	s.logger.Warn("local-api: backup REFUSED — more whole-guest backups than the schedule allows (R-973 cadence cap; check the guest)",
+		"vmid", key.vmid, "target", key.target, "runs_in_window", len(runs), "allowed", max,
+		"window_h", int(window.Hours()), "next_allowed", next.UTC().Format(time.RFC3339))
+	writeStatus(w, http.StatusTooManyRequests, false, nil, fmt.Sprintf(
+		"too many whole-guest backups on target %q: %d successful in the last %d h, the schedule allows %d — refused (the next one is allowed after %s)",
+		key.target, len(runs), int(window.Hours()), max, next.UTC().Format(time.RFC3339)))
+	return true
+}
+
+type capRefusal struct {
+	target string
+	vmid   int
+	at     time.Time
+}
+
+// BackupCapRefusal reports the newest cadence-cap refusal still inside the 24 h hold (this process's or
+// the saved one) — the source of the operator signal in the host report (R-973). ok=false = none.
+func (s *Server) BackupCapRefusal() (target string, vmid int, at time.Time, ok bool) {
+	s.capMu.Lock()
+	r := s.capRefusal
+	s.capMu.Unlock()
+	if s.lastKnown != nil {
+		if t, v, a, found := s.lastKnown.LastBackupRefusal(); found && a.After(r.at) {
+			r = capRefusal{target: t, vmid: v, at: a}
+		}
+	}
+	if r.at.IsZero() || s.now().Sub(r.at) > backupCapRefusalHold {
+		return "", 0, time.Time{}, false
+	}
+	return r.target, r.vmid, r.at, true
+}

internal/localapi/backup_cap_test.go

@@ -0,0 +1,293 @@
+package localapi
+
+import (
+	"context"
+	"fmt"
+	"io"
+	"log/slog"
+	"net/http"
+	"path/filepath"
+	"sync"
+	"testing"
+	"time"
+
+	"gitea.dooplex.hu/admin/felhom-agent/internal/backup"
+	"gitea.dooplex.hu/admin/felhom-agent/internal/hub"
+)
+
+// R-973 — the whole-guest backup cadence cap. A compromised controller calling POST /backup in a loop
+// pushed out every pre-compromise restore point: each success prunes the tier. These tests model the tier
+// the way Proxmox and ep0 keep it (keep-last N, pruned after each success) and ask what is LEFT.
+
+// keepLastTier is a whole-guest tier: every successful run adds an archive stamped with the clock and
+// prunes to keepLast, as vzdump's --prune-backups keep-last (local) and ep0's keep-last 2 (off-site) do.
+type keepLastTier struct {
+	mu       sync.Mutex
+	clock    *testClock
+	keepLast int
+	archives []time.Time
+	fail     bool
+	runs     int
+}
+
+func (k *keepLastTier) BackupWithSnapshotHook(_ context.Context, vmid int, _ func()) (hub.Backup, error) {
+	k.mu.Lock()
+	defer k.mu.Unlock()
+	k.runs++
+	now := k.clock.Now()
+	if k.fail {
+		return hub.Backup{VMID: vmid, StartedAt: now.Format(time.RFC3339)}, fmt.Errorf("vzdump failed")
+	}
+	k.archives = append(k.archives, now)
+	if len(k.archives) > k.keepLast {
+		k.archives = k.archives[len(k.archives)-k.keepLast:]
+	}
+	return hub.Backup{VMID: vmid, Success: true, Archive: "x", StartedAt: now.Format(time.RFC3339)}, nil
+}
+
+// oldestBefore reports whether the tier still holds an archive made before t (a pre-compromise point).
+func (k *keepLastTier) oldestBefore(t time.Time) bool {
+	k.mu.Lock()
+	defer k.mu.Unlock()
+	return len(k.archives) > 0 && k.archives[0].Before(t)
+}
+
+func (k *keepLastTier) count() (runs, kept int) {
+	k.mu.Lock()
+	defer k.mu.Unlock()
+	return k.runs, len(k.archives)
+}
+
+type testClock struct {
+	mu sync.Mutex
+	t  time.Time
+}
+
+func (c *testClock) Now() time.Time          { c.mu.Lock(); defer c.mu.Unlock(); return c.t }
+func (c *testClock) Advance(d time.Duration) { c.mu.Lock(); c.t = c.t.Add(d); c.mu.Unlock() }
+
+// capServer builds the two-tier box of `07` §6: local 24 h keep-last 3, off-site 7 d keep-last 2, the
+// success history on disk at path (a second capServer on the same path is an agent restart).
+func capServer(t *testing.T, path string, clock *testClock, local, offsite *keepLastTier) *Server {
+	t.Helper()
+	srv, err := NewServer(Options{
+		ListenAddr: "127.0.0.1:0", Guests: &fakeGuests{}, Backups: &fakeBackups{}, Store: &fakeStore{},
+		Storage: fakeStorage{targets: []hub.StorageTarget{{Name: "local"}, {Name: "felhom-pbs"}}},
+		Tokens:  staticTokens{"A": 8200},
+		BackupTiers: []BackupTier{
+			{TargetID: "local", Cadence: 24 * time.Hour, Primary: true, Service: local},
+			{TargetID: "felhom-pbs", Cadence: 7 * 24 * time.Hour, Service: offsite},
+		},
+		LastKnownBackups: backup.NewBackupSuccessState(path),
+		Logger:           slog.New(slog.NewTextHandler(io.Discard, nil)),
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+	srv.baseCtx = context.Background()
+	srv.now = clock.Now
+	return srv
+}
+
+// runBackup POSTs one backup and waits for the job to end. Returns the HTTP status of the POST.
+func runBackup(t *testing.T, srv *Server, target string) int {
+	t.Helper()
+	rr := do(t, srv.Handler(), "POST", "/backup?target="+target, "A", "")
+	if rr.Code == http.StatusAccepted {
+		waitFor(t, func() bool {
+			srv.jobsMu.Lock()
+			defer srv.jobsMu.Unlock()
+			j := srv.jobs[backupJobKey{vmid: 8200, target: target}]
+			return j != nil && !backupInFlight(j.Phase)
+		})
+	}
+	return rr.Code
+}
+
+// The off-site tier — the one `07` §8 row 10 counts on against ransomware. Before the cap ten calls in a
+// row left ep0 holding only the attacker's two copies. Now ONE runs (the tier was due), the other nine
+// are refused, and the newest pre-compromise copy survives; five days of flooding later it still does.
+//
+// COMPANION RED-PROOF (observed 2026-10-18): make capRefuses always return false → this fails with
+// "the off-site tier lost every pre-compromise copy: 10 runs" . Restored.
+func TestBackupCap_OffsiteFloodKeepsAPreCompromiseCopy(t *testing.T) {
+	clock := &testClock{t: testNow}
+	local := &keepLastTier{clock: clock, keepLast: 3}
+	off := &keepLastTier{clock: clock, keepLast: 2}
+	path := filepath.Join(t.TempDir(), "backup-success-state.json")
+	srv := capServer(t, path, clock, local, off)
+
+	// The honest history: two weekly copies, the newest 7 days and 1 hour old (so the tier is due).
+	for _, age := range []time.Duration{14*24*time.Hour + time.Hour, 7*24*time.Hour + time.Hour} {
+		clock.t = testNow.Add(-age)
+		if code := runBackup(t, srv, "felhom-pbs"); code != http.StatusAccepted {
+			t.Fatalf("seeding the honest weekly copy: %d", code)
+		}
+	}
+	clock.t = testNow
+	compromise := testNow
+
+	refused := 0
+	for i := 0; i < 10; i++ {
+		if runBackup(t, srv, "felhom-pbs") == http.StatusTooManyRequests {
+			refused++
+		}
+		clock.Advance(time.Minute)
+	}
+	runs, _ := off.count()
+	if !off.oldestBefore(compromise) {
+		t.Fatalf("the off-site tier lost every pre-compromise copy: %d runs", runs-2)
+	}
+	if runs-2 != 1 || refused != 9 {
+		t.Fatalf("flood: %d runs, %d refused — want 1 run (the tier was due) and 9 refused", runs-2, refused)
+	}
+	for d := 0; d < 5; d++ { // five more days of flooding, every hour
+		for h := 0; h < 24; h++ {
+			clock.Advance(time.Hour)
+			if code := runBackup(t, srv, "felhom-pbs"); code != http.StatusTooManyRequests {
+				t.Fatalf("day %d hour %d: an off-site run was allowed %v after the last one", d, h, clock.Now().Sub(compromise))
+			}
+		}
+	}
+	if !off.oldestBefore(compromise) {
+		t.Fatal("five days of flooding pushed out the pre-compromise off-site copy")
+	}
+}
+
+// The local tier: a flood runs exactly 3 in a day (the schedule's own day: two nights + one press) and
+// every further request is refused. Honest about the limit: with keep-last 3 those 3 replace the 3 kept
+// points — the cap slows a flush from minutes to a day; it cannot keep a point a 3-per-day pace prunes.
+func TestBackupCap_LocalFloodRunsAtMostThreeADay(t *testing.T) {
+	clock := &testClock{t: testNow}
+	local := &keepLastTier{clock: clock, keepLast: 3}
+	srv := capServer(t, filepath.Join(t.TempDir(), "s.json"), clock, local, &keepLastTier{clock: clock, keepLast: 2})
+	refused := 0
+	for i := 0; i < 20; i++ {
+		if runBackup(t, srv, "local") == http.StatusTooManyRequests {
+			refused++
+		}
+		clock.Advance(10 * time.Minute)
+	}
+	if runs, _ := local.count(); runs != 3 || refused != 17 {
+		t.Fatalf("local flood over ~3 h: %d runs, %d refused — want 3 and 17", runs, refused)
+	}
+	clock.Advance(24 * time.Hour) // a day later the window has moved on
+	if code := runBackup(t, srv, "local"); code != http.StatusAccepted {
+		t.Fatalf("a run a day later must be allowed: %d", code)
+	}
+}
+
+// The schedule itself is never refused: a night at W+6h, a press at 10:00, the next night at W+2h (all
+// three inside one rolling 24 h) run; a second press that day is refused; nights keep running after it.
+func TestBackupCap_TheScheduledDayIsNeverRefused(t *testing.T) {
+	day := time.Date(2026, 6, 10, 0, 0, 0, 0, time.UTC)
+	clock := &testClock{t: day.Add(8*time.Hour + 29*time.Minute)} // W=02:30 → W+6h ≈ 08:29
+	local := &keepLastTier{clock: clock, keepLast: 3}
+	srv := capServer(t, filepath.Join(t.TempDir(), "s.json"), clock, local, &keepLastTier{clock: clock, keepLast: 2})
+	steps := []struct {
+		at   time.Time
+		want int
+		what string
+	}{
+		{day.Add(8*time.Hour + 29*time.Minute), http.StatusAccepted, "night 1 at W+6h"},
+		{day.Add(10 * time.Hour), http.StatusAccepted, "a press at 10:00"},
+		{day.Add(24*time.Hour + 4*time.Hour + 30*time.Minute), http.StatusAccepted, "night 2 at W+2h (20 h later)"},
+		{day.Add(24*time.Hour + 7*time.Hour), http.StatusTooManyRequests, "a second press inside the same 24 h"},
+		{day.Add(48*time.Hour + 4*time.Hour + 30*time.Minute), http.StatusAccepted, "night 3"},
+	}
+	for _, s := range steps {
+		clock.t = s.at
+		if got := runBackup(t, srv, "local"); got != s.want {
+			t.Fatalf("%s: got %d, want %d", s.what, got, s.want)
+		}
+	}
+}
+
+// A restart must not reset the cap: the history is read back from disk.
+//
+// COMPANION RED-PROOF (observed 2026-10-18): make recentCapRuns ignore s.lastKnown → this fails with
+// "after a restart the 4th run in a day was allowed (got 202)". Restored.
+func TestBackupCap_SurvivesARestart(t *testing.T) {
+	clock := &testClock{t: testNow}
+	path := filepath.Join(t.TempDir(), "s.json")
+	local := &keepLastTier{clock: clock, keepLast: 3}
+	srv := capServer(t, path, clock, local, &keepLastTier{clock: clock, keepLast: 2})
+	for i := 0; i < 3; i++ {
+		if code := runBackup(t, srv, "local"); code != http.StatusAccepted {
+			t.Fatalf("run %d: %d", i+1, code)
+		}
+		clock.Advance(time.Minute)
+	}
+	srv2 := capServer(t, path, clock, local, &keepLastTier{clock: clock, keepLast: 2}) // the restart
+	if code := runBackup(t, srv2, "local"); code != http.StatusTooManyRequests {
+		t.Fatalf("after a restart the 4th run in a day was allowed (got %d)", code)
+	}
+}
+
+// Failed runs prune nothing and must stay retryable: they do not count.
+func TestBackupCap_FailuresDoNotCount(t *testing.T) {
+	clock := &testClock{t: testNow}
+	local := &keepLastTier{clock: clock, keepLast: 3, fail: true}
+	srv := capServer(t, filepath.Join(t.TempDir(), "s.json"), clock, local, &keepLastTier{clock: clock, keepLast: 2})
+	for i := 0; i < 6; i++ {
+		if code := runBackup(t, srv, "local"); code != http.StatusAccepted {
+			t.Fatalf("failed attempt %d was refused: %d", i+1, code)
+		}
+		clock.Advance(15 * time.Minute)
+	}
+	local.mu.Lock()
+	local.fail = false
+	local.mu.Unlock()
+	if code := runBackup(t, srv, "local"); code != http.StatusAccepted {
+		t.Fatalf("a success after six failures was refused: %d", code)
+	}
+}
+
+// A refusal raises the operator signal for 24 h, and the signal survives a restart.
+func TestBackupCap_RefusalRaisesTheOperatorSignal(t *testing.T) {
+	clock := &testClock{t: testNow}
+	path := filepath.Join(t.TempDir(), "s.json")
+	off := &keepLastTier{clock: clock, keepLast: 2}
+	srv := capServer(t, path, clock, &keepLastTier{clock: clock, keepLast: 3}, off)
+	if _, _, _, ok := srv.BackupCapRefusal(); ok {
+		t.Fatal("a fresh box must not raise the signal")
+	}
+	runBackup(t, srv, "felhom-pbs")
+	clock.Advance(time.Hour)
+	if code := runBackup(t, srv, "felhom-pbs"); code != http.StatusTooManyRequests {
+		t.Fatalf("second off-site run an hour later: %d, want 429", code)
+	}
+	target, vmid, at, ok := srv.BackupCapRefusal()
+	if !ok || target != "felhom-pbs" || vmid != 8200 || !at.Equal(clock.Now()) {
+		t.Fatalf("signal after a refusal: ok=%v target=%q vmid=%d at=%v", ok, target, vmid, at)
+	}
+	clock.Advance(23 * time.Hour)
+	srv2 := capServer(t, path, clock, &keepLastTier{clock: clock, keepLast: 3}, off) // a restart
+	if _, _, _, ok := srv2.BackupCapRefusal(); !ok {
+		t.Fatal("the signal did not survive a restart")
+	}
+	clock.Advance(2 * time.Hour)
+	if _, _, _, ok := srv2.BackupCapRefusal(); ok {
+		t.Fatal("the signal must clear 24 h after the last refusal")
+	}
+}
+
+func TestBackupCadenceCap(t *testing.T) {
+	cases := []struct {
+		cadence time.Duration
+		window  time.Duration
+		max     int
+	}{
+		{0, 24 * time.Hour, 3},
+		{24 * time.Hour, 24 * time.Hour, 3},
+		{12 * time.Hour, 24 * time.Hour, 3},
+		{7 * 24 * time.Hour, 6 * 24 * time.Hour, 1},
+		{36 * time.Hour, 18 * time.Hour, 1},
+		{30 * 24 * time.Hour, 29 * 24 * time.Hour, 1},
+	}
+	for _, c := range cases {
+		if w, m := backupCadenceCap(c.cadence); w != c.window || m != c.max {
+			t.Errorf("cadence %v: got %v/%d, want %v/%d", c.cadence, w, m, c.window, c.max)
+		}
+	}
+}

internal/localapi/escrow_ceremony.go

@@ -24,9 +24,16 @@ import (
 //     be structurally incapable of carrying R).
 //   - One claim, then the holder is zeroed. Unclaimed past the TTL → zeroed + phase
 //     unclaimed_void ("R unclaimed → ceremony void → a re-run supersedes").
-//   - Crash-safety is IN-MEMORY BY DESIGN: an agent restart loses R, which is SAFE (the blob is
+//   - Crash-safety is IN-MEMORY BY DESIGN: an agent restart loses R AND the package it opens,
-//     on the hub; a re-run supersedes it). Status reporting "none" after a restart is the honest
+//     which is SAFE (R-978: the package was never uploaded, so the hub still holds the previous one
-//     answer. No journal, deliberately — persistence is the one property R must never have.
+//     and the customer's previous code still opens it; a re-run starts over). Status reporting "none"
+//     after a restart is the honest answer. No journal, deliberately — persistence is the one
+//     property R must never have.
+//   - R-978: the NEW package reaches the hub only at the claim, and the code is handed out only after
+//     the hub took it. Until then the previous package stays current — so a ceremony that is started
+//     and abandoned (never claimed, superseded, TTL, restart) leaves the customer's written-down code
+//     working. The TTL and a new start both wait while a claim's upload is in flight, so a package is
+//     never stored without its code being handed out by that same claim.
 //   - Nothing from stdout is ever logged; stderr (log-clean, spike §2.1) is tail-captured for
 //     failure diagnostics only.
 
@@ -50,8 +57,16 @@ type EscrowCeremonyConfig struct {
 	// DRConfigured answers "is the DR tier applied on this box?" (pbsdr.Manager.DRConfigured,
 	// late-bound). nil → reported not-applied.
 	DRConfigured func() bool
+	// Upload sends a ceremony's package (CeremonyOutput.PendingUpload, the hub upload body) to this
+	// host's hub record — called by the claim, BEFORE the code is handed out (R-978). nil → a claim of
+	// a pending package is refused (503): a code must never be handed out for a package the hub lacks.
+	Upload func(ctx context.Context, pkg []byte) error
 }
 
+// escrowUploadTimeout bounds the claim-time package upload (one small PUT; the subprocess's whole
+// ceremony incl. its upload was measured at ~2.4 s).
+const escrowUploadTimeout = 30 * time.Second
+
 // ceremonyRunner executes the fixed-argv sudo self-invocation. stdout is SECRET-BEARING until
 // parsed (it carries R inside the JSON object); the caller must zero it. Tests inject canned
 // spike-shaped output; production is runCeremonySubprocess.
@@ -129,6 +144,9 @@ func (s *Server) tryStartEscrowCeremony(job *escrowCeremonyJob) bool {
 	if s.escrowJob != nil && s.escrowJob.Phase == escrowPhaseRunning {
 		return false
 	}
+	if s.escrowClaiming {
+		return false // R-978: a claim is uploading this job's package — its code must reach the claimer
+	}
 	s.wipeEscrowRLocked()
 	cp := *job
 	s.escrowJob = &cp
@@ -137,7 +155,7 @@ func (s *Server) tryStartEscrowCeremony(job *escrowCeremonyJob) bool {
 
 // finishEscrowCeremony records the terminal phase and (on success) arms the one-shot R holder.
 // R is stored as a byte slice so the claim/TTL paths can zero it in place.
-func (s *Server) finishEscrowCeremony(job *escrowCeremonyJob, r string) {
+func (s *Server) finishEscrowCeremony(job *escrowCeremonyJob, r string, pending []byte) {
 	s.escrowMu.Lock()
 	defer s.escrowMu.Unlock()
 	job.UpdatedAt = s.nowFn().UTC().Format(time.RFC3339)
@@ -145,6 +163,7 @@ func (s *Server) finishEscrowCeremony(job *escrowCeremonyJob, r string) {
 	s.escrowJob = &cp
 	if job.Phase == escrowPhaseDone && r != "" {
 		s.escrowR = []byte(r)
+		s.escrowPending = append([]byte(nil), pending...)
 		s.escrowRClaimed = false
 		s.escrowRExpiry = s.nowFn().Add(escrowClaimTTL)
 		// Active TTL belt: zero the holder even if nobody ever polls again. The lazy check in
@@ -164,6 +183,7 @@ func (s *Server) finishEscrowCeremony(job *escrowCeremonyJob, r string) {
 func (s *Server) wipeEscrowRLocked() {
 	zeroBytes(s.escrowR)
 	s.escrowR = nil
+	s.escrowPending = nil // R-978: the package goes with its code — never uploaded once the code is gone
 }
 
 // expireEscrowRLocked applies the TTL outcome: an armed, unclaimed holder is zeroed and the job
@@ -172,6 +192,9 @@ func (s *Server) expireEscrowRLocked() {
 	if s.escrowJob == nil || s.escrowJob.Phase != escrowPhaseDone || s.escrowRClaimed {
 		return
 	}
+	if s.escrowClaiming {
+		return // R-978: the claim is uploading the package; voiding now would store a package nobody gets the code to
+	}
 	if len(s.escrowR) == 0 {
 		return
 	}
@@ -217,7 +240,7 @@ func (s *Server) handleEscrowCeremonyStart(w http.ResponseWriter, r *http.Reques
 		UpdatedAt: s.nowFn().UTC().Format(time.RFC3339),
 	}
 	if !s.tryStartEscrowCeremony(job) {
-		writeErr(w, http.StatusConflict, "an escrow ceremony is already running")
+		writeErr(w, http.StatusConflict, "an escrow ceremony is already running, or its code is being claimed")
 		return
 	}
 	s.logger.Info("local-api: escrow ceremony started (controller-driven)", "vmid", vmid, "job_id", job.JobID)
@@ -247,7 +270,7 @@ func (s *Server) handleEscrowCeremonyStart(w http.ResponseWriter, r *http.Reques
 				detail += " | stderr: " + tail
 			}
 			job.Phase, job.Detail = escrowPhaseFailed, detail
-			s.finishEscrowCeremony(job, "")
+			s.finishEscrowCeremony(job, "", nil)
 			s.logger.Warn("local-api: escrow ceremony failed", "job_id", job.JobID,
 				"exit", exit, "duration_ms", time.Since(start).Milliseconds())
 			return
@@ -255,24 +278,34 @@ func (s *Server) handleEscrowCeremonyStart(w http.ResponseWriter, r *http.Reques
 		var out escrow.CeremonyOutput
 		if err := json.Unmarshal(stdout, &out); err != nil {
 			job.Phase, job.Detail = escrowPhaseFailed, "ceremony output is not the expected JSON object"
-			s.finishEscrowCeremony(job, "")
+			s.finishEscrowCeremony(job, "", nil)
 			s.logger.Warn("local-api: escrow ceremony output unparseable (never logged)", "job_id", job.JobID)
 			return
 		}
 		if out.Version != escrow.CeremonyOutputVersion || out.RecoveryCode == "" {
 			out.RecoveryCode = ""
 			job.Phase, job.Detail = escrowPhaseFailed, fmt.Sprintf("unexpected ceremony output (version %d)", out.Version)
-			s.finishEscrowCeremony(job, "")
+			s.finishEscrowCeremony(job, "", nil)
+			return
+		}
+		// R-978: the package must come back un-uploaded. An output that uploaded by itself (or carries no
+		// package) would mean the previous package is already gone, or the new one can never be stored.
+		if out.Uploaded || len(out.PendingUpload) == 0 {
+			out.RecoveryCode = ""
+			job.Phase, job.Detail = escrowPhaseFailed, "ceremony output carries no pending package"
+			s.finishEscrowCeremony(job, "", nil)
+			s.logger.Warn("local-api: escrow ceremony output has no pending package — failed (R-978)", "job_id", job.JobID,
+				"uploaded", out.Uploaded)
 			return
 		}
 		job.Phase = escrowPhaseDone
 		job.KeyFingerprint = out.KeyFingerprint
 		job.EntropyBits = out.EntropyBits
 		job.ResticPwSealed = out.ResticPwSealed
-		job.Uploaded = out.Uploaded
+		job.Uploaded = false // R-978: uploaded by the claim
-		s.finishEscrowCeremony(job, out.RecoveryCode)
+		s.finishEscrowCeremony(job, out.RecoveryCode, out.PendingUpload)
 		out.RecoveryCode = "" // drop the parsed reference promptly (GC caveat: best-effort)
-		s.logger.Info("local-api: escrow ceremony complete — R claimable (in-memory, one-shot)",
+		s.logger.Info("local-api: escrow ceremony complete — R claimable (in-memory, one-shot); package uploads at the claim",
 			"job_id", job.JobID, "restic_pw_sealed", job.ResticPwSealed, "uploaded", job.Uploaded,
 			"claim_ttl_s", int(escrowClaimTTL.Seconds()), "duration_ms", time.Since(start).Milliseconds())
 	}()
@@ -339,11 +372,60 @@ func (s *Server) handleEscrowCeremonyClaim(w http.ResponseWriter, r *http.Reques
 		s.escrowMu.Unlock()
 		writeErr(w, http.StatusConflict, "ceremony failed — nothing to claim")
 		return
+	case s.escrowClaiming:
+		s.escrowMu.Unlock()
+		writeErr(w, http.StatusConflict, "the recovery code is being claimed — try again in a moment")
+		return
 	case s.escrowRClaimed || len(s.escrowR) == 0 || s.escrowJob.Phase == escrowPhaseVoid:
 		s.escrowMu.Unlock()
 		writeErr(w, http.StatusGone, "the recovery code is no longer available (already claimed or expired) — run a new ceremony; the new code supersedes")
 		return
 	}
+	// R-978: store the NEW package on the hub first; only then hand out the code that opens it. Until this
+	// upload succeeds the hub keeps the previous package, and the customer's previous code keeps working.
+	if len(s.escrowPending) > 0 {
+		if s.escrowCeremony.Upload == nil {
+			s.escrowMu.Unlock()
+			writeErr(w, http.StatusServiceUnavailable, "the recovery package cannot be stored on this host (no upload configured) — the previous code still works")
+			return
+		}
+		pkg := s.escrowPending
+		jobID := s.escrowJob.JobID
+		s.escrowClaiming = true
+		s.escrowMu.Unlock()
+
+		start := time.Now()
+		base := s.baseCtx
+		if base == nil {
+			base = context.Background()
+		}
+		uctx, cancel := context.WithTimeout(base, escrowUploadTimeout)
+		err := s.escrowCeremony.Upload(uctx, pkg)
+		cancel()
+
+		s.escrowMu.Lock()
+		s.escrowClaiming = false
+		if err != nil {
+			s.escrowMu.Unlock()
+			s.logger.Warn("local-api: escrow claim — package upload to the hub failed; code kept claimable, previous package stays current (R-978)",
+				"vmid", vmid, "job_id", jobID, "err", err, "duration_ms", time.Since(start).Milliseconds())
+			writeErr(w, http.StatusBadGateway, "the new recovery package could not be stored on the hub — the previous code still works; try again")
+			return
+		}
+		s.escrowPending = nil
+		if s.escrowJob != nil && s.escrowJob.JobID == jobID {
+			s.escrowJob.Uploaded = true
+			s.escrowJob.UpdatedAt = s.nowFn().UTC().Format(time.RFC3339)
+		}
+		s.logger.Info("local-api: escrow claim — package uploaded to the hub; handing out the code (R-978)",
+			"vmid", vmid, "job_id", jobID, "duration_ms", time.Since(start).Milliseconds())
+		if len(s.escrowR) == 0 { // cannot happen (TTL and supersede wait for the claim) — never hand out nothing
+			s.escrowMu.Unlock()
+			s.logger.Error("local-api: escrow claim — code vanished during the upload (invariant broken)", "job_id", jobID)
+			writeErr(w, http.StatusInternalServerError, "the recovery code was lost during the claim — run a new ceremony")
+			return
+		}
+	}
 	recovery := string(s.escrowR)
 	s.wipeEscrowRLocked()
 	s.escrowRClaimed = true

internal/localapi/escrow_ceremony_test.go

@@ -2,12 +2,14 @@ package localapi
 
 import (
 	"context"
+	"encoding/base64"
 	"encoding/json"
 	"fmt"
 	"net/http"
 	"os"
 	"path/filepath"
 	"strings"
+	"sync"
 	"sync/atomic"
 	"testing"
 	"time"
@@ -19,25 +21,91 @@ import (
 
 const testR = "canary-alpha-bravo-charlie-delta-echo-foxtrot-golf-hotel-india"
 
-// cannedCeremonyJSON is shaped exactly like the agent's --output=json object (spike §2.6 values).
+// cannedCeremonyJSON is shaped exactly like the agent's --output=json object (spike §2.6 values;
+// version 2, R-978: the package comes back un-uploaded in pending_upload). The package is a MODEL of
+// the R-wrapped blob: it "opens" with exactly the code it was sealed under (fakeEscrowHub.opensWith).
 func cannedCeremonyJSON(r string) string {
-	return fmt.Sprintf(`{"version":1,"recovery_code":%q,"key_fingerprint":"f2:87:68:2a:88:50:16:01","entropy_bits":129,"blob_bytes":383,"identity_blob_bytes":450,"restic_pw_sealed":true,"uploaded":true}`, r)
+	return fmt.Sprintf(`{"version":2,"recovery_code":%q,"key_fingerprint":"f2:87:68:2a:88:50:16:01","entropy_bits":129,"blob_bytes":383,"identity_blob_bytes":450,"restic_pw_sealed":true,"uploaded":false,"pending_upload":%s}`,
+		r, sealedPackage(r))
+}
+
+// sealedPackage is the model hub upload body for a ceremony that minted code r.
+func sealedPackage(r string) string {
+	return fmt.Sprintf(`{"blob_b64":%q,"key_fingerprint":"f2:87:68:2a:88:50:16:01","posture":"zero_knowledge","created_at":"2026-06-10T12:00:00Z"}`,
+		base64.StdEncoding.EncodeToString([]byte("sealed-under:"+r)))
+}
+
+// fakeEscrowHub is the hub's current escrow package for this host (the PUT target). Upload replaces it,
+// as the hub does when the sealed secrets are unchanged (the R-978 case: no retained copy).
+type fakeEscrowHub struct {
+	mu      sync.Mutex
+	current []byte
+	uploads int
+	err     error         // non-nil → the upload fails, nothing stored
+	gate    chan struct{} // non-nil → the upload blocks until closed
+	entered chan struct{} // non-nil → closed when an upload starts
+}
+
+func (f *fakeEscrowHub) upload(ctx context.Context, pkg []byte) error {
+	if f.entered != nil {
+		close(f.entered)
+		f.entered = nil
+	}
+	if f.gate != nil {
+		<-f.gate
+	}
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	if f.err != nil {
+		return f.err
+	}
+	f.current = append([]byte(nil), pkg...)
+	f.uploads++
+	return nil
+}
+
+// opensWith reports whether the hub's CURRENT package opens with code r — the customer's question.
+func (f *fakeEscrowHub) opensWith(r string) bool {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	var pkg struct {
+		BlobB64 string `json:"blob_b64"`
+	}
+	if json.Unmarshal(f.current, &pkg) != nil {
+		return false
+	}
+	raw, err := base64.StdEncoding.DecodeString(pkg.BlobB64)
+	return err == nil && string(raw) == "sealed-under:"+r
+}
+
+func (f *fakeEscrowHub) count() int {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	return f.uploads
 }
 
 // newEscrowTestServer builds a server with the ceremony configured and every seam faked.
 func newEscrowTestServer(t *testing.T, run ceremonyRunner) *Server {
+	t.Helper()
+	srv, _ := newEscrowTestServerHub(t, run, &fakeEscrowHub{})
+	return srv
+}
+
+// newEscrowTestServerHub is newEscrowTestServer with the hub's escrow record exposed (R-978).
+func newEscrowTestServerHub(t *testing.T, run ceremonyRunner, hub *fakeEscrowHub) (*Server, *fakeEscrowHub) {
 	t.Helper()
 	srv := newTestServerS(t, &fakeGuests{}, &fakeBackups{}, &fakeStore{}, nil)
 	srv.escrowCeremony = &EscrowCeremonyConfig{
 		PBSStorageID:  "felhom-pbs",
 		HubConfigured: true,
 		DRConfigured:  func() bool { return true },
+		Upload:        hub.upload,
 	}
 	srv.ceremonyRun = run
 	srv.escrowSudoCheck = func(context.Context) error { return nil }
 	srv.escrowLookPath = func(string) (string, error) { return "/usr/bin/age", nil }
 	srv.statFile = func(string) bool { return true }
-	return srv
+	return srv, hub
 }
 
 // okRunner returns the canned success output and counts invocations.
@@ -160,8 +228,8 @@ func TestEscrowCeremony_OneShotClaim(t *testing.T) {
 	if !strings.Contains(body, `"phase":"done"`) || !strings.Contains(body, `"claimable":true`) {
 		t.Fatalf("status not done/claimable: %s", body)
 	}
-	if !strings.Contains(body, `"restic_pw_sealed":true`) || !strings.Contains(body, `"uploaded":true`) {
+	if !strings.Contains(body, `"restic_pw_sealed":true`) || !strings.Contains(body, `"uploaded":false`) {
-		t.Fatalf("summary fields missing: %s", body)
+		t.Fatalf("summary fields missing (R-978: not uploaded before the claim): %s", body)
 	}
 	assertNoR(t, "status payload", body)
 
@@ -199,7 +267,8 @@ func TestEscrowCeremony_OneShotClaim(t *testing.T) {
 
 	// Post-claim status: claimed, not claimable, still phase done.
 	st2 := do(t, h, "GET", "/escrow/ceremony/status", "A", "")
-	if !strings.Contains(st2.Body.String(), `"claimed":true`) || strings.Contains(st2.Body.String(), `"claimable":true`) {
+	if !strings.Contains(st2.Body.String(), `"claimed":true`) || strings.Contains(st2.Body.String(), `"claimable":true`) ||
+		!strings.Contains(st2.Body.String(), `"uploaded":true`) {
 		t.Fatalf("post-claim status wrong: %s", st2.Body.String())
 	}
 }
@@ -344,7 +413,7 @@ func TestEscrowCeremony_FailurePaths(t *testing.T) {
 	})
 	t.Run("wrong version", func(t *testing.T) {
 		srv := newEscrowTestServer(t, func(context.Context) ([]byte, []byte, int, error) {
-			return []byte(`{"version":2,"recovery_code":"` + testR + `"}`), nil, 0, nil
+			return []byte(`{"version":1,"recovery_code":"` + testR + `","uploaded":true}`), nil, 0, nil
 		})
 		h := srv.Handler()
 		startAndWait(t, srv, h)

internal/localapi/escrow_r978_test.go

@@ -0,0 +1,219 @@
+package localapi
+
+import (
+	"context"
+	"errors"
+	"net/http"
+	"strings"
+	"testing"
+	"time"
+)
+
+// R-978 — starting the escrow ceremony again must not make the customer's written-down recovery code
+// useless. The hub overwrites the current package when the sealed secrets are unchanged, so the NEW
+// package may reach the hub only when its code is claimed. These tests ask the customer's question of
+// the hub's current package — „does my code still open it?" — not whether some function ran.
+
+const (
+	firstR  = "first-ceremony-code-alpha"
+	secondR = "second-ceremony-code-bravo"
+)
+
+// sequenceRunner mints the given codes in order, one per ceremony.
+func sequenceRunner(codes ...string) ceremonyRunner {
+	i := 0
+	return func(context.Context) ([]byte, []byte, int, error) {
+		r := codes[i]
+		if i < len(codes)-1 {
+			i++
+		}
+		return []byte(cannedCeremonyJSON(r)), nil, 0, nil
+	}
+}
+
+func claimCode(t *testing.T, h http.Handler) (int, string) {
+	t.Helper()
+	w := do(t, h, "POST", "/escrow/ceremony/claim", "A", "")
+	return w.Code, w.Body.String()
+}
+
+// THE R-978 CASE. Ceremony 1 is claimed (the customer writes code 1 down). Ceremony 2 is started and
+// abandoned — three ways: its code expires unclaimed, a third start supersedes it, the agent restarts.
+// In every case the hub's current package must still open with code 1, and nothing was uploaded for 2.
+//
+// COMPANION RED-PROOF (observed 2026-10-18): restore the pre-R-978 shape — the package uploaded when the
+// ceremony finishes (s.escrowCeremony.Upload(base, out.PendingUpload) in the job goroutine, as the root
+// subprocess did with `--upload`) and nothing left for the claim (finishEscrowCeremony(job, R, nil)) →
+// all three subtests FAIL with "the abandoned second ceremony replaced the package: code 1 no longer
+// opens it (uploads=2)" (uploads=3 for the superseding start). Restored.
+func TestEscrowCeremony_R978_AbandonedSecondCeremonyKeepsFirstCode(t *testing.T) {
+	setup := func(t *testing.T) (*Server, *fakeEscrowHub, http.Handler, *time.Time) {
+		srv, hub := newEscrowTestServerHub(t, sequenceRunner(firstR, secondR, "third-code-charlie"), &fakeEscrowHub{})
+		cur := testNow
+		srv.now = func() time.Time { return cur }
+		h := srv.Handler()
+		startAndWait(t, srv, h)
+		if code, body := claimCode(t, h); code != http.StatusOK || !strings.Contains(body, firstR) {
+			t.Fatalf("claim of ceremony 1: %d %s", code, body)
+		}
+		if !hub.opensWith(firstR) || hub.count() != 1 {
+			t.Fatalf("after the first claim the hub must hold code 1's package (uploads=%d)", hub.count())
+		}
+		startAndWait(t, srv, h) // ceremony 2 — about to be abandoned
+		return srv, hub, h, &cur
+	}
+	check := func(t *testing.T, hub *fakeEscrowHub) {
+		t.Helper()
+		if !hub.opensWith(firstR) {
+			t.Fatalf("the abandoned second ceremony replaced the package: code 1 no longer opens it (uploads=%d)", hub.count())
+		}
+		if hub.opensWith(secondR) || hub.count() != 1 {
+			t.Fatalf("a package for the never-claimed code 2 reached the hub (uploads=%d)", hub.count())
+		}
+	}
+
+	t.Run("code 2 expires unclaimed", func(t *testing.T) {
+		_, hub, h, cur := setup(t)
+		*cur = cur.Add(escrowClaimTTL + time.Minute)
+		if code, _ := claimCode(t, h); code != http.StatusGone {
+			t.Fatalf("claim after TTL: got %d, want 410", code)
+		}
+		check(t, hub)
+	})
+	t.Run("a third start supersedes code 2", func(t *testing.T) {
+		srv, hub, h, _ := setup(t)
+		startAndWait(t, srv, h) // ceremony 3, also left unclaimed
+		check(t, hub)
+	})
+	t.Run("the agent restarts before code 2 is claimed", func(t *testing.T) {
+		_, hub, _, _ := setup(t)
+		// A restart is a new process with an empty slot; only the hub's record survives it.
+		srv2, _ := newEscrowTestServerHub(t, sequenceRunner("unused"), hub)
+		if code, _ := claimCode(t, srv2.Handler()); code != http.StatusNotFound {
+			t.Fatalf("claim after a restart: got %d, want 404", code)
+		}
+		check(t, hub)
+	})
+}
+
+// The positive control: CLAIMING code 2 is what replaces the package — the code handed out opens the
+// package the hub now holds, and the claim answers only after the hub took it.
+func TestEscrowCeremony_R978_ClaimStoresThePackageBeforeHandingOutTheCode(t *testing.T) {
+	srv, hub := newEscrowTestServerHub(t, sequenceRunner(firstR, secondR), &fakeEscrowHub{})
+	h := srv.Handler()
+	startAndWait(t, srv, h)
+	if hub.count() != 0 {
+		t.Fatalf("the package was uploaded before any claim (uploads=%d)", hub.count())
+	}
+	claimCode(t, h)
+	startAndWait(t, srv, h)
+	code, body := claimCode(t, h)
+	if code != http.StatusOK || !strings.Contains(body, secondR) {
+		t.Fatalf("claim of ceremony 2: %d %s", code, body)
+	}
+	if !hub.opensWith(secondR) || hub.count() != 2 {
+		t.Fatalf("after claiming code 2 the hub must hold its package (uploads=%d, opens with 2: %v)", hub.count(), hub.opensWith(secondR))
+	}
+}
+
+// A failed upload hands out NO code, keeps the code claimable and the previous package current; a retry
+// that reaches the hub succeeds.
+func TestEscrowCeremony_R978_UploadFailureKeepsTheCodeClaimable(t *testing.T) {
+	hub := &fakeEscrowHub{err: errors.New("hub returned HTTP 502")}
+	srv, _ := newEscrowTestServerHub(t, sequenceRunner(firstR), hub)
+	h := srv.Handler()
+	startAndWait(t, srv, h)
+
+	code, body := claimCode(t, h)
+	if code != http.StatusBadGateway {
+		t.Fatalf("claim with the hub failing: got %d, want 502 (%s)", code, body)
+	}
+	assertNoRIn(t, "failed-upload claim payload", body, firstR)
+	if st := do(t, h, "GET", "/escrow/ceremony/status", "A", "").Body.String(); !strings.Contains(st, `"claimable":true`) || !strings.Contains(st, `"uploaded":false`) {
+		t.Fatalf("after a failed upload the code must stay claimable and un-uploaded: %s", st)
+	}
+
+	hub.mu.Lock()
+	hub.err = nil
+	hub.mu.Unlock()
+	code, body = claimCode(t, h)
+	if code != http.StatusOK || !strings.Contains(body, firstR) || !hub.opensWith(firstR) {
+		t.Fatalf("retry after the hub recovered: %d %s (opens: %v)", code, body, hub.opensWith(firstR))
+	}
+}
+
+// While the claim's upload is in flight, neither the TTL nor a new start may take the code away — else
+// the hub would hold a package whose code nobody was given (the R-978 failure, reached the other way).
+func TestEscrowCeremony_R978_TTLAndRestartWaitForAnInFlightClaim(t *testing.T) {
+	hub := &fakeEscrowHub{gate: make(chan struct{}), entered: make(chan struct{})}
+	entered := hub.entered
+	srv, _ := newEscrowTestServerHub(t, sequenceRunner(firstR, secondR), hub)
+	var mu = make(chan struct{}, 1)
+	mu <- struct{}{}
+	cur := testNow
+	srv.now = func() time.Time { <-mu; defer func() { mu <- struct{}{} }(); return cur }
+	h := srv.Handler()
+	startAndWait(t, srv, h)
+
+	type res struct {
+		code int
+		body string
+	}
+	done := make(chan res, 1)
+	go func() { c, b := claimCode(t, h); done <- res{c, b} }()
+	<-entered
+
+	<-mu
+	cur = cur.Add(escrowClaimTTL + time.Minute) // the TTL passes mid-upload
+	mu <- struct{}{}
+	if st := do(t, h, "GET", "/escrow/ceremony/status", "A", "").Body.String(); strings.Contains(st, `"phase":"unclaimed_void"`) {
+		t.Fatalf("the TTL voided the code while its package was uploading: %s", st)
+	}
+	if w := do(t, h, "POST", "/escrow/ceremony", "A", ""); w.Code != http.StatusConflict {
+		t.Fatalf("a new start during the claim's upload: got %d, want 409", w.Code)
+	}
+	if code, _ := claimCode(t, h); code != http.StatusConflict {
+		t.Fatalf("a second claim during the upload: got %d, want 409", code)
+	}
+	close(hub.gate)
+	r := <-done
+	if r.code != http.StatusOK || !strings.Contains(r.body, firstR) || !hub.opensWith(firstR) {
+		t.Fatalf("the in-flight claim must deliver the code whose package it stored: %d %s", r.code, r.body)
+	}
+}
+
+// No upload configured → the claim refuses rather than hand out a code for a package the hub lacks.
+func TestEscrowCeremony_R978_NoUploaderRefusesTheClaim(t *testing.T) {
+	srv, _ := newEscrowTestServerHub(t, sequenceRunner(firstR), &fakeEscrowHub{})
+	srv.escrowCeremony.Upload = nil
+	h := srv.Handler()
+	startAndWait(t, srv, h)
+	code, body := claimCode(t, h)
+	if code != http.StatusServiceUnavailable {
+		t.Fatalf("claim with no uploader: got %d, want 503 (%s)", code, body)
+	}
+	assertNoRIn(t, "no-uploader claim payload", body, firstR)
+}
+
+// An output that uploaded by itself (the pre-R-978 shape) or carries no package fails the ceremony.
+func TestEscrowCeremony_R978_OutputWithoutPendingPackageFails(t *testing.T) {
+	srv, _ := newEscrowTestServerHub(t, func(context.Context) ([]byte, []byte, int, error) {
+		return []byte(`{"version":2,"recovery_code":"` + firstR + `","uploaded":true}`), nil, 0, nil
+	}, &fakeEscrowHub{})
+	h := srv.Handler()
+	startAndWait(t, srv, h)
+	st := do(t, h, "GET", "/escrow/ceremony/status", "A", "").Body.String()
+	if !strings.Contains(st, `"phase":"failed"`) || !strings.Contains(st, "no pending package") {
+		t.Fatalf("want failed: %s", st)
+	}
+	if code, _ := claimCode(t, h); code != http.StatusConflict {
+		t.Fatalf("claim on a failed ceremony: got %d, want 409", code)
+	}
+}
+
+func assertNoRIn(t *testing.T, where, body, r string) {
+	t.Helper()
+	if strings.Contains(body, r) {
+		t.Fatalf("%s carries the recovery code", where)
+	}
+}

internal/localapi/server.go

@@ -90,6 +90,10 @@ type BackupStore interface {
 type LastKnownBackupStore interface {
 	RecordBackupSuccess(target string, b hub.Backup) error
 	LastKnownSuccess(target string, vmid int) (time.Time, bool)
+	// R-973: the cadence cap's restart-proof history and its refusals.
+	RecentBackupSuccesses(target string, vmid int, since time.Time) []time.Time
+	RecordBackupRefusal(target string, vmid int, at time.Time) error
+	LastBackupRefusal() (target string, vmid int, at time.Time, ok bool)
 }
 
 // StorageView yields the host's observed storage targets (for mapping a mount's storage id →
@@ -302,6 +306,11 @@ type Server struct {
 	logger             *slog.Logger
 	now                func() time.Time
 
+	// R-973 cadence cap: this process's successful runs per tier+guest, and the newest refusal.
+	capMu      sync.Mutex
+	capRuns    map[backupJobKey][]time.Time
+	capRefusal capRefusal
+
 	disks           DiskOps           // slice 8C (optional)
 	diskGate        StorageGate       // slice 8C (optional)
 	guestList       GuestLister       // slice 8C (optional)
@@ -426,6 +435,8 @@ type Server struct {
 	escrowMu       sync.Mutex
 	escrowJob      *escrowCeremonyJob
 	escrowR        []byte
+	escrowPending  []byte // R-978: the ceremony's hub upload body, sent by the claim before R is handed out
+	escrowClaiming bool   // R-978: a claim is uploading escrowPending (TTL and a new start wait)
 	escrowRClaimed bool
 	escrowRExpiry  time.Time
 	escrowDone     <-chan struct{} // closes when the detached job finishes (tests wait on it)
@@ -878,6 +889,12 @@ func (s *Server) handleBackup(w http.ResponseWriter, r *http.Request, vmid int)
 			"a backup is already in flight on target "+busyTarget+" (job "+busyJob+") — only one backup runs at a time per guest")
 		return
 	}
+	// R-973: the cadence cap (backup_cap.go). Checked under jobsMu, after the in-flight answers, so two
+	// requests cannot both pass it (only one run per guest is ever in flight).
+	if s.capRefuses(w, key, tier) {
+		s.jobsMu.Unlock()
+		return
+	}
 	// Job ids must be unique PER TIER, and by construction rather than by clock luck: two tiers
 	// started inside the same nanosecond (the weekly both-due night, or any injected clock) would
 	// otherwise collide and hand the second caller the first tier's id. The PRIMARY keeps the
@@ -941,6 +958,9 @@ func (s *Server) handleBackup(w http.ResponseWriter, r *http.Request, vmid int)
 			s.logger.Info("local-api: backup job complete", "vmid", vmid, "target", tier.TargetID, "job", jobID, "archive", b.Archive)
 		}
 		s.store.RecordBackup(b)
+		if b.Success {
+			s.recordCapRun(key, b) // R-973
+		}
 		if b.Success && s.lastKnown != nil {
 			if err := s.lastKnown.RecordBackupSuccess(tier.TargetID, b); err != nil {
 				// Not fatal: the backup exists. Only the fallback after a restart loses this copy.

internal/storage/hostops.go

@@ -419,8 +419,9 @@ func (h *SudoHostOps) InspectDevice(ctx context.Context, device string) (DeviceP
 	// any TYPE/PTTYPE/USAGE line is positive data-bearing evidence. Its exit code is NOT relied
 	// on (blkid exits 2 on a blank device) — output presence is what matters. A broken/empty
 	// blkid simply adds no positive evidence; lsblk (below) is the read-success authority.
-	bout, _, _ := h.runner.Run(ctx, h.bins.Blkid, "-p", "-o", "export", device)
+	bout, _, berr := h.runner.Run(ctx, h.bins.Blkid, "-p", "-o", "export", device)
-	for k, v := range parseBlkidExport(bout) {
+	blk := parseBlkidExport(bout)
+	for k, v := range blk {
 		switch k {
 		case "TYPE":
 			probe.HasFilesystem = true
@@ -439,11 +440,11 @@ func (h *SudoHostOps) InspectDevice(ctx context.Context, device string) (DeviceP
 	// lsblk -J is the READ-SUCCESS authority + the partition/mount view. It exits 0 on any valid
 	// device (blank or not), so a clean parse means the agent reliably read the device. If lsblk
 	// fails, Probed stays false → DataBearing()=true (fail-safe — never call a device blank on a
-	// failed read).
+	// failed read). An exit 0 whose output does not parse into a device is NOT a read either (R-1093
+	// review): it stays unprobed, never "blank".
 	lout, _, lerr := h.runner.Run(ctx, h.bins.Lsblk, "-J", "-o", "NAME,FSTYPE,PTTYPE,MOUNTPOINT", device)
-	if lerr == nil {
+	if hasChildren, mounted, fstype, pttype, parsed := parseLsblkDevice(lout); lerr == nil && parsed {
 		probe.Probed = true
-		hasChildren, mounted, fstype, pttype := parseLsblkDevice(lout)
 		if hasChildren {
 			probe.HasPartitions = true
 		}
@@ -456,13 +457,33 @@ func (h *SudoHostOps) InspectDevice(ctx context.Context, device string) (DeviceP
 				probe.FSType = fstype
 			}
 		}
-		if pttype != "" {
+		if pttype != "" && !inheritedPartitionTable(blk, berr, pttype) {
 			probe.HasPartitionTable = true
 		}
 	}
 	return probe, nil
 }
 
+// inheritedPartitionTable reports whether lsblk's PTTYPE for the probed node is only the PARENT disk's
+// table (R-1093). lsblk prints the disk's table type on every one of its partitions, so an EMPTY
+// partition read as "device has a partition table" and asked for the typed erase confirmation.
+//
+// The partition's OWN signatures come from `blkid -p` on the partition itself (above): a nested table
+// inside it is its PTTYPE, a filesystem is its TYPE. So lsblk's value is set aside ONLY when blkid
+// positively proved two things — that it completed its probe of the device (exit 0: PART_ENTRY_* alone
+// comes from the PARENT's table, so output without a clean exit does not prove the partition's contents
+// were read), and that the device is a partition of a table of exactly that type (PART_ENTRY_NUMBER +
+// PART_ENTRY_SCHEME == lsblk's PTTYPE). Anything else keeps the old, safe reading: a blkid that failed,
+// exited non-zero or printed nothing, a whole disk, or a scheme that differs from lsblk's (then lsblk sees
+// a table blkid did not attribute to the parent). Pinned by TestInspect_R1093_* (hostops_disk_test.go).
+func inheritedPartitionTable(blkid map[string]string, blkidErr error, lsblkPTType string) bool {
+	if blkidErr != nil || blkid["PART_ENTRY_NUMBER"] == "" {
+		return false
+	}
+	scheme := blkid["PART_ENTRY_SCHEME"]
+	return scheme != "" && strings.EqualFold(scheme, lsblkPTType)
+}
+
 // Format runs mkfs.<fstype> on a validated device. The caller is responsible for authorization
 // (8C: only after classifying the device non-data-bearing, or via a slice-10 operator signature).
 func (h *SudoHostOps) Format(ctx context.Context, device, fstype string) error {
@@ -541,13 +562,14 @@ type lsblkDevice struct {
 }
 
 // parseLsblkDevice parses `lsblk -J -o NAME,FSTYPE,PTTYPE,MOUNTPOINT <device>` for the top device:
-// whether it has child partitions, is mounted (itself or any child), and its fstype/pttype.
+// whether it has child partitions, is mounted (itself or any child), and its fstype/pttype. parsed=false
-func parseLsblkDevice(out []byte) (hasChildren, mounted bool, fstype, pttype string) {
+// when the output is not a device listing — the caller must then treat the device as unprobed.
+func parseLsblkDevice(out []byte) (hasChildren, mounted bool, fstype, pttype string, parsed bool) {
 	var doc struct {
 		BlockDevices []lsblkDevice `json:"blockdevices"`
 	}
 	if json.Unmarshal(out, &doc) != nil || len(doc.BlockDevices) == 0 {
-		return false, false, "", ""
+		return false, false, "", "", false
 	}
 	d := doc.BlockDevices[0]
 	fstype, pttype = d.FSType, d.PTType
@@ -558,7 +580,7 @@ func parseLsblkDevice(out []byte) (hasChildren, mounted bool, fstype, pttype str
 			mounted = true
 		}
 	}
-	return hasChildren, mounted, fstype, pttype
+	return hasChildren, mounted, fstype, pttype, true
 }
 
 // NoopHostOps is the safe fallback when the privileged surface is unavailable or declined

internal/storage/hostops_disk_test.go

@@ -222,3 +222,114 @@ func TestInspect_ReadsFilesystemUUID(t *testing.T) {
 		t.Fatalf("FSUUID = %q, want the blkid UUID", p.FSUUID)
 	}
 }
+
+// ---- R-1093: an EMPTY partition is blank; lsblk's PTTYPE on a partition is the PARENT disk's ----------
+
+// blkidEmptyGPTPart is `blkid -p -o export /dev/sdb1` for an empty GPT partition: only the partition's
+// own entry in the parent table (PART_ENTRY_*), no TYPE, no PTTYPE — nothing inside the partition.
+const blkidEmptyGPTPart = "DEVNAME=/dev/sdb1\nPART_ENTRY_SCHEME=gpt\nPART_ENTRY_UUID=3b0c1a52-55a4-4c4e-9d0b-7a1f0e0f7c11\n" +
+	"PART_ENTRY_TYPE=0fc63daf-8483-4772-8e79-3d69d8477de4\nPART_ENTRY_NUMBER=1\nPART_ENTRY_OFFSET=2048\n" +
+	"PART_ENTRY_SIZE=2097152\nPART_ENTRY_DISK=8:16\n"
+
+// lsblkPartInheritsGPT is `lsblk -J -o NAME,FSTYPE,PTTYPE,MOUNTPOINT /dev/sdb1`: lsblk prints the DISK's
+// table type on the partition — the input that made the agent read an empty partition as data-bearing.
+const lsblkPartInheritsGPT = `{"blockdevices":[{"name":"sdb1","fstype":null,"pttype":"gpt","mountpoint":null}]}`
+
+// TestInspect_R1093_PartitionVerdicts pins the CONSEQUENCE (DataBearing + Reason, what decides whether
+// the typed erase confirmation is asked) for a partition, in both directions.
+//
+// COMPANION RED-PROOFS (observed 2026-10-18): (1) restore the pre-fix `if pttype != "" {` (drop the
+// inheritedPartitionTable call) → BOTH empty-partition rows (GPT and dos) FAIL with
+// `DataBearing=true reason="device has a partition table", want blank`. (2) drop the `blkidErr != nil`
+// check → the "exit 2" and "I/O error" rows FAIL with `DataBearing=false reason="device is blank"`.
+// (3) drop `&& parsed` → both lsblk-exit-0 rows FAIL with `Probed=true … reason="device is blank"`
+// (the unparsable-output hole predates R-1093; closed here). Restored.
+func TestInspect_R1093_PartitionVerdicts(t *testing.T) {
+	cases := []struct {
+		name       string
+		dev        string
+		blkid      string
+		blkidErr   error
+		lsblk      string
+		lsblkErr   error
+		wantBear   bool
+		wantReason string
+		unprobed   bool // the verdict must be "could not be reliably inspected"
+	}{
+		{name: "empty GPT partition is blank (the R-1093 case)", dev: "/dev/sdb1",
+			blkid: blkidEmptyGPTPart, lsblk: lsblkPartInheritsGPT,
+			wantBear: false, wantReason: "device is blank"},
+		{name: "empty MBR (dos) partition is blank", dev: "/dev/sdb2",
+			blkid:    "DEVNAME=/dev/sdb2\nPART_ENTRY_SCHEME=dos\nPART_ENTRY_TYPE=0x83\nPART_ENTRY_NUMBER=2\nPART_ENTRY_DISK=8:16\n",
+			lsblk:    `{"blockdevices":[{"name":"sdb2","fstype":null,"pttype":"dos","mountpoint":null}]}`,
+			wantBear: false, wantReason: "device is blank"},
+		{name: "partition with a filesystem stays data-bearing", dev: "/dev/sdb1",
+			blkid:    "DEVNAME=/dev/sdb1\nUUID=8c1e\nTYPE=ext4\nUSAGE=filesystem\n" + blkidEmptyGPTPart[len("DEVNAME=/dev/sdb1\n"):],
+			lsblk:    `{"blockdevices":[{"name":"sdb1","fstype":"ext4","pttype":"gpt","mountpoint":null}]}`,
+			wantBear: true, wantReason: "device has a ext4 filesystem"},
+		{name: "partition holding its OWN (nested) table stays data-bearing", dev: "/dev/sdb1",
+			blkid:    "DEVNAME=/dev/sdb1\nPTUUID=0b8d4358\nPTTYPE=dos\n" + blkidEmptyGPTPart[len("DEVNAME=/dev/sdb1\n"):],
+			lsblk:    lsblkPartInheritsGPT,
+			wantBear: true, wantReason: "device has a partition table"},
+		{name: "blkid printed nothing: lsblk's PTTYPE stands (fail-safe)", dev: "/dev/sdb1",
+			blkid: "", blkidErr: errors.New("sudo: a password is required"), lsblk: lsblkPartInheritsGPT,
+			wantBear: true, wantReason: "device has a partition table"},
+		// Fail-closed paths (the 2026-10-18 review): PART_ENTRY_* is read from the PARENT's table, so it
+		// proves nothing about the partition's contents unless blkid also completed (exit 0).
+		{name: "blkid exit 2 with only PART_ENTRY output: lsblk stands", dev: "/dev/sdb1",
+			blkid: blkidEmptyGPTPart, blkidErr: errors.New("exit status 2"), lsblk: lsblkPartInheritsGPT,
+			wantBear: true, wantReason: "device has a partition table"},
+		{name: "blkid I/O error after PART_ENTRY output: lsblk stands", dev: "/dev/sdb1",
+			blkid: blkidEmptyGPTPart, blkidErr: errors.New("exit status 1"), lsblk: lsblkPartInheritsGPT,
+			wantBear: true, wantReason: "device has a partition table"},
+		{name: "blkid binary missing: lsblk stands", dev: "/dev/sdb1",
+			blkid: "", blkidErr: errors.New(`exec: "/usr/sbin/blkid": file does not exist`), lsblk: lsblkPartInheritsGPT,
+			wantBear: true, wantReason: "device has a partition table"},
+		{name: "lsblk exit 0 with unparsable output: unprobed", dev: "/dev/sdb1",
+			blkid: blkidEmptyGPTPart, lsblk: `lsblk: something odd`,
+			wantBear: true, unprobed: true},
+		{name: "lsblk exit 0 with no device listed: unprobed", dev: "/dev/sdb1",
+			blkid: blkidEmptyGPTPart, lsblk: `{"blockdevices":[]}`,
+			wantBear: true, unprobed: true},
+		{name: "lsblk failed: unprobed", dev: "/dev/sdb1",
+			blkid: blkidEmptyGPTPart, lsblk: "", lsblkErr: errors.New("exit status 32"),
+			wantBear: true, unprobed: true},
+		{name: "scheme differs from lsblk's PTTYPE: lsblk stands", dev: "/dev/sdb1",
+			blkid:    strings.Replace(blkidEmptyGPTPart, "PART_ENTRY_SCHEME=gpt", "PART_ENTRY_SCHEME=dos", 1),
+			lsblk:    lsblkPartInheritsGPT,
+			wantBear: true, wantReason: "device has a partition table"},
+		{name: "whole disk with a GPT stays data-bearing", dev: "/dev/sdb",
+			blkid:    "DEVNAME=/dev/sdb\nPTUUID=a07f88ff\nPTTYPE=gpt\n",
+			lsblk:    `{"blockdevices":[{"name":"sdb","fstype":null,"pttype":"gpt","mountpoint":null}]}`,
+			wantBear: true, wantReason: "device has a partition table"},
+		{name: "whole disk: blkid silent, lsblk GPT stands", dev: "/dev/sdb",
+			blkid: "", lsblk: `{"blockdevices":[{"name":"sdb","fstype":null,"pttype":"gpt","mountpoint":null}]}`,
+			wantBear: true, wantReason: "device has a partition table"},
+	}
+	for _, c := range cases {
+		t.Run(c.name, func(t *testing.T) {
+			r := &scriptedRunner{
+				outputs: map[string][]byte{"blkid": []byte(c.blkid), "lsblk": []byte(c.lsblk)},
+				errs:    map[string]error{"blkid": c.blkidErr, "lsblk": c.lsblkErr},
+			}
+			p, err := newSudo(r).InspectDevice(context.Background(), c.dev)
+			if err != nil {
+				t.Fatal(err)
+			}
+			if c.unprobed {
+				if p.Probed || !p.DataBearing() || p.Reason() != "device could not be reliably inspected" {
+					t.Fatalf("Probed=%v DataBearing=%v reason=%q, want unprobed + data-bearing (probe %+v)",
+						p.Probed, p.DataBearing(), p.Reason(), p)
+				}
+				return
+			}
+			if !p.Probed {
+				t.Fatal("expected a clean probe (lsblk read cleanly)")
+			}
+			if p.DataBearing() != c.wantBear || p.Reason() != c.wantReason {
+				t.Fatalf("DataBearing=%v reason=%q, want DataBearing=%v reason=%q (probe %+v)",
+					p.DataBearing(), p.Reason(), c.wantBear, c.wantReason, p)
+			}
+		})
+	}
+}