hub v0.161.0: passphrase keys only until the box uses its own (box-config route; R-1159, D19); ep0 namespaces without a prune job alarmed, tenantsync v1.3.0 creates the job (R-1165)

2026-10-19 15:55:45 +02:00
parent 224fe41e06
commit b6a911c3f9
16 changed files with 802 additions and 35 deletions
@@ -1,3 +1,33 @@
+## v0.161.0 — the passphrase gives keys only until the box uses its own (R-1159, operator decision D19); an alarm for an ep0 household with no prune job (R-1165) (2026-10-19)
+
+- **The retrieval passphrase's keys** (R-1159, D19 = „box key first, then lock"; `05` §8 [DESIGN R-1159]). Measured the
+  same day: the passphrase alone got the controller's hub key (`GET /config`, `GET /recovery`) and the existing host's
+  key (`POST /host-enroll`). A new box needs it: the installer's preflight, download login and enroll, and the guest's
+  first boot (`bootstrap.json` → `GET /config`, which is how the box gets its own key); and until controller 0.334.0
+  every live box on each config change. Now: **`GET /box-config/{id}`** — the same generated `controller.yaml`, with the
+  controller's own hub key (self-scoped or global; route table entry `authCustomer`). And **`passphraseKeysClosed`**: a
+  host record exists AND a controller report arrived since it was made AND that controller is `boxConfigMinController`
+  (0.334.0) or newer → `/config`, `/recovery` and the reuse path of `/host-enroll` answer **409 with no config and no
+  key** and the operator gets `passphrase_keys_refused` (warning, operator-only, hub-origin). Open otherwise (a first
+  install; a reinstall after the operator's delete; the install in progress; a box on an older controller, which still
+  refreshes with the passphrase — the lock arms box by box as the floor moves). A wrong passphrase stays a plain 401.
+  `/artifacts` (versions and checksums, no key) is unchanged. `internal/api/passphrase_scope.go`; tests
+  `TestR1159_PassphraseGivesKeysOnlyUntilTheBoxUsesItsOwn` (five cases; red-proved: with the lock off the live box
+  gets all three keys), `TestR1159_BoxPullsItsConfigWithItsOwnKey` (own key 200; no key, a wrong key, another
+  household's key, the passphrase as a bearer → 401), `TestR1159_BoxConfigMinControllerIsTheReleaseThatPullsWithItsKey`
+  (the constant names the controller release whose heading says box-config; skips, saying so, without the sibling).
+  **Known edge:** an installer re-run on a live box (repair, a late `--resume`) now stops at its preflight with
+  „unexpected hub status 409" — the installer does not word the 409 yet (R-1167).
+- **ep0: a household namespace with no prune job is alarmed** (R-1165). `tenantsync` v1.3.0's `usage` answer carries
+  `unpruned` (namespaces no job covers) and `unpruned_checked`; `PBSDRBoxChecker` raises the operator-scope
+  `pbsdr_namespace_without_prune_job` (warning) when the set changes to a non-empty one, re-arms silently when it
+  empties, and does nothing on an unread list (an old endpoint) — never „all covered". `TestPBSDRBox_R1165_NamespaceWithoutPruneJob`
+  (red-proved), `TestUsage_R1165_PruneCoverage`. Both new event types are in `operatorOnlyEvents`.
+- **The script, `scripts/felhom-tenantsync.sh` v1.3.0** (not in the image; installed on ep0 by hand, `runbooks/offsite-endpoint.md`
+  §10): provision/reissue create `prune-<id>` (keep-weekly 2, 03:30, ns-scoped, max-depth 0) unless a job already
+  names the namespace; a failed create never fails the provision; deprovision removes the namespace's jobs first.
+  `scripts/test_tenantsync_prune.py` (12 cases, run by the script-tests gate on every push; red-proved).
+
 ## v0.160.0 — a reinstall within 45 minutes mails the operator too (R-1156); further refused backups are counted and the alarm repeats once a day (R-1157); the off-site re-issue on a reinstall says what it still does (R-1158); hub gofmt gate (R-1160) (2026-10-19)

 - **A live re-enroll is a WARNING mailed to the operator** (R-1156). `host_reenroll_while_live` was info (recorded,
@@ -408,6 +408,8 @@ func (h *Handler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 	case r.Method == http.MethodGet && strings.HasPrefix(path, "/config/"):
 		customerID := strings.TrimPrefix(path, "/config/")
 		h.handleConfigRetrieve(w, r, customerID)
+	case r.Method == http.MethodGet && strings.HasPrefix(path, "/box-config/"):
+		h.handleBoxConfig(w, r, strings.TrimPrefix(path, "/box-config/"))
 	case r.Method == http.MethodPost && strings.HasPrefix(path, "/offsite/consume-password/"):
 		customerID := strings.TrimPrefix(path, "/offsite/consume-password/")
 		h.handleOffsiteConsumePassword(w, r, customerID)
@@ -1239,6 +1241,11 @@ func (h *Handler) handleHostEnroll(w http.ResponseWriter, r *http.Request) {
 			return
 		}
 		h.noteReenrollOverExistingHost(cc, existing, time.Now())
+		// R-1159: once the box uses its own key, the passphrase no longer re-serves the host key.
+		if closed, why := h.passphraseKeysClosed(req.CustomerID); closed {
+			h.refusePassphraseKeys(w, r, req.CustomerID, "POST /host-enroll", why)
+			return
+		}
 		w.Header().Set("Content-Type", "application/json")
 		w.WriteHeader(http.StatusOK)
 		json.NewEncoder(w).Encode(map[string]string{"host_id": existing.HostID, "api_key": existing.APIKey})
@@ -2824,6 +2831,10 @@ func (h *Handler) handleRecovery(w http.ResponseWriter, r *http.Request, custome
 		http.Error(w, "Unauthorized: invalid password", http.StatusUnauthorized)
 		return
 	}
+	if closed, why := h.passphraseKeysClosed(customerID); closed {
+		h.refusePassphraseKeys(w, r, customerID, "GET /recovery", why)
+		return
+	}

 	// Generate controller.yaml. The claim state is baked read-only (no issue on the DR path — a
 	// recovered box whose settings.json is gone re-gates on the EXISTING hash; the reset flow
@@ -2897,36 +2908,16 @@ func (h *Handler) handleConfigRetrieve(w http.ResponseWriter, r *http.Request, c
 		return
 	}

-	if h.templateProvider == nil {
-		http.Error(w, "Config generation not available", http.StatusServiceUnavailable)
+	// R-1159: the passphrase gives keys only until the box uses its own (passphrase_scope.go).
+	if closed, why := h.passphraseKeysClosed(customerID); closed {
+		h.refusePassphraseKeys(w, r, customerID, "GET /config", why)
 		return
 	}
-
-	// Customer-claim arc (v0.50.0): the REAL config pull (Day-0 installer / controller refresh) is
-	// the Day-0 claim entry point — issue + email the first code here (idempotent), and bake the
-	// active hash into the generated web.claim_code_hash so the box is gated from FIRST boot.
-	// (The operator-UI preview deliberately does NOT issue — it only bakes an existing hash.)
-	var claimState *store.ClaimState
-	if h.claimEngine != nil {
-		var cerr error
-		claimState, cerr = h.claimEngine.EnsureIssued(cfg)
-		if cerr != nil {
-			// Loud but non-fatal: the config is still served; if a hash was stored the gate is armed
-			// and the operator resends the email from the customer page.
-			h.logger.Printf("[WARN] claim issue for %s on config retrieve: %v", customerID, cerr)
-		}
-	}
-
-	yamlOutput, err := configgen.Generate(h.templateProvider.Template(), cfg, claimState)
-	if err != nil {
-		h.logger.Printf("[ERROR] Failed to generate config for %s: %v", customerID, err)
-		http.Error(w, "Internal error", http.StatusInternalServerError)
-		return
-	}
-
-	h.logger.Printf("[INFO] Config downloaded for customer %s", customerID)
-	w.Header().Set("Content-Type", "text/yaml; charset=utf-8")
-	w.Write([]byte(yamlOutput))
+	// Customer-claim arc (v0.50.0): the REAL config pull (Day-0 installer / controller refresh) is the Day-0 claim
+	// entry point — serveGeneratedConfig issues + emails the first code (idempotent) and bakes the active hash into
+	// the generated web.claim_code_hash so the box is gated from FIRST boot. (The operator-UI preview deliberately
+	// does NOT issue — it only bakes an existing hash.)
+	h.serveGeneratedConfig(w, cfg, "retrieval passphrase")
 }

 // artifactManifestResponse is the wire shape the host-bootstrap script consumes: the operator-vouched
@@ -0,0 +1,136 @@
+package api
+
+import (
+	"encoding/json"
+	"net/http"
+	"time"
+
+	"gitea.dooplex.hu/admin/felhom-hub/internal/configgen"
+	"gitea.dooplex.hu/admin/felhom-hub/internal/semver"
+	"gitea.dooplex.hu/admin/felhom-hub/internal/store"
+)
+
+// R-1159 (hub v0.161.0, operator decision D19 = „box key first, then lock", 2026-10-19) — what the household's
+// retrieval passphrase may still get.
+//
+// MEASURED 2026-10-19 (audits/day-2026-10-19b/P2/passphrase-reach.txt): the passphrase alone returned the controller's
+// hub key (GET /config, GET /recovery) and the existing host's key (POST /host-enroll), and through them every off-site
+// key route and the host routes. The passphrase is needed by a NEW box: the installer (preflight, its download login,
+// enroll), and the new guest's first boot (bootstrap.json → GET /config, which is how the box gets its own key). And,
+// until controller 0.334.0, by every LIVE box on each config change (report.ConfigRefresher → GET /config).
+//
+// THE RULE. The passphrase gives keys only until the box has a key of its own and uses it:
+//   - no host record → open (a first install; a reinstall after the operator deleted the old host);
+//   - a host record, and no controller report received since the record was made → open (the install in progress,
+//     its --resume, the new guest's first boot);
+//   - a host record, and a controller report received since it was made from a controller that pulls its config with
+//     its OWN key (boxConfigMinController and newer, GET /box-config) → CLOSED: /config and /recovery answer 409 with
+//     no config, /host-enroll answers 409 with no key, and the operator gets passphrase_keys_refused (warning, mailed).
+//     A reinstall then needs the operator's delete first — the rule since R-1154.
+//   - a controller OLDER than boxConfigMinController still refreshes with the passphrase, so it stays open for it
+//     (the lock arms itself box by box as the floor moves; no ordering between the hub and controller releases).
+//
+// GET /box-config/{id} — the box's own config pull, authenticated with the controller's hub key (self-scoped, or the
+// global key), the same generated controller.yaml as GET /config.
+//
+// Pinned by r1159_passphrase_scope_test.go.
+
+const (
+	// boxConfigMinController is the first controller that refreshes its config with its own key (GET /box-config).
+	boxConfigMinController = "0.334.0"
+
+	eventPassphraseKeysRefused = "passphrase_keys_refused"
+)
+
+// passphraseKeysClosed applies THE RULE above. A failed lookup is closed (fail-safe: a key is a key).
+func (h *Handler) passphraseKeysClosed(customerID string) (closed bool, why string) {
+	host, err := h.store.GetHostByCustomer(customerID)
+	if err != nil {
+		return true, "the host lookup failed: " + err.Error()
+	}
+	if host == nil {
+		return false, "no host record"
+	}
+	found, at, ver, err := h.store.LatestControllerReport(customerID)
+	if err != nil {
+		return true, "the controller report lookup failed: " + err.Error()
+	}
+	if !found || at.Before(host.CreatedAt) {
+		return false, "the box's controller has not reported since host " + host.HostID + " was made"
+	}
+	if !semver.Valid(ver) || semver.Compare(ver, boxConfigMinController) < 0 {
+		return false, "the box's controller (" + ver + ") still refreshes its config with the passphrase"
+	}
+	return true, "host " + host.HostID + " is live: its controller " + ver + " reported at " + at.UTC().Format("2006-01-02 15:04 UTC") +
+		" and pulls its config with its own key"
+}
+
+// refusePassphraseKeys answers 409 with a plain sentence (no config, no key) and tells the operator.
+func (h *Handler) refusePassphraseKeys(w http.ResponseWriter, r *http.Request, customerID, route, why string) {
+	h.logger.Printf("[WARN] %s for %s with the retrieval passphrase REFUSED (R-1159): %s — visitor %s", route, customerID, why, r.RemoteAddr)
+	msg := "Someone with the household's retrieval passphrase asked " + route + " for the box keys of " + customerID +
+		"; refused because " + why + ". If this is a reinstall: delete the old host (Hosts page), then run the installer again. " +
+		"If nobody in the household did this, the passphrase may be known to someone else."
+	details, _ := json.Marshal(map[string]string{"route": route, "reason": why})
+	if _, err := h.store.SaveEvent(customerID, eventPassphraseKeysRefused, "warning", msg, string(details), "hub"); err != nil {
+		h.logger.Printf("[WARN] save %s for %s: %v", eventPassphraseKeysRefused, customerID, err)
+	} else if h.dispatcher != nil {
+		go h.dispatcher.ProcessEvent(customerID, eventPassphraseKeysRefused, "warning", msg, string(details), "hub")
+	}
+	http.Error(w, "Conflict: this customer already has a live box, which uses its own key. The passphrase gives no keys now. "+
+		"For a reinstall the operator deletes the old host first.", http.StatusConflict)
+}
+
+// handleBoxConfig — GET /box-config/{id}: the box's own config pull with its controller key (controller ≥ 0.334.0).
+func (h *Handler) handleBoxConfig(w http.ResponseWriter, r *http.Request, customerID string) {
+	authCustomerID, isGlobal, ok := h.checkAuthCustomer(r)
+	if !ok || (!isGlobal && authCustomerID != customerID) {
+		http.Error(w, "unauthorized", http.StatusUnauthorized)
+		return
+	}
+	cfg, err := h.store.GetCustomerConfig(customerID)
+	if err != nil {
+		h.logger.Printf("[ERROR] box-config: customer lookup failed for %s: %v", customerID, err)
+		http.Error(w, "Internal error", http.StatusInternalServerError)
+		return
+	}
+	if cfg == nil {
+		http.Error(w, "Not found", http.StatusNotFound)
+		return
+	}
+	if cfg.SecretsUnreadable {
+		h.logger.Printf("[ERROR] customer %s: sealed secrets do not open (OFFSITE_SECRET_KEY)", cfg.CustomerID)
+		http.Error(w, "Internal error", http.StatusInternalServerError)
+		return
+	}
+	h.serveGeneratedConfig(w, cfg, "box key")
+}
+
+// serveGeneratedConfig writes the generated controller.yaml (GET /config and GET /box-config). The claim code is
+// issued here (idempotent) — the real config pull is the Day-0 claim entry point.
+func (h *Handler) serveGeneratedConfig(w http.ResponseWriter, cfg *store.CustomerConfig, by string) {
+	if h.templateProvider == nil {
+		http.Error(w, "Config generation not available", http.StatusServiceUnavailable)
+		return
+	}
+	var claimState *store.ClaimState
+	if h.claimEngine != nil {
+		var cerr error
+		claimState, cerr = h.claimEngine.EnsureIssued(cfg)
+		if cerr != nil {
+			// Loud but non-fatal: the config is still served; if a hash was stored the gate is armed
+			// and the operator resends the email from the customer page.
+			h.logger.Printf("[WARN] claim issue for %s on config retrieve: %v", cfg.CustomerID, cerr)
+		}
+	}
+	start := time.Now()
+	yamlOutput, err := configgen.Generate(h.templateProvider.Template(), cfg, claimState)
+	if err != nil {
+		h.logger.Printf("[ERROR] Failed to generate config for %s: %v", cfg.CustomerID, err)
+		http.Error(w, "Internal error", http.StatusInternalServerError)
+		return
+	}
+	h.logger.Printf("[INFO] Config downloaded for customer %s (by %s, %s)", cfg.CustomerID, by, time.Since(start).Round(time.Millisecond))
+	w.Header().Set("Content-Type", "text/yaml; charset=utf-8")
+	w.Write([]byte(yamlOutput))
+}
@@ -0,0 +1,34 @@
+package api
+
+import (
+	"os"
+	"path/filepath"
+	"regexp"
+	"strings"
+	"testing"
+)
+
+// boxConfigMinController must be the controller release that pulls its config with its own key (GET /box-config). If
+// it named an earlier one, a box on that controller would be refused its passphrase config refresh while it still
+// needs it (its config changes would stop arriving). Read from the controller's CHANGELOG beside this repo (the
+// workspace layout; a CI clone has no sibling, and says so).
+func TestR1159_BoxConfigMinControllerIsTheReleaseThatPullsWithItsKey(t *testing.T) {
+	p := filepath.Join("..", "..", "..", "..", "felhom-controller", "CHANGELOG.md")
+	b, err := os.ReadFile(p)
+	if err != nil {
+		t.Skipf("NOT CHECKED: the controller clone is not beside this repo (%v) — the workspace run checks it", err)
+	}
+	m := regexp.MustCompile(`(?m)^## v` + regexp.QuoteMeta(boxConfigMinController) + ` .*$`).FindString(string(b))
+	if m == "" {
+		t.Fatalf("the controller CHANGELOG has no `## v%s` heading — boxConfigMinController names a release that does not exist", boxConfigMinController)
+	}
+	if !strings.Contains(m, "box-config") && !strings.Contains(m, "own key") {
+		t.Fatalf("controller v%s's heading does not say it pulls its config with its own key: %s", boxConfigMinController, m)
+	}
+	// And no EARLIER release may claim it (then the constant would be too high, and the lock would arm late).
+	for _, h := range regexp.MustCompile(`(?m)^## v(\d+\.\d+\.\d+) .*$`).FindAllStringSubmatch(string(b), -1) {
+		if h[1] != boxConfigMinController && strings.Contains(h[0], "box-config") {
+			t.Errorf("controller v%s also names box-config: %s", h[1], h[0])
+		}
+	}
+}
@@ -0,0 +1,198 @@
+package api
+
+import (
+	"database/sql"
+	"encoding/json"
+	"net/http"
+	"net/http/httptest"
+	"strings"
+	"testing"
+
+	"gitea.dooplex.hu/admin/felhom-hub/internal/store"
+	"gopkg.in/yaml.v3"
+)
+
+// R-1159 (operator decision D19, 2026-10-19): the household's retrieval passphrase gives keys only until the box uses
+// its own. The measurement of 2026-10-19 (audits/day-2026-10-19b/P2/) turned into the regression: the same calls, now
+// with the answer each must get.
+
+type r1159Tpl struct{}
+
+func (r1159Tpl) Template() string { return "hub: {}\n" }
+
+type r1159Fixture struct {
+	h  *Handler
+	st *store.Store
+	db *sql.DB
+}
+
+func newR1159(t *testing.T) *r1159Fixture {
+	t.Helper()
+	h, st, path := newTestHandler(t)
+	h.templateProvider = r1159Tpl{}
+	if err := st.SaveCustomerConfig(&store.CustomerConfig{CustomerID: "c1", APIKey: "ckey-1159", RetrievalPassword: "owner-pass"}); err != nil {
+		t.Fatal(err)
+	}
+	if err := st.SaveCustomerConfig(&store.CustomerConfig{CustomerID: "c2", APIKey: "ckey-other", RetrievalPassword: "other-pass"}); err != nil {
+		t.Fatal(err)
+	}
+	db, err := sql.Open("sqlite", path)
+	if err != nil {
+		t.Fatal(err)
+	}
+	t.Cleanup(func() { db.Close() })
+	return &r1159Fixture{h: h, st: st, db: db}
+}
+
+func (f *r1159Fixture) exec(t *testing.T, q string, args ...any) {
+	t.Helper()
+	if _, err := f.db.Exec(q, args...); err != nil {
+		t.Fatalf("%s: %v", q, err)
+	}
+}
+
+func (f *r1159Fixture) host(t *testing.T) {
+	t.Helper()
+	if err := f.st.UpsertHost(&store.Host{HostID: "h1", CustomerID: "c1", APIKey: "HKEY-1159"}); err != nil {
+		t.Fatal(err)
+	}
+	f.exec(t, `UPDATE hosts SET created_at = datetime('now', '-1 hour') WHERE host_id = 'h1'`)
+}
+
+// report stores a controller report `ago` SQLite modifier old (e.g. "-5 minutes") naming version.
+func (f *r1159Fixture) report(t *testing.T, version, ago string) {
+	t.Helper()
+	if err := f.st.SaveReport("c1", []byte(`{"customer_id":"c1","controller_version":"`+version+`"}`)); err != nil {
+		t.Fatal(err)
+	}
+	f.exec(t, `UPDATE reports SET received_at = datetime('now', ?) WHERE id = (SELECT max(id) FROM reports)`, ago)
+}
+
+func (f *r1159Fixture) refusals(t *testing.T) int {
+	t.Helper()
+	var n int
+	if err := f.db.QueryRow(`SELECT count(*) FROM events WHERE customer_id='c1' AND event_type=?`, eventPassphraseKeysRefused).Scan(&n); err != nil {
+		t.Fatal(err)
+	}
+	return n
+}
+
+// keysByPassphrase makes the three passphrase calls that ever returned a key and reports, per route, the status and
+// whether a key came back.
+func (f *r1159Fixture) keysByPassphrase(t *testing.T) map[string][2]any {
+	t.Helper()
+	out := map[string][2]any{}
+	rr := r879Get(f.h, "/config/c1", "owner-pass")
+	var y map[string]any
+	_ = yaml.Unmarshal(rr.Body.Bytes(), &y)
+	key := ""
+	if hm, ok := y["hub"].(map[string]any); ok {
+		key, _ = hm["api_key"].(string)
+	}
+	out["config"] = [2]any{rr.Code, key == "ckey-1159"}
+	rr = r879Get(f.h, "/recovery/c1", "owner-pass")
+	out["recovery"] = [2]any{rr.Code, strings.Contains(rr.Body.String(), "ckey-1159")}
+	er := doEnroll(f.h, "c1", "owner-pass")
+	var e enrollResp
+	_ = json.Unmarshal(er.Body.Bytes(), &e)
+	out["enroll"] = [2]any{er.Code, e.APIKey != ""}
+	return out
+}
+
+// THE RULE, case by case.
+//
+// COMPANION RED-PROOF (observed 2026-10-19): with passphraseKeysClosed returning false always (the pre-fix shape) the
+// "live box on 0.334.0" case fails with
+//
+//	r1159_passphrase_scope_test.go: live box on 0.334.0: config → [200 true], want 409 and no key
+//
+// Restored.
+func TestR1159_PassphraseGivesKeysOnlyUntilTheBoxUsesItsOwn(t *testing.T) {
+	open := func(t *testing.T, name string, f *r1159Fixture) {
+		t.Helper()
+		got := f.keysByPassphrase(t)
+		if got["config"] != [2]any{200, true} || got["recovery"] != [2]any{200, true} {
+			t.Errorf("%s: config → %v, recovery → %v, want 200 with the key (the box still needs the passphrase)", name, got["config"], got["recovery"])
+		}
+		if c := got["enroll"][0]; c != 200 && c != 201 {
+			t.Errorf("%s: enroll → %v, want 200/201 with a key", name, got["enroll"])
+		}
+		if f.refusals(t) != 0 {
+			t.Errorf("%s: a refusal event was raised for an allowed call", name)
+		}
+	}
+
+	t.Run("no host record (a first install)", func(t *testing.T) {
+		f := newR1159(t)
+		open(t, "no host", f)
+	})
+	t.Run("host made, its controller has not reported (the install, its first boot)", func(t *testing.T) {
+		f := newR1159(t)
+		f.host(t)
+		open(t, "host, no report", f)
+	})
+	t.Run("only reports from BEFORE the host record (the old box, a reinstall after the delete)", func(t *testing.T) {
+		f := newR1159(t)
+		f.report(t, "0.334.0", "-3 hours")
+		f.host(t)
+		open(t, "older report", f)
+	})
+	t.Run("live box on an older controller (still refreshes with the passphrase)", func(t *testing.T) {
+		f := newR1159(t)
+		f.host(t)
+		f.report(t, "0.333.0", "-5 minutes")
+		open(t, "0.333.0", f)
+	})
+	t.Run("live box on 0.334.0", func(t *testing.T) {
+		f := newR1159(t)
+		f.host(t)
+		f.report(t, "0.334.0", "-5 minutes")
+		got := f.keysByPassphrase(t)
+		for _, route := range []string{"config", "recovery", "enroll"} {
+			if got[route] != [2]any{http.StatusConflict, false} {
+				t.Errorf("live box on 0.334.0: %s → %v, want 409 and no key", route, got[route])
+			}
+		}
+		if n := f.refusals(t); n != 3 {
+			t.Errorf("want one passphrase_keys_refused per refused call (3), got %d", n)
+		}
+		rr := r879Get(f.h, "/config/c1", "owner-pass")
+		if !strings.Contains(rr.Body.String(), "deletes the old host first") || strings.Contains(rr.Body.String(), "ckey-1159") {
+			t.Errorf("the refusal must say what to do and carry no key: %q", rr.Body.String())
+		}
+		// A wrong passphrase is still a plain 401, never the live-box answer (it must not tell a guesser anything).
+		if c := r879Get(f.h, "/config/c1", "nope").Code; c != http.StatusUnauthorized {
+			t.Errorf("a wrong passphrase on a live box → %d, want 401", c)
+		}
+	})
+}
+
+// GET /box-config/{id}: the box's own pull, with its controller key only.
+func TestR1159_BoxPullsItsConfigWithItsOwnKey(t *testing.T) {
+	f := newR1159(t)
+	f.host(t)
+	f.report(t, "0.334.0", "-5 minutes")
+	get := func(id, bearer string) *httptest.ResponseRecorder {
+		req := httptest.NewRequest(http.MethodGet, "/api/v1/box-config/"+id, nil)
+		if bearer != "" {
+			req.Header.Set("Authorization", "Bearer "+bearer)
+		}
+		rr := httptest.NewRecorder()
+		f.h.ServeHTTP(rr, req)
+		return rr
+	}
+	if rr := get("c1", "ckey-1159"); rr.Code != 200 || !strings.Contains(rr.Body.String(), "ckey-1159") {
+		t.Fatalf("own key → %d, want 200 with the config", rr.Code)
+	}
+	for name, c := range map[string]struct{ id, key string }{
+		"no key": {"c1", ""}, "a wrong key": {"c1", "nope"}, "another household's key": {"c1", "ckey-other"},
+		"the passphrase as a bearer": {"c1", "owner-pass"},
+	} {
+		if rr := get(c.id, c.key); rr.Code != http.StatusUnauthorized || strings.Contains(rr.Body.String(), "ckey-1159") {
+			t.Errorf("%s → %d, want 401 and no config", name, rr.Code)
+		}
+	}
+	if f.refusals(t) != 0 {
+		t.Error("the box's own pull raised a refusal event")
+	}
+}
@@ -68,6 +68,7 @@ var apiRoutes = []apiRoute{
 	{Method: "GET", Prefix: "/customers/", Auth: authGlobal},
 	{Method: "GET", Prefix: "/recovery/", Auth: authPassphrase},
 	{Method: "GET", Prefix: "/config/", Auth: authPassphrase},
+	{Method: "GET", Prefix: "/box-config/", Auth: authCustomer},
 	{Method: "POST", Prefix: "/offsite/consume-password/", Auth: authCustomer},
 	{Method: "POST", Prefix: "/offsite/register-key/", Auth: authCustomer},
 	{Method: "POST", Prefix: "/offsite/confirm-key/", Auth: authCustomer},
@@ -82,9 +82,10 @@ func TestOffsiteBox_OversubMath(t *testing.T) {
 }

 // captureEvents records (eventType, severity, customerID) per emit.
-type capturedBox struct{ typ, sev, cust []string }
+type capturedBox struct{ typ, sev, cust, msg []string }

-func (c *capturedBox) fn(cust, et, sev, _, _, _ string) {
+func (c *capturedBox) fn(cust, et, sev, msg, _, _ string) {
+	c.msg = append(c.msg, msg)
 	c.cust = append(c.cust, cust)
 	c.typ = append(c.typ, et)
 	c.sev = append(c.sev, sev)
@@ -6,6 +6,8 @@ import (
 	"errors"
 	"fmt"
 	"log"
+	"sort"
+	"strings"
 	"sync"
 	"time"

@@ -55,6 +57,11 @@ type PBSDRBoxChecker struct {
 	reachAlerted   bool      // an unreachable event has been emitted for the CURRENT outage
 	reachFirstFail time.Time // start of the current outage → the honest blind_for duration
 	lastOK         time.Time // last successful read; zero = never (born blind — do NOT fabricate one)
+
+	// PRUNE COVERAGE (v0.161.0, R-1165) — a THIRD signal: household namespaces on ep0 that no prune job covers keep
+	// every whole-guest copy for ever (box tokens are write-only; pruning is server-side). Alarmed when the set
+	// changes to a non-empty one (a hub restart alarms once, the F2 rule); an empty set re-arms silently.
+	unprunedAlerted string // the sorted, comma-joined set last alarmed; "" = none outstanding
 }

 // pbsdrBoxScope is the customer-less cooldown/display key for PBS-DR box events.
@@ -191,6 +198,35 @@ func (c *PBSDRBoxChecker) Check() {
 		c.emitFill(snap, snap.FillBand)
 	}
 	c.fillBand = snap.FillBand
+
+	c.checkPruneCoverage(usage)
+}
+
+// checkPruneCoverage raises pbsdr_namespace_without_prune_job (R-1165). An unread list (an endpoint before
+// tenantsync v1.3.0, or a job list it could not read) changes nothing — never "all covered".
+func (c *PBSDRBoxChecker) checkPruneCoverage(usage tenantsync.BoxUsage) {
+	if !usage.UnprunedChecked {
+		return
+	}
+	ns := append([]string(nil), usage.Unpruned...)
+	sort.Strings(ns)
+	key := strings.Join(ns, ",")
+	if key == c.unprunedAlerted {
+		return
+	}
+	c.unprunedAlerted = key
+	if key == "" {
+		c.logger.Printf("[INFO] PBS-DR box: every household namespace has a prune job again")
+		return
+	}
+	details, _ := json.Marshal(map[string]any{"scope": pbsdrBoxScope, "namespaces": ns})
+	msg := fmt.Sprintf("PBS DR datastore: %d household namespace(s) have no prune job, so their whole-guest copies are never pruned: %s. "+
+		"Add one per namespace (keep-weekly 2, 03:30 — the shape tenantsync v1.3.0 creates), after reading what it would remove.",
+		len(ns), key)
+	c.logger.Printf("[WARN] PBS-DR box: namespaces without a prune job: %s", key)
+	if c.onEvent != nil {
+		c.onEvent(pbsdrBoxScope, "pbsdr_namespace_without_prune_job", "warning", msg, string(details), "hub")
+	}
 }

 // Snapshot returns the cached aggregate + whether one exists (mutex copy-out). The web layer's only
@@ -3,6 +3,7 @@ package monitor
 import (
 	"context"
 	"fmt"
+	"strings"
 	"sync"
 	"testing"
 	"time"
@@ -158,3 +159,60 @@ func TestPBSDRBox_DegradedKeepsLast(t *testing.T) {
 		t.Fatalf("a failed poll must not emit, got %d", count(ev.typ, "pbsdr_box_fill"))
 	}
 }
+
+// R-1165: household namespaces with no prune job. A non-empty set alarms once (operator scope); the same set again is
+// silent; a changed set alarms again; an empty set re-arms silently; an UNREAD list (old endpoint, unreadable jobs)
+// changes nothing — it never reads as „all covered".
+//
+// COMPANION RED-PROOF (observed 2026-10-19): with checkPruneCoverage's call removed from Check this fails with
+//
+//	pbsdr_box_test.go: two unpruned namespaces → want 1 pbsdr_namespace_without_prune_job, got 0
+//
+// Restored.
+func TestPBSDRBox_R1165_NamespaceWithoutPruneJob(t *testing.T) {
+	f := &fakeUsage{}
+	ev := &capturedBox{}
+	var cur time.Time
+	c := NewPBSDRBoxChecker(f, 80, 90, 0, ev.fn, quietLog())
+	c.now = func() time.Time { return cur }
+	cur = time.Now().UTC()
+	step := func(checked bool, ns ...string) {
+		cur = cur.Add(16 * time.Minute)
+		u := usageBytes(1000, 100)
+		u.UnprunedChecked, u.Unpruned = checked, ns
+		f.set(u, nil)
+		c.Check()
+	}
+	const typ = "pbsdr_namespace_without_prune_job"
+
+	step(false) // an endpoint before v1.3.0: nothing to say
+	if count(ev.typ, typ) != 0 {
+		t.Fatalf("an unread list must not alarm, got %v", ev.typ)
+	}
+	step(true, "tester-1", "Tester-2")
+	if count(ev.typ, typ) != 1 {
+		t.Fatalf("two unpruned namespaces → want 1 %s, got %d", typ, count(ev.typ, typ))
+	}
+	i := len(ev.typ) - 1
+	if ev.sev[i] != "warning" || ev.cust[i] != "pbsdr-box" || !strings.Contains(ev.msg[i], "Tester-2,tester-1") {
+		t.Errorf("the alarm must be an operator-scope warning naming both: sev=%s scope=%s msg=%q", ev.sev[i], ev.cust[i], ev.msg[i])
+	}
+	step(true, "Tester-2", "tester-1") // the same set, another order
+	step(false)                        // an unreadable list in between: no change, no re-arm
+	step(true, "tester-1", "Tester-2")
+	if count(ev.typ, typ) != 1 {
+		t.Fatalf("the same set must not re-alarm, got %d", count(ev.typ, typ))
+	}
+	step(true, "tester-1", "Tester-2", "house-3") // a new household without a job
+	if count(ev.typ, typ) != 2 {
+		t.Fatalf("a changed set must alarm again, got %d", count(ev.typ, typ))
+	}
+	step(true) // every namespace covered: silent re-arm
+	if count(ev.typ, typ) != 2 {
+		t.Fatalf("all covered must be silent, got %d", count(ev.typ, typ))
+	}
+	step(true, "house-3")
+	if count(ev.typ, typ) != 3 {
+		t.Fatalf("after a re-arm a gap must alarm again, got %d", count(ev.typ, typ))
+	}
+}
@@ -896,6 +896,11 @@ var operatorOnlyEvents = map[string]bool{
 	// Listed in the SAME commit that mints them.
 	"host_reinstall_over_old_host": true,
 	"host_reenroll_while_live":     true,
+	// R-1159 (v0.161.0): the retrieval passphrase asked for a live box's keys and was refused (api/passphrase_scope.go).
+	// The household must not be mailed "someone may know your passphrase" by a hub rule; the operator decides.
+	// R-1165 (v0.161.0): an ep0 household namespace with no prune job (monitor/pbsdr_box.go). Hub-origin only.
+	"passphrase_keys_refused":           true,
+	"pbsdr_namespace_without_prune_job": true,
 }

 // IsOperatorOnly reports whether an event type is barred from customer dispatch. Exported so the
@@ -1263,6 +1263,22 @@ func (s *Store) LatestControllerReportJSON(customerID string) (string, error) {
 	return reportJSON, err
 }

+// LatestControllerReport returns when the customer's newest controller report arrived and the controller version it
+// named (R-1159: the passphrase scope reads it). found=false when the customer never reported.
+func (s *Store) LatestControllerReport(customerID string) (found bool, receivedAt time.Time, version string, err error) {
+	var recv string
+	var ver sql.NullString
+	err = s.db.QueryRow(`SELECT received_at, controller_version FROM reports WHERE customer_id = ? ORDER BY id DESC LIMIT 1`,
+		customerID).Scan(&recv, &ver)
+	if err == sql.ErrNoRows {
+		return false, time.Time{}, "", nil
+	}
+	if err != nil {
+		return false, time.Time{}, "", err
+	}
+	return true, parseSQLiteTime(recv), ver.String, nil
+}
+
 // LatestReportOffsitePresence reports whether the customer's most recent controller report exists
 // and whether it carries an offsite status object (R-70 detector input).
 func (s *Store) LatestReportOffsitePresence(customerID string) (found bool, receivedAt time.Time, hasOffsite bool, err error) {
@@ -150,6 +150,10 @@ type BoxUsage struct {
 	Total int64
 	Used  int64
 	Avail int64
+	// R-1165 (tenantsync v1.3.0): the top-level namespaces no prune job covers. UnprunedChecked=false when the
+	// endpoint could not read its job list, or predates v1.3.0 — then Unpruned says nothing.
+	Unpruned        []string
+	UnprunedChecked bool
 }

 // Usage returns the felhom-offsite datastore's total/used/avail from the endpoint — a read-only op
@@ -164,7 +168,7 @@ func (c *Client) Usage(ctx context.Context) (BoxUsage, error) {
 	if resp.Total <= 0 {
 		return BoxUsage{}, fmt.Errorf("tenantsync: usage op returned a non-positive total")
 	}
-	return BoxUsage{Total: resp.Total, Used: resp.Used, Avail: resp.Avail}, nil
+	return BoxUsage{Total: resp.Total, Used: resp.Used, Avail: resp.Avail, Unpruned: resp.Unpruned, UnprunedChecked: resp.UnprunedChecked}, nil
 }

 func (c *Client) tenancyOp(ctx context.Context, op, customerID string) (*Result, error) {
@@ -199,6 +203,9 @@ type response struct {
 	Total int64 `json:"total"`
 	Used  int64 `json:"used"`
 	Avail int64 `json:"avail"`
+	// usage op (v1.3.0, R-1165): namespaces with no prune job, and whether that list was read.
+	Unpruned        []string `json:"unpruned"`
+	UnprunedChecked bool     `json:"unpruned_checked"`
 	Result
 }

@@ -225,6 +225,32 @@ func TestUsage_Op(t *testing.T) {
 	}
 }

+// R-1165 (tenantsync v1.3.0): the usage answer's prune coverage reaches the checker; an older endpoint (no field)
+// reads as NOT checked, never as „all covered".
+func TestUsage_R1165_PruneCoverage(t *testing.T) {
+	for _, c := range []struct {
+		name, body string
+		checked    bool
+		unpruned   []string
+	}{
+		{"v1.3.0", `{"status":"ok","total":10,"used":1,"avail":9,"unpruned_checked":true,"unpruned":["Tester-2","tester-1"]}`, true, []string{"Tester-2", "tester-1"}},
+		{"v1.3.0, unread", `{"status":"ok","total":10,"used":1,"avail":9,"unpruned_checked":false}`, false, nil},
+		{"v1.2.0", `{"status":"ok","total":10,"used":1,"avail":9}`, false, nil},
+	} {
+		clientPEM, clientSigner := testKeys(t)
+		_, hostSigner := testKeys(t)
+		srv := startTestServer(t, hostSigner, clientSigner, c.body, "", 0)
+		cl := newTestClient(t, srv.addr, hostKeyLine(t, hostSigner), clientPEM)
+		u, err := cl.Usage(context.Background())
+		if err != nil {
+			t.Fatalf("%s: Usage: %v", c.name, err)
+		}
+		if u.UnprunedChecked != c.checked || strings.Join(u.Unpruned, ",") != strings.Join(c.unpruned, ",") {
+			t.Errorf("%s: checked=%v unpruned=%v, want %v %v", c.name, u.UnprunedChecked, u.Unpruned, c.checked, c.unpruned)
+		}
+	}
+}
+
 // TestUsage_UnknownOpTypedUnsupported (v0.65.0, graceful degradation): an ep0 still on tenantsync
 // ≤ v1.1.0 answers `{"op":"usage"}` with its generic `bad_request "unknown op"` (exit 1). That must
 // map to the typed ErrUsageUnsupported so the PBS checker records "unavailable", not an error.
@@ -1,7 +1,16 @@
 #!/usr/bin/env bash
-# felhom-tenantsync v1.2.0 — the offsite endpoint's per-customer PBS tenancy surface (PBS DR tier
+# felhom-tenantsync v1.3.0 — the offsite endpoint's per-customer PBS tenancy surface (PBS DR tier
 # SLICE 1; spike SPIKE-pbs-tier-provisioning-2026-07-10 §3).
 #
+# v1.3.0 (2026-10-19, R-1165): a household's namespace gets its PRUNE JOB with it. provision/reissue ensure
+#   `prune-<id>` (keep-weekly 2, 03:30, ns-scoped, max-depth 0 — the shape of the demo-hp/demo-felhom jobs, STATUS
+#   D18) unless a job on felhom-offsite already covers that namespace (idempotent: a second run changes nothing);
+#   a failed create is logged and never fails the provision (the hub's alarm below names the namespace). deprovision
+#   removes the namespace's jobs before the namespace. `usage` adds `unpruned` — the top-level namespaces no job
+#   covers (read-only: the namespace folders + `prune-job list`), or `unpruned_checked:false` when the list cannot be
+#   read. Box tokens stay write-only: pruning is server-side, so a namespace without a job grows for ever.
+#   Pinned by scripts/tenantsync-prune-harness.sh.
+#
 # v1.2.0 (2026-07-17, hub R-5 PBS visibility): + read-only `usage` op — the felhom-offsite datastore's
 #   total/used/avail (bytes, via `df` on the datastore path). No customer_id, no admin token, NO
 #   mutation. Structural twin of `fingerprint`. Backward-compatible: an older hub never sends it, and a
@@ -28,7 +37,8 @@
 #       This is the DELIBERATE, gated data-destruction the slice-1 note reserved — the operator RESET
 #       confirm (typed customer-id + separate escrow-custody ack) is the human decision it demanded.
 #   {"op":"fingerprint"}                     → {"status":"ok","fingerprint":"<PBS cert sha256>"}
-#   {"op":"usage"}                           → {"status":"ok","total":<bytes>,"used":<bytes>,"avail":<bytes>}
+#   {"op":"usage"}                           → {"status":"ok","total":<bytes>,"used":<bytes>,"avail":<bytes>,
+#                                                "unpruned":[<ns>…],"unpruned_checked":true}
 #       READ-ONLY datastore fill (df on the felhom-offsite path). No customer_id, no admin token, no mutation.
 #
 # Secret hygiene (load-bearing):
@@ -85,7 +95,21 @@ if [ "$OP" = "usage" ]; then
  # df -B1: 1-byte blocks. --output pins the columns; tail -1 skips the header row.
  read -r TOTAL USED AVAIL < <(df -B1 --output=size,used,avail "$DS_PATH" | tail -1)
  [ -n "$TOTAL" ] && [ -n "$USED" ] && [ -n "$AVAIL" ] || err_json internal "could not read datastore usage"
-  printf '{"status":"ok","total":%s,"used":%s,"avail":%s}\n' "$TOTAL" "$USED" "$AVAIL"
+  # v1.3.0 (R-1165): the top-level namespaces no prune job covers. A job covers a namespace when it names it, or when
+  # it sits on the root (no ns) without a max-depth of 0. Read-only; a list that cannot be read says so.
+  NSES='[]'
+  if [ -d "$DS_PATH/ns" ]; then
+    NSES=$(find "$DS_PATH/ns" -mindepth 1 -maxdepth 1 -type d -printf '%f\n' | jq -R . | jq -cs .)
+  fi
+  if JOBS=$(proxmox-backup-manager prune-job list --output-format json 2>/dev/null) && jq -e 'type == "array"' >/dev/null 2>&1 <<<"$JOBS"; then
+    jq -cn --argjson t "$TOTAL" --argjson u "$USED" --argjson a "$AVAIL" --argjson ns "$NSES" --argjson jobs "$JOBS" --arg ds "$DS" '
+      ($jobs | map(select(.store == $ds))) as $j
+      | ($j | any(((.ns // "") == "") and ((."max-depth" // 7) > 0))) as $root
+      | {"status":"ok","total":$t,"used":$u,"avail":$a,"unpruned_checked":true,
+         "unpruned": (if $root then [] else [$ns[] as $n | select(($j | any((.ns // "") == $n)) | not) | $n] | sort end)}'
+  else
+    printf '{"status":"ok","total":%s,"used":%s,"avail":%s,"unpruned_checked":false}\n' "$TOTAL" "$USED" "$AVAIL"
+  fi
  exit 0
 fi

@@ -126,6 +150,11 @@ if [ "$OP" = "deprovision" ]; then
  # residual namespace ACLs (belt-and-suspenders — the user grant is not token-scoped).
  proxmox-backup-manager acl update "/datastore/$DS/$CID" DatastoreBackup --auth-id "$PBS_USER" --delete >&2 2>/dev/null || true
  proxmox-backup-manager acl update "/datastore/$DS/$CID" DatastoreBackup --auth-id "$TOKEN_ID" --delete >&2 2>/dev/null || true
+  # v1.3.0 (R-1165): the namespace's prune job(s) go first — a job left on a deleted namespace fails every night.
+  for job in $(proxmox-backup-manager prune-job list --output-format json 2>/dev/null \
+      | jq -r --arg ns "$CID" --arg ds "$DS" '.[] | select(.store == $ds and (.ns // "") == $ns) | .id'); do
+    proxmox-backup-manager prune-job remove "$job" >&2 && log "deprovision: prune job $job removed"
+  done
  ns_existed=false
  if PBS_PASSWORD="$ADM" proxmox-backup-client namespace list --repository "$ADMIN_REPO" \
      --output-format json | jq -e --arg ns "$CID" '(.data // .) | any(.[]; .ns == $ns)' >/dev/null; then
@@ -155,6 +184,23 @@ if ! PBS_PASSWORD="$ADM" proxmox-backup-client namespace list --repository "$ADM
  log "created namespace $CID"
 fi

+# 3b. v1.3.0 (R-1165): the namespace's prune job — keep-weekly 2 at 03:30, the D18 shape. Idempotent: any job on this
+#     datastore that already names the namespace counts (whatever its id). A PBS job id is at most 32 characters, so a
+#     long customer id gets a shortened, hashed one. A failure is logged, never fatal (the hub's usage alarm names it).
+JOB_ID="prune-$CID"
+if [ "${#JOB_ID}" -gt 32 ]; then
+  JOB_ID="prune-${CID:0:20}-$(printf '%s' "$CID" | sha256sum | cut -c1-5)"
+fi
+if proxmox-backup-manager prune-job list --output-format json \
+    | jq -e --arg ns "$CID" --arg ds "$DS" 'any(.[]; .store == $ds and (.ns // "") == $ns)' >/dev/null; then
+  log "prune job for namespace $CID already present"
+elif proxmox-backup-manager prune-job create "$JOB_ID" --store "$DS" --ns "$CID" --max-depth 0 --keep-weekly 2 \
+    --schedule 03:30 --comment "R-1165: one copy per ISO week, 2 weeks, server-side (box tokens are write-only)" >&2; then
+  log "created prune job $JOB_ID (keep-weekly 2, 03:30) for namespace $CID"
+else
+  log "WARNING: could not create prune job $JOB_ID for namespace $CID — its copies are not pruned until one exists"
+fi
+
 token_exists() {
  proxmox-backup-manager user list-tokens "$PBS_USER" --output-format json \
    | jq -e --arg t "$TOKEN_ID" 'any(.[]; .tokenid == $t)' >/dev/null
@@ -23,6 +23,7 @@ echo "pbm $*" >> "$CALLLOG"
 case "$1 $2" in
  "cert info")        echo "Fingerprint (sha256): aa:bb:cc:dd" ;;
  "datastore list")   printf '[{"name":"scratch","path":"/srv/pbs-scratch"},{"name":"felhom-offsite","path":"%s"}]\n' "$DS_PATH" ;;
+  "prune-job list")   echo '[]' ;;   # read-only (v1.3.0 usage reads it for `unpruned`)
  "user generate-token") echo "MUTATION: $*" >> "$MUTLOG"; echo '{"value":"FAKE-ADMIN-TOKEN"}' ;;
  "user delete-token")   echo "MUTATION: $*" >> "$MUTLOG" ;;
  "acl update")          echo "MUTATION: $*" >> "$MUTLOG" ;;
@@ -54,7 +55,7 @@ fail() { echo "FAIL: $*"; RC=1; }
 # --- Test 1: usage op ---
 OUT=$(echo '{"op":"usage"}' | PATH="$STUB:$PATH" bash "$SCRIPT" 2>/dev/null); EC=$?
 [ "$EC" -eq 0 ] || fail "usage exit $EC (want 0)"
-EXP='{"status":"ok","total":41943040,"used":8388608,"avail":33554432}'
+EXP='{"status":"ok","total":41943040,"used":8388608,"avail":33554432,"unpruned_checked":true,"unpruned":[]}'
 [ "$OUT" = "$EXP" ] || fail "usage stdout = $OUT (want $EXP)"
 if [ -s "$MUTLOG" ]; then fail "usage ran mutations: $(cat "$MUTLOG")"; else echo "PASS: usage emits totals, exit 0, ZERO mutations"; fi

@@ -0,0 +1,181 @@
+#!/usr/bin/env python3
+# -*- coding: utf-8 -*-
+"""test_tenantsync_prune.py — felhom-tenantsync v1.3.0 gives every household namespace its prune job (R-1165).
+
+Runs the REAL script against stubbed `proxmox-backup-manager` / `proxmox-backup-client` / `df` on PATH (real bash, real
+jq). The stubs keep the prune jobs in a JSON file and LOG every call, so each case asserts what the script DID:
+
+  1. provision of a new household → exactly one `prune-job create prune-<id> --store felhom-offsite --ns <id>
+     --max-depth 0 --keep-weekly 2 --schedule 03:30`, and the provision still succeeds;
+  2. a second run (reissue) with the job present → no create (idempotent); a job under ANOTHER id that already names
+     the namespace counts too;
+  3. a 31-character customer id → a job id of at most 32 characters (PBS's limit);
+  4. a create that fails → logged, the provision still succeeds (the alarm names the namespace);
+  5. deprovision → the namespace's job removed, another household's kept;
+  6. usage → `unpruned` = the namespaces no job covers; a root job without max-depth 0 covers all; an unreadable job
+     list → `unpruned_checked: false`, never an empty list that reads as „all covered".
+
+Exit 0 all passed · 1 a case failed. Without bash or jq it says NOT CHECKED out loud and exits 0 (the CI runner may
+lack jq; the workspace pre-push run has both).
+"""
+import json
+import os
+import shutil
+import subprocess
+import sys
+import tempfile
+
+HERE = os.path.dirname(os.path.abspath(__file__))
+SCRIPT = os.path.join(HERE, "felhom-tenantsync.sh")
+
+PBM = r'''#!/usr/bin/env bash
+echo "pbm $*" >> "$CALLLOG"
+case "$1 $2" in
+  "cert info")            echo "Fingerprint (sha256): aa:bb:cc:dd" ;;
+  "datastore list")       printf '[{"name":"felhom-offsite","path":"%s"}]\n' "$DS_PATH" ;;
+  "user generate-token")  echo '{"value": "FAKE-TOKEN-VALUE"}' ;;
+  "user list")            echo '[{"userid":"felhom@pbs"}]' ;;
+  "user list-tokens")     echo '[]' ;;
+  "prune-job list")       [ -n "${JOBS_FAIL:-}" ] && exit 1; cat "$JOBS" ;;
+  "prune-job create")     [ -n "${CREATE_FAIL:-}" ] && exit 1
+                          id="$3"; shift 3; ns=""
+                          while [ $# -gt 0 ]; do [ "$1" = "--ns" ] && ns="$2"; shift; done
+                          jq -c --arg id "$id" --arg ns "$ns" '. + [{"id":$id,"store":"felhom-offsite","ns":$ns,"keep-weekly":2,"schedule":"03:30"}]' "$JOBS" > "$JOBS.n" && mv "$JOBS.n" "$JOBS" ;;
+  "prune-job remove")     jq -c --arg id "$3" 'map(select(.id != $id))' "$JOBS" > "$JOBS.n" && mv "$JOBS.n" "$JOBS" ;;
+  *)                      : ;;
+esac
+'''
+PBC = r'''#!/usr/bin/env bash
+echo "pbc $*" >> "$CALLLOG"
+case "$1 $2" in
+  "namespace list")   echo '{"data":[]}' ;;
+  *)                  : ;;
+esac
+'''
+DF = '#!/usr/bin/env bash\necho "1B-blocks Used Avail"\necho "41943040 8388608 33554432"\n'
+
+fails = []
+ran = 0
+
+
+class Box(object):
+    def __init__(self):
+        self.dir = tempfile.mkdtemp(prefix="tenantsync-prune-")
+        self.bin = os.path.join(self.dir, "bin")
+        os.makedirs(self.bin)
+        for name, body in (("proxmox-backup-manager", PBM), ("proxmox-backup-client", PBC), ("df", DF)):
+            p = os.path.join(self.bin, name)
+            open(p, "w").write(body)
+            os.chmod(p, 0o755)
+        self.ds = os.path.join(self.dir, "ds")
+        os.makedirs(os.path.join(self.ds, "ns"))
+        self.jobs = os.path.join(self.dir, "jobs.json")
+        self.calls = os.path.join(self.dir, "calls.log")
+        self.set_jobs([])
+
+    def set_jobs(self, jobs):
+        open(self.jobs, "w").write(json.dumps(jobs))
+
+    def get_jobs(self):
+        return json.load(open(self.jobs))
+
+    def run(self, payload, **env):
+        open(self.calls, "w").close()
+        e = dict(os.environ, PATH=self.bin + os.pathsep + os.environ["PATH"], DS_PATH=self.ds, JOBS=self.jobs,
+                 CALLLOG=self.calls)
+        e.update(env)
+        p = subprocess.run(["bash", SCRIPT], input=json.dumps(payload), env=e, capture_output=True, text=True)
+        return p.returncode, p.stdout, p.stderr, open(self.calls).read().splitlines()
+
+    def close(self):
+        shutil.rmtree(self.dir, ignore_errors=True)
+
+
+def check(name, cond, detail=""):
+    global ran
+    ran += 1
+    if cond:
+        print("ok   %s" % name)
+    else:
+        fails.append("%s %s" % (name, detail))
+
+
+def creates(calls):
+    return [c for c in calls if c.startswith("pbm prune-job create")]
+
+
+def main():
+    if not shutil.which("bash") or not shutil.which("jq"):
+        print("test_tenantsync_prune: NOT CHECKED — bash or jq is missing here (the workspace run has both)")
+        return 0
+
+    b = Box()
+    try:
+        rc, out, err, calls = b.run({"op": "provision", "customer_id": "house-1"})
+        cr = creates(calls)
+        check("provision/creates-the-job", rc == 0 and len(cr) == 1 and cr[0] == (
+            "pbm prune-job create prune-house-1 --store felhom-offsite --ns house-1 --max-depth 0 --keep-weekly 2 "
+            "--schedule 03:30 --comment R-1165: one copy per ISO week, 2 weeks, server-side (box tokens are write-only)"),
+            "rc=%d creates=%s err=%s" % (rc, cr, err[-300:]))
+        check("provision/still-answers-ok", '"status":"ok"' in out and "token_secret" in out, out[-200:])
+
+        rc, out, err, calls = b.run({"op": "reissue", "customer_id": "house-1"})
+        check("reissue/idempotent", rc == 0 and not creates(calls) and len(b.get_jobs()) == 1,
+              "rc=%d creates=%s jobs=%s" % (rc, creates(calls), b.get_jobs()))
+
+        b.set_jobs([{"id": "hand-made", "store": "felhom-offsite", "ns": "house-2", "keep-weekly": 2}])
+        rc, out, err, calls = b.run({"op": "provision", "customer_id": "house-2"})
+        check("provision/a-job-under-another-id-counts", rc == 0 and not creates(calls), "creates=%s" % creates(calls))
+
+        long_id = "a" * 31
+        rc, out, err, calls = b.run({"op": "provision", "customer_id": long_id})
+        cr = creates(calls)
+        jid = cr[0].split()[3] if cr else ""
+        check("provision/long-id-fits-32", rc == 0 and 0 < len(jid) <= 32 and jid.startswith("prune-"), "id=%r" % jid)
+
+        rc, out, err, calls = b.run({"op": "provision", "customer_id": "house-3"}, CREATE_FAIL="1")
+        check("provision/create-failure-is-not-fatal", rc == 0 and '"status":"ok"' in out and "could not create prune job" in err,
+              "rc=%d out=%s err=%s" % (rc, out[-120:], err[-200:]))
+
+        b.set_jobs([{"id": "prune-house-1", "store": "felhom-offsite", "ns": "house-1"},
+                    {"id": "prune-house-9", "store": "felhom-offsite", "ns": "house-9"}])
+        rc, out, err, calls = b.run({"op": "deprovision", "customer_id": "house-1"})
+        ids = [j["id"] for j in b.get_jobs()]
+        check("deprovision/removes-only-its-job", rc == 0 and ids == ["prune-house-9"], "rc=%d ids=%s" % (rc, ids))
+
+        for n in ("demo-hp", "tester-1", "Tester-2"):
+            os.makedirs(os.path.join(b.ds, "ns", n))
+        b.set_jobs([{"id": "prune-demo-hp", "store": "felhom-offsite", "ns": "demo-hp", "max-depth": 0},
+                    {"id": "other-store", "store": "elsewhere", "ns": "tester-1"}])
+        rc, out, err, calls = b.run({"op": "usage"})
+        u = json.loads(out) if out.strip().startswith("{") else {}
+        check("usage/names-the-unpruned", rc == 0 and u.get("unpruned_checked") is True and u.get("unpruned") == ["Tester-2", "tester-1"]
+              and u.get("total") == 41943040, out)
+        check("usage/no-mutation", not [c for c in calls if any(k in c for k in ("create", "remove", "generate-token", "acl"))], calls)
+
+        b.set_jobs([{"id": "root", "store": "felhom-offsite", "keep-weekly": 2}])
+        rc, out, err, calls = b.run({"op": "usage"})
+        u = json.loads(out) if out.strip().startswith("{") else {}
+        check("usage/a-root-job-covers-all", u.get("unpruned") == [] and u.get("unpruned_checked") is True, out)
+
+        b.set_jobs([{"id": "root0", "store": "felhom-offsite", "max-depth": 0}])
+        rc, out, err, calls = b.run({"op": "usage"})
+        u = json.loads(out) if out.strip().startswith("{") else {}
+        check("usage/a-root-job-with-depth-0-covers-none", u.get("unpruned") == ["Tester-2", "demo-hp", "tester-1"], out)
+
+        rc, out, err, calls = b.run({"op": "usage"}, JOBS_FAIL="1")
+        u = json.loads(out) if out.strip().startswith("{") else {}
+        check("usage/unreadable-jobs-say-so", rc == 0 and u.get("unpruned_checked") is False and "unpruned" not in u, out)
+    finally:
+        b.close()
+
+    if fails:
+        for f in fails:
+            print("FAIL: %s" % f)
+        return 1
+    print("test_tenantsync_prune: all %d cases passed" % ran)
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())