hub v0.129.0: operator raises ONE clean-up window's cap (R-833); restore-beside script + runbooks (R-834)

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_0159rPz1ZhFKsS53msqPYxtS

documentation/audits/backup-close-2026-10-04/partA/beside-9298-config.txt

@@ -0,0 +1,14 @@
+arch: amd64
+cores: 7
+features: nesting=1,keyctl=1
+hookscript: local:snippets/felhom-guest-hook.sh
+hostname: demo-hp
+memory: 25898
+mp0: nvme-scratch:9298/vm-9298-disk-1.raw,mp=/var/lib/felhom,backup=1,size=70G
+net0: name=eth0,bridge=vmbr0,hwaddr=BC:24:11:0F:7E:C5,ip=dhcp,link_down=1,type=veth
+net1: name=eth1,bridge=vmbr9,hwaddr=BC:24:11:28:B4:F5,ip=169.254.253.2/30,link_down=1,type=veth
+onboot: 0
+ostype: debian
+rootfs: nvme-scratch:9298/vm-9298-disk-0.raw,size=32G
+swap: 512
+unprivileged: 1

documentation/audits/backup-close-2026-10-04/partA/beside.log

@@ -0,0 +1,25 @@
+restore-beside: restoring local:backup/vzdump-lxc-9201-2026_10_04-04_34_55.tar.zst -> 9298 on nvme-scratch with onboot 0 (never started)
+recovering backed-up configuration from 'local:backup/vzdump-lxc-9201-2026_10_04-04_34_55.tar.zst'
+Formatting '/mnt/hdd_1/images/9298/vm-9298-disk-0.raw', fmt=raw size=34359738368 preallocation=off
+Creating filesystem with 8388608 4k blocks and 2097152 inodes
+Filesystem UUID: 7ddc4545-38c5-4362-8e3d-31a715b29c1b
+Superblock backups stored on blocks: 
+	32768, 98304, 163840, 229376, 294912, 819200, 884736, 1605632, 2654208, 
+	4096000, 7962624
+Formatting '/mnt/hdd_1/images/9298/vm-9298-disk-1.raw', fmt=raw size=75161927680 preallocation=off
+Creating filesystem with 18350080 4k blocks and 4587520 inodes
+Filesystem UUID: 28215aa3-d7b4-484b-b2dc-9f85265e80a5
+Superblock backups stored on blocks: 
+	32768, 98304, 163840, 229376, 294912, 819200, 884736, 1605632, 2654208, 
+	4096000, 7962624, 11239424
+restoring 'local:backup/vzdump-lxc-9201-2026_10_04-04_34_55.tar.zst' now..
+extracting archive '/var/lib/vz/dump/vzdump-lxc-9201-2026_10_04-04_34_55.tar.zst'
+tar: ./var/lib/felhom/docker/volumes/kimai_kimai_var/_data/cache/prod/pools/app/vl6nzGcpxe/7/D/WpVn7g-RmFPG2mIs6Gaw: time stamp 2026-10-05 04:16:20 is 69996.875442074 s in the future
+Total bytes read: 15959541760 (15GiB, 276MiB/s)
+merging backed-up and given configuration..
+restore-beside: removing host-path binds: mp8,mp9
+restore-beside: link down: net0
+restore-beside: link down: net1
+restore-beside: OK: 9298 has onboot 0, no host-path binds, every NIC link_down; it was not started
+restore-beside: read it with: pct mount 9298   (then pct unmount 9298; pct destroy 9298 --purge when done)
+rc=0

documentation/audits/backup-close-2026-10-04/partA/restoretest-config-poll.log

@@ -0,0 +1,54 @@
+=== 2026-10-04T06:44:15Z vmid 990000 status=status: stopped
+lock: create
+=== 2026-10-04T06:44:19Z vmid 990000 status=status: stopped
+lock: create
+=== 2026-10-04T06:44:23Z vmid 990000 status=status: stopped
+lock: create
+=== 2026-10-04T06:44:27Z vmid 990000 status=status: stopped
+lock: create
+=== 2026-10-04T06:44:31Z vmid 990000 status=status: stopped
+lock: create
+=== 2026-10-04T06:44:36Z vmid 990000 status=status: stopped
+lock: create
+=== 2026-10-04T06:44:40Z vmid 990000 status=status: stopped
+lock: create
+=== 2026-10-04T06:44:44Z vmid 990000 status=status: stopped
+lock: create
+=== 2026-10-04T06:44:48Z vmid 990000 status=status: stopped
+lock: create
+=== 2026-10-04T06:44:52Z vmid 990000 status=status: stopped
+lock: create
+=== 2026-10-04T06:44:56Z vmid 990000 status=status: stopped
+lock: create
+=== 2026-10-04T06:45:00Z vmid 990000 status=status: stopped
+lock: create
+=== 2026-10-04T06:45:04Z vmid 990000 status=status: stopped
+lock: create
+=== 2026-10-04T06:45:08Z vmid 990000 status=status: stopped
+lock: create
+=== 2026-10-04T06:45:12Z vmid 990000 status=status: stopped
+lock: create
+=== 2026-10-04T06:45:17Z vmid 990000 status=status: stopped
+hostname: demo-hp
+mp0: nvme-scratch:990000/vm-990000-disk-1.raw,mp=/var/lib/felhom,backup=1,size=70G
+mp8: nvme-scratch:990000/vm-990000-disk-2.raw,mp=/mnt/felhom-drives,backup=0,size=1G
+mp9: nvme-scratch:990000/vm-990000-disk-3.raw,mp=/etc/felhom-bootstrap,backup=0,size=1G
+net0: name=eth0,bridge=vmbr0,hwaddr=BC:24:11:0F:7E:C5,ip=dhcp,type=veth
+net1: name=eth1,bridge=vmbr9,hwaddr=BC:24:11:28:B4:F5,ip=169.254.253.2/30,type=veth
+onboot: 0
+=== 2026-10-04T06:45:20Z vmid 990000 status=status: running
+hostname: demo-hp
+mp0: nvme-scratch:990000/vm-990000-disk-1.raw,mp=/var/lib/felhom,backup=1,size=70G
+mp8: nvme-scratch:990000/vm-990000-disk-2.raw,mp=/mnt/felhom-drives,backup=0,size=1G
+mp9: nvme-scratch:990000/vm-990000-disk-3.raw,mp=/etc/felhom-bootstrap,backup=0,size=1G
+net0: name=eth0,bridge=vmbr0,hwaddr=BC:24:11:0F:7E:C5,ip=dhcp,link_down=1,type=veth
+net1: name=eth1,bridge=vmbr9,hwaddr=BC:24:11:28:B4:F5,ip=169.254.253.2/30,link_down=1,type=veth
+onboot: 0
+=== 2026-10-04T06:45:25Z vmid 990000 status=status: running
+hostname: demo-hp
+mp0: nvme-scratch:990000/vm-990000-disk-1.raw,mp=/var/lib/felhom,backup=1,size=70G
+mp8: nvme-scratch:990000/vm-990000-disk-2.raw,mp=/mnt/felhom-drives,backup=0,size=1G
+mp9: nvme-scratch:990000/vm-990000-disk-3.raw,mp=/etc/felhom-bootstrap,backup=0,size=1G
+net0: name=eth0,bridge=vmbr0,hwaddr=BC:24:11:0F:7E:C5,ip=dhcp,link_down=1,type=veth
+net1: name=eth1,bridge=vmbr9,hwaddr=BC:24:11:28:B4:F5,ip=169.254.253.2/30,link_down=1,type=veth
+onboot: 0

documentation/audits/backup-close-2026-10-04/partA/restoretest.log

@@ -0,0 +1,27 @@
+=== felhom-agent 0.138.0 selftest=restore-test ===
+  --- recover: reaping any leaked scratch from a prior crashed test ---
+  recover: examined=0 scratch_destroyed=0 scratch_clean=0
+  restoring local:backup/vzdump-lxc-9201-2026_10_04-04_34_55.tar.zst into scratch band [990000,990009] on nvme-scratch …
+time=2026-10-04T08:44:13.035+02:00 level=INFO msg="restore-test: space preflight passed" storage=nvme-scratch required_bytes=24520159232 avail_bytes=876487208960
+time=2026-10-04T08:44:13.137+02:00 level=INFO msg="restore-test: full-fidelity restore params derived from the archive config" scratch=990000 params=4
+time=2026-10-04T08:45:25.769+02:00 level=INFO msg="audit: gate decision" class=guest_destroy host=demo-hp-bb76ea guest=990000 source=one_shot_job disposition=benign allowed=true reason=benign key_id="" nonce="" durable_id=""
+time=2026-10-04T08:45:25.769+02:00 level=INFO msg="gate decision" class=guest_destroy guest=990000 source=one_shot_job disposition=benign allowed=true reason=benign
+time=2026-10-04T08:45:31.852+02:00 level=INFO msg="restore-test: scratch guest torn down" vmid=990000
+  --- restore-test record ---
+  {
+    "source_archive": "local:backup/vzdump-lxc-9201-2026_10_04-04_34_55.tar.zst",
+    "source_tier": "local",
+    "scratch_vmid": 990000,
+    "pass": true,
+    "verified": "boot+running",
+    "tested_at": "2026-10-04T06:45:31Z",
+    "duration_seconds": 79.434773547,
+    "mount_parity": "ok",
+    "mount_inventory": [
+      "mp0=/var/lib/felhom (70G)",
+      "mp8=/mnt/felhom-drives (throwaway for the archived bind)",
+      "mp9=/etc/felhom-bootstrap (throwaway for the archived bind)"
+    ]
+  }
+  space preflight passed: storage=nvme-scratch required=24520159232 avail=876487208960
+=== selftest=restore-test OK (scratch 990000 restored+booted+verified+torn-down in 1m19s) ===

documentation/audits/backup-close-2026-10-04/partB/hub-redproofs.txt

@@ -0,0 +1,10 @@
+# R-833 hub red-proofs, 2026-10-04 (hub tree before v0.129.0 release). Each mutation applied, test run, reverted.
+RP1 raise ignored (OpenWindowFor keeps the default cap):
+--- FAIL: TestWindow_RaisedCapIsOneWindowOnly — raised grant: {Granted:true WindowID:2 ... MaxRemove:20} — want MaxRemove 30
+RP2 grant not consumed (TakeOffsiteWindowGrant does not clear the setting):
+--- FAIL: TestWindow_RaisedCapIsOneWindowOnly — the raised grant was not consumed
+RP3 close check uses the default cap instead of the window's own:
+--- FAIL: TestWindow_RaisedCapIsOneWindowOnly — a drop of 28 under a raised cap of 30 alarmed — the close check ignored the window's own cap
+RP4 the box API window-open sets a grant from a max_remove in its body:
+--- FAIL: TestOffsiteWindow_BoxCannotGrantItself — a box call left a grant (ok=true max=400)
+After revert: ok internal/offsitekeys, ok internal/api

documentation/audits/backup-close-2026-10-04/partB/lab-proof.md

@@ -0,0 +1,20 @@
+# R-833 lab proof — 2026-10-04 (a lab repo on DooPlex scratch, not a demo box's)
+
+restic 0.14.0 from the controller image `felhom-controller:0.290.0` (`docker run --rm --entrypoint sh`), a local repo,
+password `lab-only-password` (a lab value). 98 daily snapshots, `--host demo-lab --tag felhom-offsite`, dated 97..0 days
+back — the shape of a box whose windows were off for three months.
+
+The box's own guard (`offsiteGuard`, controller v0.290.0 source) was run on the repo's REAL `snapshots --json` and the
+policy's REAL `forget --dry-run --json` by `offbox_window_lab_test.go.txt` (copy it into
+`controller/internal/backup/` and set `OFFSITE_LAB_DIR` to rerun; it skips without it, so it is not committed).
+
+| Step | Result |
+|---|---|
+| Honest plan | 98 snapshots, the policy removes **85** |
+| Default cap (hub `MaxRemove` = half) | **49 → REFUSED**: `the plan would remove 85 snapshots, more than one week's retention may (49)` |
+| Operator-raised cap 90 (one window) | **85 ids, no refusal** |
+| `restic forget <those 85 ids> --prune` | rc 0; **98 → 13**; `restic check` rc 0 |
+| Next window, default cap again (6) | plan 0, **no refusal** — the cap only needed raising once |
+
+The hub half (the grant is one-shot, the next window has the default cap, the close check uses the window's own cap, a
+box cannot grant itself) is in `hub-redproofs.txt` and the hub suite.

documentation/audits/backup-close-2026-10-04/partB/offbox_window_lab_test.go.txt

@@ -0,0 +1,64 @@
+package backup
+
+import (
+	"encoding/json"
+	"os"
+	"path/filepath"
+	"strings"
+	"testing"
+	"time"
+)
+
+// R-833 LAB (not run in CI: needs OFFSITE_LAB_DIR from the lab script). Reads a REAL restic 0.14.0
+// repo's `snapshots --json` and the policy's `forget --dry-run --json`, and runs the box's own guard
+// with the hub's default cap and with an operator-raised cap. Writes the ids the guard would remove to
+// $OFFSITE_LAB_DIR/ids.txt for the lab script to prune. Evidence: audits/backup-close-2026-10-04/partB/.
+func TestOffsiteGuard_R833_LabRepo(t *testing.T) {
+	dir := os.Getenv("OFFSITE_LAB_DIR")
+	if dir == "" {
+		t.Skip("lab only: set OFFSITE_LAB_DIR")
+	}
+	var all []guardSnap
+	b, err := os.ReadFile(filepath.Join(dir, "snaps.json"))
+	if err != nil {
+		t.Fatal(err)
+	}
+	if err := json.Unmarshal(b, &all); err != nil {
+		t.Fatal(err)
+	}
+	b, err = os.ReadFile(filepath.Join(dir, "plan.json"))
+	if err != nil {
+		t.Fatal(err)
+	}
+	var groups []struct {
+		Remove []guardSnap `json:"remove"`
+	}
+	if err := json.Unmarshal(b, &groups); err != nil {
+		t.Fatal(err)
+	}
+	var plan []guardSnap
+	for _, g := range groups {
+		plan = append(plan, g.Remove...)
+	}
+	now := time.Now()
+	def := len(all) / 2 // the hub's MaxRemove(count): half, minimum 5
+	if def < 5 {
+		def = 5
+	}
+	_, refuse := offsiteGuard(all, plan, now, now, def)
+	t.Logf("snapshots=%d plan=%d default cap=%d → refusal: %q", len(all), len(plan), def, refuse)
+	raised := 0
+	if v := os.Getenv("OFFSITE_LAB_RAISED"); v != "" {
+		for _, c := range v {
+			raised = raised*10 + int(c-'0')
+		}
+	}
+	if raised == 0 {
+		return
+	}
+	ids, refuse2 := offsiteGuard(all, plan, now, now, raised)
+	t.Logf("raised cap=%d → %d id(s), refusal: %q", raised, len(ids), refuse2)
+	if err := os.WriteFile(filepath.Join(dir, "ids.txt"), []byte(strings.Join(ids, "\n")), 0o600); err != nil {
+		t.Fatal(err)
+	}
+}

documentation/runbooks/RUNBOOK-manual-guest-restore.md

@@ -73,6 +73,12 @@ pct list | awk '{print $1}' | grep -x -e <TARGET-VMID> -e <SOURCE-VMID>
 
 ## 3. Restore, then fix the binds BEFORE first boot
 
+> **Restoring BESIDE a live original** (a copy to read, a check, a drill)? **Do not use this section.** Use
+> `felhom.eu/scripts/felhom-restore-beside.sh <scratch-vmid> <volid> <storage>`: onboot 0 at restore, every host-path
+> bind removed, every NIC down, never started, config read back (R-834, proven live 2026-10-04). This section is for
+> a **replaced host**, where the original is gone and the binds are right. A bare `pct restore` keeps the archive's
+> `onboot: 1`, so a host reboot would start it.
+
 Restore **without starting** the guest. `pct restore` does not auto-start, but never pass anything that
 would, and do not `pct start` until §4 passes.
 

documentation/runbooks/ep0-datastore-copy.md

@@ -45,14 +45,16 @@ recovered with the household's recovery code — the same as restoring from ep0)
    `namespace <customer>` / `username root@pam!<name>`.
    **Do not use `pvesm add pbs` without `--password`:** it validates with the password from its command line, fails 401,
    and on failure DELETES the `.pw`/`.enc` files you placed (measured). Passing `--password` puts the token on argv.
-3. `pvesm list <id>` → the household's snapshots (measured: 2 s). `pct restore <scratch VMID> <id>:backup/ct/<vmid>/<time>
-   --storage <dir storage> --unique 1` (measured: **186 s for a 15 GB-logical / 14 GB-on-disk backup** over the LAN, key
-   fingerprint printed by the restore).
-4. **⚠ BEFORE ANYTHING ELSE — the restored config is the PRODUCTION one:** `onboot: 1`, `mp8` bound to the host's REAL
-   household drives (`/mnt/felhom-drives`) and `mp9` to the original guest's bootstrap. Starting it, or a host reboot,
-   runs a second controller for the same household against the same drives. On a restore BESIDE the original:
-   `pct set <vmid> --onboot 0 --delete mp8,mp9` immediately (R-834). On a true replacement host, where the original is
-   gone, the binds are what you want.
+3. `pvesm list <id>` → the household's snapshots (measured: 2 s). Then restore with the safe script (R-834):
+   `felhom-restore-beside.sh <scratch VMID> <id>:backup/ct/<vmid>/<time> <dir storage>` (copy it from
+   `felhom.eu/scripts/`; measured 2026-10-03 by hand: **186 s for a 15 GB-logical / 14 GB-on-disk backup** over the LAN).
+4. **⚠ Why the script, not a bare `pct restore`:** the archive's config is the PRODUCTION one — `onboot: 1`, `mp8` bound
+   to the host's REAL household drives (`/mnt/felhom-drives`) and `mp9` to the original guest's bootstrap. Started, or
+   after a host reboot, it is a second controller for the same household on the same drives. The script restores with
+   `--onboot 0`, removes every host-path bind, takes every NIC down, never starts it, and reads the config back
+   (proven live 2026-10-04, `audits/backup-close-2026-10-04/partA/`). On a true replacement host, where the original is
+   gone, the binds are what you want: use `RUNBOOK-manual-guest-restore.md` §3 or the agent's DR bring-up instead (the
+   DR bring-up refuses beside a live original since agent v0.139.0).
 5. Read the data without starting it: `pct mount <vmid>` → `/var/lib/lxc/<vmid>/rootfs` (measured: 1 s; rootfs, the
    controller data volume and `/var/lib/felhom` present, `settings.json` dated 10 min before the backup) → `pct unmount`.
 6. Teardown: `pct destroy <vmid> --purge`; `pvesm remove <id>` (removes the entry and its priv files — never the

hub/CHANGELOG.md

@@ -1,3 +1,18 @@
+## v0.129.0 — the operator can raise ONE clean-up window's cap after a long gap (R-833)
+
+- After weeks without windows the honest backlog exceeds half the snapshots, and the box's guard refuses every window
+  at the default cap (half, ≥ 5) — the clean-up wedges. `POST /offsite/window-grant/<id>` with `max_remove=<n>`
+  (operator login only; `1..500`, a known customer) grants the customer's NEXT window with that cap. Only the count
+  cap moves: the box's fake-snapshot guard still runs in full. The grant is consumed by that one window; the next has
+  the default cap again. Each window row records the cap it was opened with (`offsite_windows.max_remove`), and the
+  close check compares the drop with THAT cap, so a granted clean-up does not raise a false `offsite_window_drop`.
+  Operator event `offsite_window_large_grant` (warning, operator-only). Without `max_remove` the route is unchanged.
+- The box needs no change: it already takes the cap from the hub's window answer (controller ≥ 0.289.0).
+- Tests: `TestWindow_RaisedCapIsOneWindowOnly`, `TestGrantLargeWindow_EventAndBounds`,
+  `TestOffsiteWindow_BoxCannotGrantItself` (no box-authenticated call leaves a grant). Red-proofs RP1–RP4:
+  `felhom.eu/documentation/audits/backup-close-2026-10-04/partB/`. Lab proof on a real restic 0.14.0 repo: 98
+  snapshots, plan 85, default cap 49 refused, raised cap 90 → 98 → 13, the next window passes the default cap.
+
 ## v0.128.0 — the household's set-aside deletion through the hub with a 7-day wait (decision 74, R-823); key-file clean-up (decision 72, R-826); the key check is read-only (R-827); the window cap fits an honest week (2026-10-04)
 
 - **Set-aside deletion (R-823).** The box (controller ≥ 0.290.0) hands a due "delete my earlier off-site backups" to

hub/cmd/hub/main.go

@@ -426,6 +426,7 @@ func main() {
 		}
 		webServer.SetOffsiteKeyAudit(runKeyAudit)
 		webServer.SetOffsiteWindowAdmin(dataStore.GrantOffsiteWindowOnce, dataStore.SetOffsiteWindowsEnabled)
+		webServer.SetOffsiteWindowLargeGrant(keySvc.GrantLargeWindow)
 		webServer.SetOffsiteKeyAdmin(keySvc.RemoveUnpinnedKeys, keySvc.CancelAbandon)
 		// Decision 68: a window the box never closed is closed by the hub at its 20-minute bound.
 		go func() {

hub/internal/api/offsite_test.go

@@ -138,3 +138,25 @@ func TestOffsiteKeyEndpoints_AuthAndNoPasswordInAnyResponse(t *testing.T) {
 		t.Fatalf("service not reached: pub=%q fp=%q", f.gotPub, f.gotFP)
 	}
 }
+
+// R-833: a box cannot raise its own window cap. No box-authenticated route sets a grant, and a
+// window-open body that names a cap is not a grant. Asserted on the consequence: the store holds no
+// grant after every box call.
+func TestOffsiteWindow_BoxCannotGrantItself(t *testing.T) {
+	h, st, _ := newTestHandler(t)
+	st.SaveCustomerConfig(&store.CustomerConfig{CustomerID: "c1", APIKey: "ckey", RetrievalPassword: "pp"})
+	h.SetOffsiteKeyService(&fakeKeySvc{})
+	for _, c := range []struct{ path, body string }{
+		{"/api/v1/offsite/window-open/c1", `{"count_before":40,"max_remove":400}`},
+		{"/api/v1/offsite/window-grant/c1", `{"max_remove":400}`},
+		{"/api/v1/offsite/window-large-grant/c1", `{"max_remove":400}`},
+		{"/offsite/window-grant/c1", `max_remove=400`},
+	} {
+		req := httptest.NewRequest(http.MethodPost, c.path, strings.NewReader(c.body))
+		req.Header.Set("Authorization", "Bearer ckey")
+		h.ServeHTTP(httptest.NewRecorder(), req)
+	}
+	if ok, n := st.TakeOffsiteWindowGrant("c1"); ok || n != 0 {
+		t.Fatalf("a box call left a grant (ok=%v max=%d)", ok, n)
+	}
+}

hub/internal/notify/dispatcher.go

@@ -678,6 +678,8 @@ var operatorOnlyEvents = map[string]bool{
 	"offsite_window_drop":         true,
 	"offsite_window_failed":       true,
 	"offsite_prune_guard_refused": true,
+	// R-833 (v0.129.0): the operator raised ONE window's removal cap — an operator act, logged.
+	"offsite_window_large_grant": true,
 	// R-197 (v0.93.0). "The sealed offsite repository key changed" is a custody fact about escrow
 	// blobs. A customer can take no action on it — the remedy is the operator's inspection of the
 	// off-site tier — and the text is operator-grade English naming host ids and retained-blob

hub/internal/offsitekeys/service.go

@@ -7,6 +7,7 @@ import (
 	"fmt"
 	"log"
 	"strings"
+	"strconv"
 	"time"
 
 	"gitea.dooplex.hu/admin/felhom-hub/internal/offsite"
@@ -208,6 +209,7 @@ const (
 	EventWindowDrop   = "offsite_window_drop"         // error: more snapshots went than a window may remove
 	EventWindowFailed = "offsite_window_failed"       // warning: a window errored or was left open
 	EventGuardRefused = "offsite_prune_guard_refused" // error: the box's fake-snapshot guard refused (R-822)
+	EventWindowLargeGrant = "offsite_window_large_grant" // warning: the operator raised one window's cap (R-833)
 	windowLength      = 20 * time.Minute
 	windowCadence     = 6*24*time.Hour + 12*time.Hour // "weekly", with slack for the night chain's drift
 )
@@ -243,10 +245,44 @@ func MaxRemove(countBefore int) int {
 	return n
 }
 
+// windowCap is the cap a window was OPENED with (an operator grant may have raised it, R-833); rows
+// from before v0.129.0 carry none and fall back to the default.
+func windowCap(w *store.OffsiteWindow) int {
+	if w.MaxRemove > 0 {
+		return w.MaxRemove
+	}
+	return MaxRemove(w.CountBefore)
+}
+
+// MaxRemoveGrantCeiling bounds an operator's raised cap: a typo of an extra zero must not turn one
+// window into "remove anything". A real backlog above it is cleared over several granted windows.
+const MaxRemoveGrantCeiling = 500
+
+// GrantLargeWindow is the OPERATOR's one-shot grant with a raised cap for that one window (R-833).
+// It is reachable only from the operator's hub login (internal/web), never from the box API. The grant
+// is consumed by the next window; the window after it has the default cap again. Logged as an
+// operator event.
+func (s *Service) GrantLargeWindow(customerID string, maxRemove int) error {
+	if maxRemove < 1 || maxRemove > MaxRemoveGrantCeiling {
+		return fmt.Errorf("offsitekeys: max_remove %d is outside 1..%d", maxRemove, MaxRemoveGrantCeiling)
+	}
+	if c, err := s.Store.GetCustomerConfig(customerID); err != nil || c == nil {
+		return fmt.Errorf("offsitekeys: no customer %q", customerID)
+	}
+	if err := s.Store.GrantOffsiteWindowOnceMax(customerID, maxRemove); err != nil {
+		return err
+	}
+	s.logf("[WARN] offsitekeys: operator granted ONE clean-up window for %s with a raised cap of %d", customerID, maxRemove)
+	s.event(customerID, EventWindowLargeGrant, "warning",
+		fmt.Sprintf("Off-site clean-up: the operator granted one window with a raised removal cap of %d (the fake-snapshot guard still applies).", maxRemove),
+		map[string]any{"max_remove": maxRemove})
+	return nil
+}
+
 // OpenWindowFor decides and, if due, opens the window: a deleting line for the box's CONFIRMED key is
 // prepended (first match wins), and a ledger row bounds it to 20 minutes.
 func (s *Service) OpenWindowFor(ctx context.Context, customerID string, countBefore int) (WindowGrant, error) {
-	oneShot := s.Store.TakeOffsiteWindowGrant(customerID)
+	oneShot, raisedMax := s.Store.TakeOffsiteWindowGrant(customerID)
 	last := s.Store.LastOffsiteWindowOpened(customerID)
 	due := last.IsZero() || s.now().Sub(last) >= windowCadence
 	if !oneShot && !(s.Store.OffsiteWindowsEnabled() && due) {
@@ -268,15 +304,23 @@ func (s *Service) OpenWindowFor(ctx context.Context, customerID string, countBef
 		return WindowGrant{}, err
 	}
 	now := s.now()
-	id, err := s.Store.OpenOffsiteWindowRow(customerID, now.Add(windowLength), countBefore)
+	// R-833: an operator grant may RAISE the cap for this one window (never lower it). Only the count
+	// cap moves; the box's fake-snapshot guard still runs in full against NewestAllowed.
+	maxRemove := MaxRemove(countBefore)
+	raised := raisedMax > maxRemove
+	if raised {
+		maxRemove = raisedMax
+	}
+	id, err := s.Store.OpenOffsiteWindowRow(customerID, now.Add(windowLength), countBefore, maxRemove)
 	if err != nil {
 		// The line is written; close it rather than leave a deleting line without a ledger row.
 		_ = s.Reg.CloseWindow(ctx, t, pw)
 		return WindowGrant{}, err
 	}
-	g := WindowGrant{Granted: true, WindowID: id, NewestAllowed: now.UTC(), MaxRemove: MaxRemove(countBefore)}
-	s.logf("[WARN] offsitekeys: clean-up window %d OPENED for %s (key %s, %d snapshot(s), max %d removed, closes by %s, one-shot=%v)",
-		id, customerID, k.Fingerprint, countBefore, g.MaxRemove, now.Add(windowLength).UTC().Format(time.RFC3339), oneShot)
+	g := WindowGrant{Granted: true, WindowID: id, NewestAllowed: now.UTC(), MaxRemove: maxRemove}
+	s.logf("[WARN] offsitekeys: clean-up window %d OPENED for %s (key %s, %d snapshot(s), max %d removed%s, closes by %s, one-shot=%v)",
+		id, customerID, k.Fingerprint, countBefore, g.MaxRemove, map[bool]string{true: " — OPERATOR-RAISED cap (default " + strconv.Itoa(MaxRemove(countBefore)) + ")", false: ""}[raised],
+		now.Add(windowLength).UTC().Format(time.RFC3339), oneShot)
 	return g, nil
 }
 
@@ -297,13 +341,14 @@ func (s *Service) CloseWindowFor(ctx context.Context, customerID string, r Windo
 		s.logf("[WARN] offsitekeys: ledger close %d: %v", w.ID, err)
 	}
 	drop := w.CountBefore - r.CountAfter
+	allowed := windowCap(w)
 	s.logf("[INFO] offsitekeys: clean-up window %d CLOSED for %s: outcome=%s, %d -> %d (drop %d, allowed %d)",
-		w.ID, customerID, r.Outcome, w.CountBefore, r.CountAfter, drop, MaxRemove(w.CountBefore))
+		w.ID, customerID, r.Outcome, w.CountBefore, r.CountAfter, drop, allowed)
 	details := map[string]any{"window_id": w.ID, "count_before": w.CountBefore, "count_after": r.CountAfter, "outcome": r.Outcome, "reason": r.Reason}
 	switch {
-	case drop > MaxRemove(w.CountBefore):
+	case drop > allowed:
 		s.event(customerID, EventWindowDrop, "error",
-			fmt.Sprintf("Off-site clean-up window %d: the snapshot count fell %d -> %d, more than a window may remove (%d).", w.ID, w.CountBefore, r.CountAfter, MaxRemove(w.CountBefore)), details)
+			fmt.Sprintf("Off-site clean-up window %d: the snapshot count fell %d -> %d, more than a window may remove (%d).", w.ID, w.CountBefore, r.CountAfter, allowed), details)
 	case r.Outcome == "guard-refused":
 		s.event(customerID, EventGuardRefused, "error",
 			fmt.Sprintf("Off-site clean-up window %d: the box's fake-snapshot guard refused to prune — nothing was deleted: %s", w.ID, r.Reason), details)

hub/internal/offsitekeys/service_test.go

@@ -220,3 +220,76 @@ func TestMaxRemove_HonestWeekFits(t *testing.T) {
 		t.Fatal("floor of 5 lost")
 	}
 }
+
+// R-833: after a long gap the honest backlog exceeds half the snapshots, and the default cap makes the
+// box's guard refuse every window. The operator's raised-cap grant opens ONE window with a larger cap;
+// it is consumed, the next window has the default cap again, the close check uses the raised cap (no
+// false drop alarm), and the grant is an operator event. Red-proof: drop the `if raised` assignment in
+// OpenWindowFor and the first assertion fails.
+func TestWindow_RaisedCapIsOneWindowOnly(t *testing.T) {
+	s, _, events := svcFixture(t)
+	ctx := context.Background()
+	pub, fp := newKey(t)
+	if _, err := s.RegisterKey(ctx, "c1", pub); err != nil {
+		t.Fatal(err)
+	}
+	if _, err := s.ConfirmKey(ctx, "c1", fp); err != nil {
+		t.Fatal(err)
+	}
+	// Without the raised grant: a plain one-shot gives half of 40 = 20.
+	_ = s.Store.GrantOffsiteWindowOnce("c1")
+	g0, err := s.OpenWindowFor(ctx, "c1", 40)
+	if err != nil || !g0.Granted || g0.MaxRemove != 20 {
+		t.Fatalf("plain grant: %+v %v — want the default cap 20", g0, err)
+	}
+	_ = s.CloseWindowFor(ctx, "c1", WindowResult{WindowID: g0.WindowID, CountAfter: 40, Outcome: "guard-refused"})
+
+	if err := s.GrantLargeWindow("c1", 30); err != nil {
+		t.Fatal(err)
+	}
+	g, err := s.OpenWindowFor(ctx, "c1", 40)
+	if err != nil || !g.Granted || g.MaxRemove != 30 {
+		t.Fatalf("raised grant: %+v %v — want MaxRemove 30", g, err)
+	}
+	*events = nil
+	if err := s.CloseWindowFor(ctx, "c1", WindowResult{WindowID: g.WindowID, CountAfter: 12, Outcome: "pruned"}); err != nil {
+		t.Fatal(err)
+	}
+	for _, e := range *events {
+		if e == EventWindowDrop {
+			t.Fatal("a drop of 28 under a raised cap of 30 alarmed — the close check ignored the window's own cap")
+		}
+	}
+	// The grant was consumed: no window without a new grant (weekly windows are off here) …
+	if g2, _ := s.OpenWindowFor(ctx, "c1", 12); g2.Granted {
+		t.Fatal("the raised grant was not consumed")
+	}
+	// … and the next granted window is back on the default cap.
+	_ = s.Store.GrantOffsiteWindowOnce("c1")
+	g3, _ := s.OpenWindowFor(ctx, "c1", 40)
+	if !g3.Granted || g3.MaxRemove != 20 {
+		t.Fatalf("next window: %+v — want the default cap 20 again", g3)
+	}
+}
+
+// The grant is an operator event, bounded, and only for a known customer.
+func TestGrantLargeWindow_EventAndBounds(t *testing.T) {
+	s, _, events := svcFixture(t)
+	for _, bad := range []int{0, -1, MaxRemoveGrantCeiling + 1} {
+		if err := s.GrantLargeWindow("c1", bad); err == nil {
+			t.Fatalf("max_remove %d accepted", bad)
+		}
+	}
+	if err := s.GrantLargeWindow("nobody", 10); err == nil {
+		t.Fatal("a grant for an unknown customer was accepted")
+	}
+	if ok, _ := s.Store.TakeOffsiteWindowGrant("c1"); ok {
+		t.Fatal("a refused grant left a grant behind")
+	}
+	if err := s.GrantLargeWindow("c1", 10); err != nil {
+		t.Fatal(err)
+	}
+	if len(*events) != 1 || (*events)[0] != EventWindowLargeGrant {
+		t.Fatalf("events = %v, want one %s", *events, EventWindowLargeGrant)
+	}
+}

hub/internal/store/offsite_keys.go

@@ -2,6 +2,9 @@ package store
 
 import (
 	"database/sql"
+	"fmt"
+	"strconv"
+	"strings"
 	"time"
 )
 
@@ -72,12 +75,15 @@ type OffsiteWindow struct {
 	CountAfter  int
 	BoxResult   string
 	CloseReason string
+	// MaxRemove is the cap this window was opened with (R-833: an operator grant may raise it for one
+	// window). 0 on rows written before v0.129.0 — readers fall back to the default cap then.
+	MaxRemove int
 }
 
-// OpenOffsiteWindowRow records a window the hub just opened.
-func (s *Store) OpenOffsiteWindowRow(customerID string, closesBy time.Time, countBefore int) (int64, error) {
-	res, err := s.db.Exec(`INSERT INTO offsite_windows (customer_id, opened_at, closes_by, count_before) VALUES (?, datetime('now'), ?, ?)`,
-		customerID, closesBy.UTC().Format("2006-01-02 15:04:05"), countBefore)
+// OpenOffsiteWindowRow records a window the hub just opened, with the cap it was opened under.
+func (s *Store) OpenOffsiteWindowRow(customerID string, closesBy time.Time, countBefore, maxRemove int) (int64, error) {
+	res, err := s.db.Exec(`INSERT INTO offsite_windows (customer_id, opened_at, closes_by, count_before, max_remove) VALUES (?, datetime('now'), ?, ?, ?)`,
+		customerID, closesBy.UTC().Format("2006-01-02 15:04:05"), countBefore, maxRemove)
 	if err != nil {
 		return 0, err
 	}
@@ -100,9 +106,9 @@ func (s *Store) GetOffsiteWindow(id int64) (*OffsiteWindow, error) {
 	var w OffsiteWindow
 	var opened, closesBy string
 	var closed, boxRes, reason sql.NullString
-	var before, after sql.NullInt64
-	err := s.db.QueryRow(`SELECT id, customer_id, opened_at, closes_by, closed_at, count_before, count_after, box_result, close_reason FROM offsite_windows WHERE id = ?`, id).
-		Scan(&w.ID, &w.CustomerID, &opened, &closesBy, &closed, &before, &after, &boxRes, &reason)
+	var before, after, maxRm sql.NullInt64
+	err := s.db.QueryRow(`SELECT id, customer_id, opened_at, closes_by, closed_at, count_before, count_after, box_result, close_reason, max_remove FROM offsite_windows WHERE id = ?`, id).
+		Scan(&w.ID, &w.CustomerID, &opened, &closesBy, &closed, &before, &after, &boxRes, &reason, &maxRm)
 	if err == sql.ErrNoRows {
 		return nil, nil
 	}
@@ -115,6 +121,7 @@ func (s *Store) GetOffsiteWindow(id int64) (*OffsiteWindow, error) {
 	}
 	w.CountBefore, w.CountAfter = int(before.Int64), int(after.Int64)
 	w.BoxResult, w.CloseReason = boxRes.String, reason.String
+	w.MaxRemove = int(maxRm.Int64)
 	return &w, nil
 }
 
@@ -166,14 +173,34 @@ func (s *Store) GrantOffsiteWindowOnce(customerID string) error {
 	return s.setSetting("offsite_window_grant:"+customerID, "1")
 }
 
-// TakeOffsiteWindowGrant consumes a one-shot grant; true when one was present.
-func (s *Store) TakeOffsiteWindowGrant(customerID string) bool {
+// GrantOffsiteWindowOnceMax is the operator's one-shot grant that ALSO raises the removal cap for that
+// one window (R-833: after a long gap the honest backlog exceeds half the snapshots and the default cap
+// refuses every window). The raised cap is consumed with the grant; the next window has the default.
+func (s *Store) GrantOffsiteWindowOnceMax(customerID string, maxRemove int) error {
+	if maxRemove <= 0 {
+		return fmt.Errorf("max_remove must be positive, got %d", maxRemove)
+	}
+	return s.setSetting("offsite_window_grant:"+customerID, "max:"+strconv.Itoa(maxRemove))
+}
+
+// TakeOffsiteWindowGrant consumes a one-shot grant: granted is true when one was present, and
+// maxRemove is the operator's raised cap for that window (0 = none, use the default).
+func (s *Store) TakeOffsiteWindowGrant(customerID string) (granted bool, maxRemove int) {
 	k := "offsite_window_grant:" + customerID
-	if s.getSetting(k) != "1" {
-		return false
+	v := s.getSetting(k)
+	switch {
+	case v == "1":
+	case strings.HasPrefix(v, "max:"):
+		n, err := strconv.Atoi(strings.TrimPrefix(v, "max:"))
+		if err != nil || n <= 0 {
+			n = 0 // a malformed value still grants the window, at the default cap
+		}
+		maxRemove = n
+	default:
+		return false, 0
 	}
 	_ = s.setSetting(k, "")
-	return true
+	return true, maxRemove
 }
 
 // ForceOffsiteWindowDueForTest back-dates a window's closes_by. TEST-ONLY.

hub/internal/store/store.go

@@ -856,6 +856,8 @@ func (s *Store) migrate() error {
 	`); err != nil {
 		return fmt.Errorf("offsite_keys/offsite_windows: %w", err)
 	}
+	// R-833 (v0.129.0): the cap each window was opened with — an operator grant may raise it once.
+	s.db.Exec("ALTER TABLE offsite_windows ADD COLUMN max_remove INTEGER")
 
 	return nil
 }

hub/internal/web/server.go

@@ -11,6 +11,7 @@ import (
 	"log"
 	"math"
 	"net/http"
+	"strconv"
 	"strings"
 	"sync"
 	"time"
@@ -71,6 +72,7 @@ type Server struct {
 	offsiteKeyAudit func(ctx context.Context) any
 	// offsiteWindowAdmin: operator one-shot grant / weekly switch (decision 68). nil → 503.
 	offsiteWindowGrant  func(customerID string) error
+	offsiteWindowGrantMax func(customerID string, maxRemove int) error
 	offsiteWindowSwitch func(on bool) error
 	// operator key-file clean-up (decision 72) and abandonment cancel (decision 74). nil → 503.
 	offsiteRemoveUnpinned func(ctx context.Context, customerID string) (int, error)
@@ -210,6 +212,9 @@ func (s *Server) SetOffsiteKeyAdmin(remove func(context.Context, string) (int, e
 	s.offsiteRemoveUnpinned, s.offsiteAbandonCancel = remove, cancel
 }
 
+// SetOffsiteWindowLargeGrant wires the operator's one-window raised cap (R-833).
+func (s *Server) SetOffsiteWindowLargeGrant(fn func(string, int) error) { s.offsiteWindowGrantMax = fn }
+
 // SetOffsiteWindowAdmin wires the operator's window controls (decision 68).
 func (s *Server) SetOffsiteWindowAdmin(grant func(string) error, sw func(bool) error) {
 	s.offsiteWindowGrant, s.offsiteWindowSwitch = grant, sw
@@ -655,6 +660,18 @@ func (s *Server) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 		var err error
 		if path == "/offsite/windows-enabled" {
 			err = s.offsiteWindowSwitch(r.FormValue("on") == "1")
+		} else if mr := r.FormValue("max_remove"); mr != "" {
+			// R-833: one window with a raised removal cap (operator only — this server is behind the
+			// operator's login; the box API has no route to it).
+			n, perr := strconv.Atoi(mr)
+			if perr != nil || s.offsiteWindowGrantMax == nil {
+				http.Error(w, "max_remove must be a number", http.StatusBadRequest)
+				return
+			}
+			if err = s.offsiteWindowGrantMax(strings.TrimPrefix(path, "/offsite/window-grant/"), n); err != nil {
+				http.Error(w, err.Error(), http.StatusBadRequest)
+				return
+			}
 		} else {
 			err = s.offsiteWindowGrant(strings.TrimPrefix(path, "/offsite/window-grant/"))
 		}

scripts/felhom-restore-beside.sh

@@ -0,0 +1,60 @@
+#!/bin/bash
+# felhom-restore-beside.sh — restore a whole-guest archive BESIDE a live original, safely (R-834).
+#
+# Run as root on a Proxmox host. The restored guest is for READING (pct mount, a file copy, a check):
+#   - it is created with onboot 0, so a host reboot never starts it;
+#   - every mpN that binds a HOST path (mp8 = the household's real drives, mp9 = the original guest's
+#     bootstrap) is removed before anything can start it;
+#   - every NIC is set link_down=1 (the archive keeps the original's MAC and island address);
+#   - it is NEVER started by this script.
+# The script then reads the config back and fails loudly if any of that is not true.
+#
+# NOT for a replaced host where the original is gone — there the binds are right; use the agent's DR
+# bring-up or RUNBOOK-manual-guest-restore.md §3.
+#
+# Usage: felhom-restore-beside.sh <scratch-vmid> <volid> <storage>
+#   e.g. felhom-restore-beside.sh 9299 tmp-dooplex-copy:backup/ct/9201/2026-10-03T19:00:00Z nvme-scratch
+#
+# Test: scripts/test_felhom_restore_beside.py (a fake pct on PATH).
+set -euo pipefail
+
+die() { echo "restore-beside: REFUSED: $*" >&2; exit 1; }
+log() { echo "restore-beside: $*"; }
+
+[ $# -eq 3 ] || die "usage: $0 <scratch-vmid> <volid> <storage>"
+vmid=$1 volid=$2 storage=$3
+[[ "$vmid" =~ ^[0-9]+$ ]] || die "vmid '$vmid' is not a number"
+# The restore-test band and the standing scratch are the agent's; a customer guest is never a target.
+if [ "$vmid" -ge 990000 ] && [ "$vmid" -le 990009 ]; then die "vmid $vmid is the agent's restore-test band"; fi
+[ "$vmid" != 9999 ] || die "vmid 9999 is the agent's standing scratch"
+if pct status "$vmid" >/dev/null 2>&1; then die "vmid $vmid already exists — this script never restores over a guest"; fi
+
+log "restoring $volid -> $vmid on $storage with onboot 0 (never started)"
+pct restore "$vmid" "$volid" --storage "$storage" --unprivileged 1 --onboot 0
+
+# Strip host-path binds and take the NICs down, BEFORE anything else can touch the guest.
+conf=$(pct config "$vmid" --current)
+binds=$(printf '%s\n' "$conf" | sed -n -E 's/^(mp[0-9]+): \/.*/\1/p' | paste -sd, -)
+if [ -n "$binds" ]; then
+	log "removing host-path binds: $binds"
+	pct set "$vmid" --delete "$binds"
+fi
+printf '%s\n' "$conf" | sed -n -E 's/^(net[0-9]+): (.*)$/\1 \2/p' | while read -r key val; do
+	case ",$val," in *",link_down=1,"*) continue ;; esac
+	log "link down: $key"
+	pct set "$vmid" "--$key" "$val,link_down=1"
+done
+pct set "$vmid" --onboot 0
+
+# Read back: the CONSEQUENCE, not the commands.
+conf=$(pct config "$vmid" --current)
+bad=""
+printf '%s\n' "$conf" | grep -qx 'onboot: 0' || bad="$bad onboot-not-0"
+printf '%s\n' "$conf" | grep -qE '^mp[0-9]+: /' && bad="$bad host-bind-left"
+printf '%s\n' "$conf" | grep -E '^net[0-9]+: ' | grep -qv 'link_down=1' && bad="$bad nic-up"
+if [ -n "$bad" ]; then
+	echo "restore-beside: FAILED read-back on $vmid:$bad — do NOT start it; destroy it with: pct destroy $vmid" >&2
+	exit 2
+fi
+log "OK: $vmid has onboot 0, no host-path binds, every NIC link_down; it was not started"
+log "read it with: pct mount $vmid   (then pct unmount $vmid; pct destroy $vmid --purge when done)"

scripts/test_felhom_restore_beside.py

@@ -0,0 +1,133 @@
+#!/usr/bin/env python3
+"""Tests for felhom-restore-beside.sh (R-834) — a fake `pct` on PATH holds the guest config in a file.
+
+The restored config is the PRODUCTION one (onboot 1, mp8 on the household's drives, mp9 on the
+original's bootstrap, NICs up). The script must end with onboot 0, no host-path bind, every NIC
+link_down, and never start the guest. Asserted on the consequence: the fake's final config file and
+its call log. Red-proof: drop the `pct set --delete` line from the script and test_strips fails.
+
+Run: python3 scripts/test_felhom_restore_beside.py
+"""
+import os
+import pathlib
+import stat
+import subprocess
+import sys
+import tempfile
+import unittest
+
+SCRIPT = pathlib.Path(__file__).with_name("felhom-restore-beside.sh")
+
+PROD_CONF = """arch: amd64
+hostname: demo-hp
+onboot: 1
+rootfs: nvme-scratch:9299/vm-9299-disk-0.raw,size=16G
+mp0: nvme-scratch:9299/vm-9299-disk-1.raw,mp=/var/lib/felhom,backup=1,size=70G
+mp8: /mnt/felhom-drives,mp=/mnt/felhom-drives
+mp9: /var/lib/felhom-agent/guests/9201/bootstrap,mp=/etc/felhom-bootstrap,ro=1
+net0: name=eth0,bridge=vmbr0,hwaddr=BC:24:11:0F:7E:C5,ip=dhcp,type=veth
+net1: name=eth1,bridge=vmbr9,hwaddr=BC:24:11:28:B4:F5,ip=169.254.253.2/30,type=veth
+"""
+
+# A minimal pct: status (exists iff the conf file exists), restore (writes PROD_CONF, records whether
+# --onboot 0 was passed), config, set (--delete / --netN / --onboot), start (recorded — must never run).
+FAKE_PCT = r'''#!/usr/bin/env python3
+import os, sys
+d = os.environ["FAKE_PCT_DIR"]
+conf = os.path.join(d, "conf")
+with open(os.path.join(d, "calls"), "a") as f:
+    f.write(" ".join(sys.argv[1:]) + "\n")
+cmd = sys.argv[1]
+def read():
+    return [l for l in open(conf).read().splitlines() if l]
+def write(lines):
+    open(conf, "w").write("\n".join(lines) + "\n")
+if cmd == "status":
+    sys.exit(0 if os.path.exists(conf) else 2)
+if cmd == "restore":
+    lines = open(os.path.join(d, "prod")).read().splitlines()
+    args = sys.argv[3:]
+    if "--onboot" in args:
+        v = args[args.index("--onboot") + 1]
+        lines = [("onboot: " + v) if l.startswith("onboot:") else l for l in lines]
+    write([l for l in lines if l])
+    sys.exit(0)
+if cmd == "config":
+    print("\n".join(read()))
+    sys.exit(0)
+if cmd == "set":
+    lines = read()
+    a = sys.argv[3:]
+    while a:
+        k, v = a[0], a[1]
+        a = a[2:]
+        if k == "--delete":
+            drop = set(v.split(","))
+            lines = [l for l in lines if l.split(":")[0] not in drop]
+        else:
+            key = k.lstrip("-")
+            lines = [l for l in lines if l.split(":")[0] != key] + [key + ": " + v]
+    write(lines)
+    sys.exit(0)
+if cmd == "start":
+    sys.exit(0)
+sys.exit(9)
+'''
+
+
+def run(tmp, *args, script=SCRIPT, exists=False):
+    bindir = pathlib.Path(tmp, "bin")
+    bindir.mkdir(exist_ok=True)
+    pct = bindir / "pct"
+    pct.write_text(FAKE_PCT)
+    pct.chmod(pct.stat().st_mode | stat.S_IEXEC)
+    pathlib.Path(tmp, "prod").write_text(PROD_CONF)
+    if exists:
+        pathlib.Path(tmp, "conf").write_text(PROD_CONF)
+    env = dict(os.environ, PATH=f"{bindir}:{os.environ['PATH']}", FAKE_PCT_DIR=tmp)
+    return subprocess.run(["bash", str(script), *args], env=env, capture_output=True, text=True)
+
+
+class RestoreBeside(unittest.TestCase):
+    def test_strips(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            r = run(tmp, "9299", "tmp:backup/ct/9201/x", "nvme-scratch")
+            self.assertEqual(r.returncode, 0, r.stderr + r.stdout)
+            conf = pathlib.Path(tmp, "conf").read_text()
+            calls = pathlib.Path(tmp, "calls").read_text()
+            self.assertIn("onboot: 0", conf)
+            self.assertNotIn("onboot: 1", conf)
+            for line in conf.splitlines():
+                self.assertFalse(line.startswith("mp") and ": /" in line, f"host bind left: {line}")
+                if line.startswith("net"):
+                    self.assertIn("link_down=1", line)
+            self.assertIn("mp0:", conf, "a storage volume must stay")
+            self.assertNotIn("\nstart", "\n" + calls, "the script started the guest")
+            self.assertIn("--onboot 0", calls.splitlines()[1], "onboot 0 must be set AT restore time")
+
+    def test_refuses_an_existing_vmid(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            r = run(tmp, "9201", "tmp:backup/ct/9201/x", "nvme-scratch", exists=True)
+            self.assertNotEqual(r.returncode, 0)
+            self.assertIn("already exists", r.stderr)
+            self.assertNotIn("restore", pathlib.Path(tmp, "calls").read_text())
+
+    def test_refuses_the_agent_bands(self):
+        for v in ("990003", "9999", "abc"):
+            with tempfile.TemporaryDirectory() as tmp:
+                r = run(tmp, v, "tmp:backup/ct/9201/x", "s")
+                self.assertNotEqual(r.returncode, 0, v)
+
+    def test_readback_catches_a_bind_left_behind(self):
+        # Red-proof built in: the same script with the --delete line removed must FAIL its read-back.
+        with tempfile.TemporaryDirectory() as tmp:
+            broken = pathlib.Path(tmp, "broken.sh")
+            broken.write_text(SCRIPT.read_text().replace('\tpct set "$vmid" --delete "$binds"\n', "\t:\n"))
+            self.assertNotEqual(broken.read_text(), SCRIPT.read_text(), "mutation did not apply")
+            r = run(tmp, "9299", "tmp:backup/ct/9201/x", "s", script=broken)
+            self.assertEqual(r.returncode, 2, r.stdout + r.stderr)
+            self.assertIn("host-bind-left", r.stderr)
+
+
+if __name__ == "__main__":
+    sys.exit(unittest.main())