v0.80.0: PBS DR tier slice 2 — the apply-bridge (pbs_dr consumer, felhom-pbs-apply set-only wrapper, verify-pin-before-consume, adoption-first, loud consumed-failed, escrow seed)

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01PSK5g6qYLknKj8u3QAFEr6
2026-07-10 21:50:36 +02:00
parent a6e8bcb475
commit e5e8f3920a
15 changed files with 1309 additions and 2 deletions
@@ -0,0 +1,82 @@
+package pbsdr
+
+import (
+	"context"
+	"log/slog"
+	"sync"
+	"time"
+
+	"gitea.dooplex.hu/admin/felhom-agent/internal/hub"
+)
+
+// Loop drives the Manager on its own cadence AND consumes fetched desired-state via the
+// desired.Syncer raw-consumer seam (the wgtunnel Loop shape). fetched=false ("no desired data
+// seen yet") is never a signal; an absent pbs_dr block on a PRESENT desired-state is a plain
+// no-op this slice (no teardown — deprovision is a deliberate future op).
+type Loop struct {
+	mgr      *Manager
+	interval time.Duration
+	logger   *slog.Logger
+
+	mu      sync.Mutex
+	fetched bool
+	block   *hub.WirePBSDR
+
+	nudge chan struct{}
+}
+
+// NewLoop builds the loop. interval defaults to 60s.
+func NewLoop(mgr *Manager, interval time.Duration, logger *slog.Logger) *Loop {
+	if interval <= 0 {
+		interval = 60 * time.Second
+	}
+	if logger == nil {
+		logger = slog.Default()
+	}
+	return &Loop{mgr: mgr, interval: interval, logger: logger, nudge: make(chan struct{}, 1)}
+}
+
+// OnDesiredState implements desired.RawConsumer: store the latest pbs_dr block (or its absence)
+// and nudge the loop. Non-blocking and panic-free by construction.
+func (l *Loop) OnDesiredState(_ context.Context, resp *hub.DesiredStateResponse) {
+	if resp == nil {
+		return
+	}
+	l.mu.Lock()
+	l.fetched = true
+	l.block = resp.DesiredState.PBSDR
+	l.mu.Unlock()
+	select {
+	case l.nudge <- struct{}{}:
+	default:
+	}
+}
+
+func (l *Loop) snapshot() (bool, *hub.WirePBSDR) {
+	l.mu.Lock()
+	defer l.mu.Unlock()
+	return l.fetched, l.block
+}
+
+// Run applies immediately, then on every tick or desired-state nudge, until ctx is cancelled.
+func (l *Loop) Run(ctx context.Context) error {
+	fetched, block := l.snapshot()
+	l.mgr.Apply(ctx, fetched, block)
+	t := time.NewTicker(l.interval)
+	defer t.Stop()
+	for {
+		select {
+		case <-ctx.Done():
+			return ctx.Err()
+		case <-t.C:
+		case <-l.nudge:
+		}
+		fetched, block = l.snapshot()
+		l.mgr.Apply(ctx, fetched, block)
+	}
+}
+
+// PBSDRStatus implements the hub collector's PBSDRReporter seam.
+func (l *Loop) PBSDRStatus(_ context.Context) *hub.PBSDRStatus {
+	return l.mgr.Status()
+}
@@ -0,0 +1,428 @@
+// Package pbsdr is the PBS-DR-tier apply-bridge (slice 2) — the host-side twin of the
+// controller's offsite apply-bridge. It consumes the hub's `pbs_dr` desired-state descriptor
+// (slice 1) and converges the box: the pbs storage entry exists, K exists, the agent token can
+// write to it, and the ceremony one-liner finds its storage id.
+//
+// The laws this package encodes (spike SPIKE-pbs-tier-provisioning-2026-07-10 + the offsite
+// bridge precedent — do not "simplify" any of them):
+//
+//   - SET-ONLY: re-apply never removes the storage entry — entry deletion destroys K
+//     (un-decryptable backups). The wrapper has no deletion verb; this package never asks for one.
+//   - ADOPTION FIRST: an existing healthy entry is adopted non-destructively — verified, granted,
+//     marked — with NO consume. Tenancy identity (namespace/username/datastore) is ENTRY-OWNED on
+//     adoption; a descriptor that names a different namespace does not repoint a live entry (the
+//     demo's manually-built felhom-offsite tenancy is the canonical case).
+//   - VERIFY-PIN-BEFORE-CONSUME: the PBS fingerprint is probed over the tunnel against the
+//     descriptor BEFORE the one-time secret is consumed (the offsite ordering law).
+//   - CONSUMED-BUT-FAILED IS LOUD: after a consume, any failure lands in a persistent alarming
+//     report state. The bridge never silently retries a burned secret; it recovers ONLY when the
+//     operator stages a fresh one (hub Re-issue) — the next consume then succeeds.
+//   - The secret rides the wrapper's STDIN (sudo logs argv) and is never logged.
+package pbsdr
+
+import (
+	"context"
+	"crypto/sha256"
+	"encoding/hex"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"log/slog"
+	"net/netip"
+	"os"
+	"path/filepath"
+	"regexp"
+	"strings"
+	"sync"
+	"time"
+
+	"gitea.dooplex.hu/admin/felhom-agent/internal/hub"
+	"gitea.dooplex.hu/admin/felhom-agent/internal/pbs"
+	"gitea.dooplex.hu/admin/felhom-agent/internal/proxmox"
+)
+
+// WrapperPath is the pinned sudoers vector (configs/felhom-pbs-apply).
+const WrapperPath = "/usr/local/sbin/felhom-pbs-apply"
+
+// StorageReader is the PVE read seam (satisfied by *proxmox.Client; tests fake it).
+type StorageReader interface {
+	StorageEntry(ctx context.Context, id string) (*proxmox.StorageEntryConfig, bool, error)
+	StorageActive(ctx context.Context, id string) (bool, error)
+}
+
+// SecretConsumer is the hub consume-once seam (satisfied by *hub.Client; tests fake it).
+type SecretConsumer interface {
+	ConsumePBSToken(ctx context.Context) (string, error)
+}
+
+var storageIDRe = regexp.MustCompile(`^[A-Za-z][A-Za-z0-9_.-]{0,27}$`)
+var nameRe = regexp.MustCompile(`^[A-Za-z0-9][A-Za-z0-9_.-]{0,31}$`)
+var tokenIDRe = regexp.MustCompile(`^[A-Za-z0-9_.-]+@[A-Za-z0-9]+![A-Za-z0-9_.-]+$`)
+var fingerprintRe = regexp.MustCompile(`^([A-Fa-f0-9]{2}:){31}[A-Fa-f0-9]{2}$`)
+
+const markerName = "marker.json"
+const consumedFailedName = "consumed-failed.json"
+
+// marker is the idempotency record: the descriptor hash last converged + how.
+type marker struct {
+	Hash      string `json:"hash"`
+	State     string `json:"state"` // "adopted" | "applied"
+	AppliedAt string `json:"applied_at"`
+}
+
+// consumedFailed is the LOUD persistent state: a one-time secret was burned and the apply failed.
+type consumedFailed struct {
+	Hash    string `json:"hash"`
+	Message string `json:"message"`
+	At      string `json:"at"`
+}
+
+// Manager converges the box toward the pbs_dr descriptor. All deps are seams for tests.
+type Manager struct {
+	runner     proxmox.Runner
+	px         StorageReader
+	hub        SecretConsumer
+	stateDir   string // <agent-state>/pbsdr
+	secretDir  string // cfg.Backup pbs secret dir (the wrapper's copy target)
+	configPath string // agent.json — for the escrow.pbs_storage_id seed ("" = no seeding)
+	logger     *slog.Logger
+
+	// probeFP is the verify-pin-before-consume seam (default pbs.ProbeFingerprint).
+	probeFP func(ctx context.Context, server, fingerprint string) error
+	now     func() time.Time
+
+	mu     sync.Mutex
+	status *hub.PBSDRStatus // latest snapshot for the report stanza
+}
+
+// NewManager builds the bridge manager.
+func NewManager(runner proxmox.Runner, px StorageReader, hubc SecretConsumer, stateDir, secretDir, configPath string, logger *slog.Logger) *Manager {
+	if logger == nil {
+		logger = slog.Default()
+	}
+	return &Manager{
+		runner: runner, px: px, hub: hubc,
+		stateDir: filepath.Join(stateDir, "pbsdr"), secretDir: secretDir, configPath: configPath,
+		logger:  logger,
+		probeFP: pbs.ProbeFingerprint,
+		now:     func() time.Time { return time.Now().UTC() },
+	}
+}
+
+func (m *Manager) markerPath() string         { return filepath.Join(m.stateDir, markerName) }
+func (m *Manager) consumedFailedPath() string { return filepath.Join(m.stateDir, consumedFailedName) }
+
+func (m *Manager) setStatus(s *hub.PBSDRStatus) {
+	m.mu.Lock()
+	m.status = s
+	m.mu.Unlock()
+}
+
+// Status returns the latest bridge state (the report stanza; nil before the first Apply).
+func (m *Manager) Status() *hub.PBSDRStatus {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	return m.status
+}
+
+// descriptorHash is the idempotency key: sha256 of the canonical (struct-ordered) JSON.
+func descriptorHash(b *hub.WirePBSDR) string {
+	j, _ := json.Marshal(b)
+	sum := sha256.Sum256(j)
+	return hex.EncodeToString(sum[:])
+}
+
+func (m *Manager) loadMarker() *marker {
+	raw, err := os.ReadFile(m.markerPath())
+	if err != nil {
+		return nil
+	}
+	var mk marker
+	if json.Unmarshal(raw, &mk) != nil || mk.Hash == "" {
+		return nil
+	}
+	return &mk
+}
+
+func (m *Manager) loadConsumedFailed() *consumedFailed {
+	raw, err := os.ReadFile(m.consumedFailedPath())
+	if err != nil {
+		return nil
+	}
+	var cf consumedFailed
+	if json.Unmarshal(raw, &cf) != nil {
+		return nil
+	}
+	return &cf
+}
+
+func (m *Manager) writeState(path string, v any) error {
+	if err := os.MkdirAll(m.stateDir, 0o700); err != nil {
+		return err
+	}
+	raw, err := json.Marshal(v)
+	if err != nil {
+		return err
+	}
+	tmp := path + ".tmp"
+	if err := os.WriteFile(tmp, raw, 0o600); err != nil {
+		return err
+	}
+	return os.Rename(tmp, path)
+}
+
+// validate checks every descriptor field BEFORE any exec (the fine gate under the coarse sudoers).
+func validate(b *hub.WirePBSDR) error {
+	if !storageIDRe.MatchString(b.StorageID) {
+		return fmt.Errorf("bad storage_id %q", b.StorageID)
+	}
+	if !nameRe.MatchString(b.Datastore) {
+		return fmt.Errorf("bad datastore %q", b.Datastore)
+	}
+	if !nameRe.MatchString(b.Namespace) {
+		return fmt.Errorf("bad namespace %q", b.Namespace)
+	}
+	if !tokenIDRe.MatchString(b.TokenID) {
+		return fmt.Errorf("bad token_id %q", b.TokenID)
+	}
+	if !fingerprintRe.MatchString(b.Fingerprint) {
+		return fmt.Errorf("bad fingerprint (want 32-pair colon sha256)")
+	}
+	if _, err := netip.ParseAddr(b.PBSTunnelIP); err != nil {
+		return fmt.Errorf("bad pbs_tunnel_ip %q", b.PBSTunnelIP)
+	}
+	return nil
+}
+
+// Apply converges toward the descriptor. fetched=false (no desired data yet) is never a signal.
+func (m *Manager) Apply(ctx context.Context, fetched bool, block *hub.WirePBSDR) {
+	if !fetched {
+		return
+	}
+	if block == nil {
+		// Absent block: pre-slice-1 hub or tier never enabled — a silent no-op (old-hub compat).
+		return
+	}
+	if !block.Enabled {
+		m.setStatus(&hub.PBSDRStatus{State: "disabled", StorageID: block.StorageID, Namespace: block.Namespace})
+		return // NO teardown in this slice — deprovision is a deliberate future op
+	}
+	if err := validate(block); err != nil {
+		m.logger.Error("pbsdr: descriptor invalid; refusing", "err", err)
+		m.setStatus(&hub.PBSDRStatus{State: "verify_failed", StorageID: block.StorageID, Namespace: block.Namespace,
+			Message: "descriptor invalid: " + err.Error()})
+		return
+	}
+
+	h := descriptorHash(block)
+	cf := m.loadConsumedFailed()
+	if mk := m.loadMarker(); mk != nil && mk.Hash == h && (cf == nil || cf.Hash != h) {
+		m.setStatus(&hub.PBSDRStatus{State: mk.State, StorageID: block.StorageID, Namespace: block.Namespace, AppliedAt: mk.AppliedAt})
+		return // idempotent: this exact descriptor already converged
+	}
+
+	entry, found, err := m.px.StorageEntry(ctx, block.StorageID)
+	if err != nil {
+		m.logger.Warn("pbsdr: storage-entry read failed (transient; retrying next tick)", "err", err)
+		return
+	}
+	if found && entry.Type != "pbs" {
+		msg := fmt.Sprintf("storage id %s exists with type %q (not pbs) — refusing to touch it", block.StorageID, entry.Type)
+		m.logger.Error("pbsdr: " + msg)
+		m.setStatus(&hub.PBSDRStatus{State: "verify_failed", StorageID: block.StorageID, Message: msg})
+		return
+	}
+
+	if found {
+		active, err := m.px.StorageActive(ctx, block.StorageID)
+		if err != nil {
+			m.logger.Warn("pbsdr: storage status probe failed (transient)", "err", err)
+			return
+		}
+		if active {
+			m.adopt(ctx, block, entry, h)
+			return
+		}
+		// Exists but unhealthy → the recovery path: verify → consume → reconcile-with-password.
+	}
+
+	// VERIFY-PIN-BEFORE-CONSUME (the ordering law): a mismatched or unreachable PBS aborts here —
+	// the one-time secret is untouched and the bridge simply retries next tick.
+	if err := m.probeFP(ctx, block.PBSTunnelIP, block.Fingerprint); err != nil {
+		m.logger.Warn("pbsdr: PBS fingerprint verify failed BEFORE consume (nothing consumed; retrying)", "err", err)
+		m.setStatus(&hub.PBSDRStatus{State: "verify_failed", StorageID: block.StorageID, Namespace: block.Namespace,
+			Message: "pre-consume fingerprint verify: " + err.Error()})
+		return
+	}
+
+	secret, err := m.hub.ConsumePBSToken(ctx)
+	if errors.Is(err, hub.ErrNoPBSSecret) {
+		if cf != nil && cf.Hash == h {
+			// The burned-secret dead-end: stay LOUD until the operator re-issues (fresh secret).
+			m.setStatus(&hub.PBSDRStatus{State: "consumed_failed", StorageID: block.StorageID, Namespace: block.Namespace,
+				ConsumedFailed: true, Message: cf.Message + " — awaiting operator re-issue (hub: Re-issue PBS credentials)"})
+			return
+		}
+		m.setStatus(&hub.PBSDRStatus{State: "waiting_secret", StorageID: block.StorageID, Namespace: block.Namespace,
+			Message: "verified; no unconsumed token secret staged on the hub"})
+		return
+	}
+	if err != nil {
+		m.logger.Warn("pbsdr: consume-token failed (transient; nothing consumed hub-side on error)", "err", err)
+		return
+	}
+	m.logger.Info("pbsdr: one-time token secret consumed (single-use; value withheld from logs)",
+		"storage_id", block.StorageID, "secret_len", len(secret))
+
+	// From here the secret is BURNED — every failure below is the loud persistent state.
+	if !found {
+		_, errOut, err := m.runner.RunStdin(ctx, strings.NewReader(secret+"\n"), WrapperPath,
+			"create", block.StorageID, block.PBSTunnelIP, block.Datastore, block.Namespace,
+			block.TokenID, block.Fingerprint, m.secretDir)
+		if err != nil {
+			m.consumedFail(h, fmt.Sprintf("wrapper create failed: %v (stderr: %s)", err, tail(errOut)), block)
+			return
+		}
+	} else {
+		_, errOut, err := m.runner.RunStdin(ctx, strings.NewReader(secret+"\n"), WrapperPath,
+			"reconcile", block.StorageID, block.PBSTunnelIP, block.Namespace,
+			block.TokenID, block.Fingerprint, m.secretDir)
+		if err != nil {
+			m.consumedFail(h, fmt.Sprintf("wrapper reconcile failed: %v (stderr: %s)", err, tail(errOut)), block)
+			return
+		}
+	}
+	if _, errOut, err := m.runner.Run(ctx, WrapperPath, "grant", block.StorageID); err != nil {
+		m.consumedFail(h, fmt.Sprintf("wrapper grant failed: %v (stderr: %s)", err, tail(errOut)), block)
+		return
+	}
+	active, err := m.px.StorageActive(ctx, block.StorageID)
+	if err != nil || !active {
+		m.consumedFail(h, fmt.Sprintf("post-apply status probe failed (active=%v err=%v)", active, err), block)
+		return
+	}
+
+	m.finishConverged(block, h, "applied")
+}
+
+// adopt is the non-destructive existing-entry path: NO consume, tenancy identity entry-owned.
+func (m *Manager) adopt(ctx context.Context, block *hub.WirePBSDR, entry *proxmox.StorageEntryConfig, h string) {
+	if _, errOut, err := m.runner.Run(ctx, WrapperPath, "grant", block.StorageID); err != nil {
+		// No secret involved — a grant failure is transient, retried next tick.
+		m.logger.Warn("pbsdr: adoption grant failed (retrying next tick)", "err", err, "stderr", tail(errOut))
+		return
+	}
+	note := ""
+	if entry.Namespace != block.Namespace {
+		note = fmt.Sprintf("adopted entry keeps its own tenancy (namespace %q; descriptor says %q — entry wins, never repointed)",
+			entry.Namespace, block.Namespace)
+		m.logger.Info("pbsdr: " + note)
+	}
+	m.finishConverged(block, h, "adopted")
+	if note != "" {
+		st := m.Status()
+		st.Message = note
+		st.Namespace = entry.Namespace // report the REAL tenancy
+		m.setStatus(st)
+	}
+}
+
+// finishConverged writes the marker, clears any consumed-failed state, seeds the escrow storage
+// id, and publishes the converged status.
+func (m *Manager) finishConverged(block *hub.WirePBSDR, h, state string) {
+	at := m.now().Format(time.RFC3339)
+	if err := m.writeState(m.markerPath(), marker{Hash: h, State: state, AppliedAt: at}); err != nil {
+		m.logger.Error("pbsdr: marker write failed (converged, but will re-run next tick)", "err", err)
+	}
+	os.Remove(m.consumedFailedPath())
+	msg := ""
+	if err := m.seedEscrowStorageID(block.StorageID); err != nil {
+		msg = "escrow.pbs_storage_id seed failed: " + err.Error() + " (set it manually before the ceremony)"
+		m.logger.Warn("pbsdr: " + msg)
+	}
+	m.logger.Info("pbsdr: converged", "state", state, "storage_id", block.StorageID)
+	m.setStatus(&hub.PBSDRStatus{State: state, StorageID: block.StorageID, Namespace: block.Namespace,
+		AppliedAt: at, Message: msg})
+}
+
+// consumedFail records the LOUD persistent burned-secret state.
+func (m *Manager) consumedFail(h, msg string, block *hub.WirePBSDR) {
+	m.logger.Error("pbsdr: CONSUMED-BUT-FAILED — the one-time secret is burned; NOT retrying silently. "+
+		"Operator action: Re-issue PBS credentials on the hub.", "detail", msg, "storage_id", block.StorageID)
+	if err := m.writeState(m.consumedFailedPath(), consumedFailed{Hash: h, Message: msg, At: m.now().Format(time.RFC3339)}); err != nil {
+		m.logger.Error("pbsdr: consumed-failed state write failed", "err", err)
+	}
+	m.setStatus(&hub.PBSDRStatus{State: "consumed_failed", StorageID: block.StorageID, Namespace: block.Namespace,
+		ConsumedFailed: true, Message: msg})
+}
+
+// seedEscrowStorageID sets escrow.pbs_storage_id in agent.json when empty/absent — the bare
+// `--selftest=escrow-create` one-liner must find its storage with no flags. A different existing
+// value is NEVER clobbered (warn-and-keep). Unknown config keys are preserved verbatim
+// (map[string]RawMessage read-modify-write, atomic rename).
+func (m *Manager) seedEscrowStorageID(storageID string) error {
+	if m.configPath == "" {
+		return nil
+	}
+	raw, err := os.ReadFile(m.configPath)
+	if err != nil {
+		return err
+	}
+	var doc map[string]json.RawMessage
+	if err := json.Unmarshal(raw, &doc); err != nil {
+		return fmt.Errorf("parse %s: %w", m.configPath, err)
+	}
+	var esc map[string]json.RawMessage
+	if cur, ok := doc["escrow"]; ok {
+		if err := json.Unmarshal(cur, &esc); err != nil {
+			return fmt.Errorf("parse escrow section: %w", err)
+		}
+	} else {
+		esc = map[string]json.RawMessage{}
+	}
+	if cur, ok := esc["pbs_storage_id"]; ok {
+		var existing string
+		_ = json.Unmarshal(cur, &existing)
+		if existing == storageID {
+			return nil // already seeded
+		}
+		if existing != "" {
+			m.logger.Warn("pbsdr: escrow.pbs_storage_id already set differently — keeping it",
+				"existing", existing, "descriptor", storageID)
+			return nil
+		}
+	}
+	idJSON, _ := json.Marshal(storageID)
+	esc["pbs_storage_id"] = idJSON
+	escJSON, err := json.Marshal(esc)
+	if err != nil {
+		return err
+	}
+	doc["escrow"] = escJSON
+	out, err := json.MarshalIndent(doc, "", "  ")
+	if err != nil {
+		return err
+	}
+	st, err := os.Stat(m.configPath)
+	if err != nil {
+		return err
+	}
+	tmp := m.configPath + ".pbsdr-tmp"
+	if err := os.WriteFile(tmp, out, st.Mode().Perm()); err != nil {
+		return err
+	}
+	if err := os.Rename(tmp, m.configPath); err != nil {
+		os.Remove(tmp)
+		return err
+	}
+	m.logger.Info("pbsdr: seeded escrow.pbs_storage_id (the ceremony one-liner needs no flags)", "storage_id", storageID)
+	return nil
+}
+
+func tail(b []byte) string {
+	s := strings.TrimSpace(string(b))
+	if len(s) > 300 {
+		s = s[len(s)-300:]
+	}
+	return s
+}
@@ -0,0 +1,402 @@
+package pbsdr
+
+// The apply-bridge laws, each pinned by a non-hollow test (fake exec recorder — no docker/pct/
+// real /dev, the REUSE §4 doctrine): set-only re-apply, secret-on-stdin-never-argv,
+// verify-pin-BEFORE-consume, non-destructive adoption (no consume, tenancy entry-owned),
+// consumed-but-failed loud + recover-only-via-fresh-secret, marker idempotency, old-hub compat.
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"io"
+	"log/slog"
+	"os"
+	"path/filepath"
+	"regexp"
+	"strings"
+	"sync"
+	"testing"
+
+	"gitea.dooplex.hu/admin/felhom-agent/internal/hub"
+	"gitea.dooplex.hu/admin/felhom-agent/internal/proxmox"
+)
+
+// recordedCall is one exec through the runner seam — argv AND the stdin bytes.
+type recordedCall struct {
+	Name  string
+	Args  []string
+	Stdin string
+}
+
+type fakeRunner struct {
+	mu    sync.Mutex
+	calls []recordedCall
+	// failVerb → error for calls whose first arg matches (e.g. "create").
+	failVerb string
+	failErr  error
+}
+
+func (f *fakeRunner) Run(ctx context.Context, name string, args ...string) ([]byte, []byte, error) {
+	return f.RunStdin(ctx, nil, name, args...)
+}
+
+func (f *fakeRunner) RunStdin(_ context.Context, stdin io.Reader, name string, args ...string) ([]byte, []byte, error) {
+	var in []byte
+	if stdin != nil {
+		in, _ = io.ReadAll(stdin)
+	}
+	f.mu.Lock()
+	f.calls = append(f.calls, recordedCall{Name: name, Args: args, Stdin: string(in)})
+	f.mu.Unlock()
+	if f.failVerb != "" && len(args) > 0 && args[0] == f.failVerb {
+		return nil, []byte("boom-stderr"), f.failErr
+	}
+	return nil, nil, nil
+}
+
+func (f *fakeRunner) recorded() []recordedCall {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	return append([]recordedCall(nil), f.calls...)
+}
+
+type fakeStorage struct {
+	entry  *proxmox.StorageEntryConfig
+	found  bool
+	active []bool // consumed per StorageActive call; last value repeats
+	calls  int
+}
+
+func (f *fakeStorage) StorageEntry(context.Context, string) (*proxmox.StorageEntryConfig, bool, error) {
+	return f.entry, f.found, nil
+}
+
+func (f *fakeStorage) StorageActive(context.Context, string) (bool, error) {
+	i := f.calls
+	f.calls++
+	if i >= len(f.active) {
+		if len(f.active) == 0 {
+			return false, nil
+		}
+		return f.active[len(f.active)-1], nil
+	}
+	return f.active[i], nil
+}
+
+type fakeConsumer struct {
+	secret string
+	err    error
+	calls  int
+}
+
+func (f *fakeConsumer) ConsumePBSToken(context.Context) (string, error) {
+	f.calls++
+	if f.err != nil {
+		return "", f.err
+	}
+	return f.secret, nil
+}
+
+const testFP = "c6:07:28:3f:5b:7b:5a:41:90:28:d7:ca:4f:37:14:70:56:39:2e:2f:0b:71:e8:06:ca:60:4a:d5:56:5f:3c:fd"
+
+func testBlock() *hub.WirePBSDR {
+	return &hub.WirePBSDR{
+		Enabled: true, StorageID: "felhom-pbs", PBSTunnelIP: "10.77.0.1",
+		Datastore: "felhom-offsite", Namespace: "peti", TokenID: "felhom@pbs!peti",
+		Fingerprint: testFP,
+	}
+}
+
+// newTestManager: fakes everywhere; the fingerprint probe defaults to PASS (override per test);
+// a real temp agent.json so the escrow seed is asserted end-to-end.
+func newTestManager(t *testing.T, r *fakeRunner, st *fakeStorage, c *fakeConsumer) (*Manager, string) {
+	t.Helper()
+	dir := t.TempDir()
+	cfgPath := filepath.Join(dir, "agent.json")
+	if err := os.WriteFile(cfgPath, []byte(`{"log_level":"info","escrow":{"posture":"zero_knowledge"},"custom_unknown":{"keep":1}}`), 0o600); err != nil {
+		t.Fatalf("seed config: %v", err)
+	}
+	m := NewManager(r, st, c, dir, "/etc/pve/priv/storage", cfgPath,
+		slog.New(slog.NewTextHandler(io.Discard, nil)))
+	m.probeFP = func(context.Context, string, string) error { return nil }
+	return m, cfgPath
+}
+
+func TestFreshPath_SecretOnStdinNeverArgv(t *testing.T) {
+	r := &fakeRunner{}
+	st := &fakeStorage{found: false, active: []bool{true}} // post-apply probe active
+	c := &fakeConsumer{secret: "SUPER-SECRET"}
+	m, cfgPath := newTestManager(t, r, st, c)
+
+	m.Apply(context.Background(), true, testBlock())
+
+	calls := r.recorded()
+	if len(calls) != 2 || calls[0].Args[0] != "create" || calls[1].Args[0] != "grant" {
+		t.Fatalf("calls = %+v, want [create, grant]", calls)
+	}
+	// THE STDIN LAW: the secret appears in NO argv, ONLY on the create call's stdin.
+	// (Red-proof: pass it as an argument → this fails with the secret visible in Args.)
+	for _, call := range calls {
+		for _, a := range call.Args {
+			if strings.Contains(a, "SUPER-SECRET") {
+				t.Fatalf("secret leaked into argv: %v", call.Args)
+			}
+		}
+	}
+	if calls[0].Stdin != "SUPER-SECRET\n" {
+		t.Errorf("create stdin = %q, want the secret + newline", calls[0].Stdin)
+	}
+	if calls[1].Stdin != "" {
+		t.Errorf("grant received stdin %q", calls[1].Stdin)
+	}
+	// Non-secret coords ride argv, descriptor-exact.
+	want := []string{"create", "felhom-pbs", "10.77.0.1", "felhom-offsite", "peti", "felhom@pbs!peti", testFP, "/etc/pve/priv/storage"}
+	if fmt.Sprint(calls[0].Args) != fmt.Sprint(want) {
+		t.Errorf("create argv = %v, want %v", calls[0].Args, want)
+	}
+	if c.calls != 1 {
+		t.Errorf("consume calls = %d, want 1", c.calls)
+	}
+	// Converged: marker applied + escrow seeded + unknown config keys preserved.
+	if s := m.Status(); s == nil || s.State != "applied" {
+		t.Fatalf("status = %+v, want applied", s)
+	}
+	raw, _ := os.ReadFile(cfgPath)
+	var doc map[string]json.RawMessage
+	json.Unmarshal(raw, &doc)
+	var esc map[string]string
+	json.Unmarshal(doc["escrow"], &esc)
+	if esc["pbs_storage_id"] != "felhom-pbs" {
+		t.Errorf("escrow not seeded: %s", doc["escrow"])
+	}
+	if esc["posture"] != "zero_knowledge" {
+		t.Errorf("existing escrow fields clobbered: %s", doc["escrow"])
+	}
+	if _, ok := doc["custom_unknown"]; !ok {
+		t.Error("unknown config key dropped by the seed write")
+	}
+	if strings.Contains(string(raw), "SUPER-SECRET") {
+		t.Error("secret leaked into agent.json")
+	}
+}
+
+func TestVerifyPinBeforeConsume(t *testing.T) {
+	r := &fakeRunner{}
+	st := &fakeStorage{found: false}
+	c := &fakeConsumer{secret: "S"}
+	m, _ := newTestManager(t, r, st, c)
+	m.probeFP = func(context.Context, string, string) error { return errors.New("pin mismatch") }
+
+	m.Apply(context.Background(), true, testBlock())
+
+	// THE ORDERING LAW: a failing pre-consume verify means NOTHING consumed, NOTHING executed.
+	// (Red-proof: reorder consume before the probe → calls=1 → FAIL.)
+	if c.calls != 0 {
+		t.Fatalf("consume calls = %d, want 0 — the secret was touched before the fingerprint verify", c.calls)
+	}
+	if len(r.recorded()) != 0 {
+		t.Fatalf("runner calls = %+v, want none", r.recorded())
+	}
+	if s := m.Status(); s == nil || s.State != "verify_failed" {
+		t.Fatalf("status = %+v, want verify_failed", s)
+	}
+}
+
+func TestAdoption_HealthyEntryNoConsumeTenancyEntryOwned(t *testing.T) {
+	r := &fakeRunner{}
+	st := &fakeStorage{
+		found: true,
+		entry: &proxmox.StorageEntryConfig{Storage: "felhom-offsite", Type: "pbs",
+			Server: "10.77.0.1", Datastore: "felhom-offsite", Namespace: "demo-felhom-01",
+			Username: "felhom@pbs!demo-felhom-01", Fingerprint: testFP},
+		active: []bool{true},
+	}
+	c := &fakeConsumer{secret: "STAGED-BUT-MUST-STAY"}
+	m, _ := newTestManager(t, r, st, c)
+
+	block := testBlock()
+	block.StorageID = "felhom-offsite"
+	block.Namespace = "demo" // the hub-provisioned tenancy differs — the entry must WIN
+	m.Apply(context.Background(), true, block)
+
+	if c.calls != 0 {
+		t.Fatalf("adoption consumed the staged secret (%d calls) — the no-consume law", c.calls)
+	}
+	calls := r.recorded()
+	if len(calls) != 1 || calls[0].Args[0] != "grant" {
+		t.Fatalf("adoption calls = %+v, want exactly [grant]", calls)
+	}
+	s := m.Status()
+	if s == nil || s.State != "adopted" {
+		t.Fatalf("status = %+v, want adopted", s)
+	}
+	if s.Namespace != "demo-felhom-01" {
+		t.Errorf("reported namespace = %q, want the ENTRY's (demo-felhom-01) — tenancy is entry-owned", s.Namespace)
+	}
+	if !strings.Contains(s.Message, "entry wins") {
+		t.Errorf("adoption note missing from message: %q", s.Message)
+	}
+}
+
+// TestSetOnlyLaw pins the data-loss-class guard twice over: (1) the re-apply path over an
+// existing entry uses `reconcile` (pvesm set) — never create, never any deletion verb; (2) the
+// wrapper script itself contains no deletion path (grep gate over the shipped file).
+// Red-proof: introduce a remove+re-add path → the recorded verbs change → FAIL.
+func TestSetOnlyLaw(t *testing.T) {
+	r := &fakeRunner{}
+	st := &fakeStorage{
+		found: true,
+		entry: &proxmox.StorageEntryConfig{Storage: "felhom-pbs", Type: "pbs", Namespace: "peti", Fingerprint: testFP},
+		// unhealthy → recovery path (verify → consume → reconcile) → post-apply healthy
+		active: []bool{false, true},
+	}
+	c := &fakeConsumer{secret: "FRESH"}
+	m, _ := newTestManager(t, r, st, c)
+
+	m.Apply(context.Background(), true, testBlock())
+
+	calls := r.recorded()
+	if len(calls) != 2 || calls[0].Args[0] != "reconcile" || calls[1].Args[0] != "grant" {
+		t.Fatalf("re-apply calls = %+v, want [reconcile, grant] (set-only)", calls)
+	}
+	deletionish := regexp.MustCompile(`remove|delete|destroy`)
+	for _, call := range calls {
+		for _, a := range call.Args {
+			if deletionish.MatchString(a) {
+				t.Fatalf("deletion-class verb reached the wrapper: %v — K destruction path", call.Args)
+			}
+		}
+	}
+	if calls[0].Stdin != "FRESH\n" {
+		t.Errorf("reconcile stdin = %q, want the fresh secret (re-issue recovery)", calls[0].Stdin)
+	}
+
+	// (2) the wrapper file: no deletion path, grep-assertable (the spike's set-only law).
+	wrapper, err := os.ReadFile(filepath.Join("..", "..", "configs", "felhom-pbs-apply"))
+	if err != nil {
+		t.Fatalf("read wrapper: %v", err)
+	}
+	if regexp.MustCompile(`pvesm (remove|delete)`).Match(wrapper) {
+		t.Fatal("configs/felhom-pbs-apply contains a pvesm deletion verb — the set-only law is dead")
+	}
+	if regexp.MustCompile(`rm\s+.*\.enc`).Match(wrapper) {
+		t.Fatal("configs/felhom-pbs-apply deletes a .enc file — K destruction path")
+	}
+}
+
+func TestConsumedButFailed_LoudAndRecoversOnlyViaFreshSecret(t *testing.T) {
+	r := &fakeRunner{failVerb: "create", failErr: errors.New("pvesm add exploded")}
+	st := &fakeStorage{found: false}
+	c := &fakeConsumer{secret: "BURNED"}
+	m, _ := newTestManager(t, r, st, c)
+	block := testBlock()
+
+	// 1. consume + create fails → LOUD persistent state.
+	m.Apply(context.Background(), true, block)
+	s := m.Status()
+	if s == nil || s.State != "consumed_failed" || !s.ConsumedFailed {
+		t.Fatalf("status = %+v, want consumed_failed", s)
+	}
+	if _, err := os.Stat(m.consumedFailedPath()); err != nil {
+		t.Fatal("consumed-failed state file not written")
+	}
+
+	// 2. next ticks WITHOUT a fresh secret: no silent recovery, state stays loud.
+	c.err = hub.ErrNoPBSSecret
+	m.Apply(context.Background(), true, block)
+	if s := m.Status(); s == nil || s.State != "consumed_failed" {
+		t.Fatalf("status after no-secret retry = %+v, want consumed_failed (never quiet waiting)", s)
+	}
+
+	// 3. operator re-issue staged a FRESH secret → the bridge recovers on the next tick.
+	c.err = nil
+	c.secret = "FRESH-AFTER-REISSUE"
+	r.failVerb = ""
+	st.active = []bool{true}
+	m.Apply(context.Background(), true, block)
+	if s := m.Status(); s == nil || s.State != "applied" {
+		t.Fatalf("status after re-issue = %+v, want applied", s)
+	}
+	if _, err := os.Stat(m.consumedFailedPath()); !os.IsNotExist(err) {
+		t.Error("consumed-failed state not cleared after recovery")
+	}
+}
+
+func TestMarkerIdempotency(t *testing.T) {
+	r := &fakeRunner{}
+	st := &fakeStorage{found: false, active: []bool{true}}
+	c := &fakeConsumer{secret: "S"}
+	m, _ := newTestManager(t, r, st, c)
+	block := testBlock()
+
+	m.Apply(context.Background(), true, block)
+	n := len(r.recorded())
+	m.Apply(context.Background(), true, block) // same descriptor hash → pure no-op
+	if len(r.recorded()) != n || c.calls != 1 {
+		t.Fatalf("re-apply over an unchanged descriptor ran ops (calls %d→%d, consume %d)", n, len(r.recorded()), c.calls)
+	}
+	if s := m.Status(); s == nil || s.State != "applied" || s.AppliedAt == "" {
+		t.Fatalf("status = %+v, want applied with applied_at", s)
+	}
+}
+
+func TestOldHubAndDisabledCompat(t *testing.T) {
+	r := &fakeRunner{}
+	c := &fakeConsumer{secret: "S"}
+	m, _ := newTestManager(t, r, &fakeStorage{}, c)
+
+	m.Apply(context.Background(), false, nil) // no desired data yet
+	m.Apply(context.Background(), true, nil)  // pre-slice-1 hub: no pbs_dr key
+	if len(r.recorded()) != 0 || c.calls != 0 || m.Status() != nil {
+		t.Fatalf("nil-block Apply had effects (runner %d, consume %d, status %+v)", len(r.recorded()), c.calls, m.Status())
+	}
+
+	m.Apply(context.Background(), true, &hub.WirePBSDR{Enabled: false, StorageID: "felhom-pbs"})
+	if len(r.recorded()) != 0 || c.calls != 0 {
+		t.Fatal("disabled descriptor ran ops (teardown is not this slice)")
+	}
+	if s := m.Status(); s == nil || s.State != "disabled" {
+		t.Fatalf("status = %+v, want disabled", s)
+	}
+}
+
+// TestWireFieldNames pins the cross-repo descriptor contract (hub/internal/web/pbsdr.go
+// pbsDRDescriptor): the exact JSON the hub writes must land in WirePBSDR field-for-field.
+func TestWireFieldNames(t *testing.T) {
+	hubJSON := `{"desired_state":{"guests":[],"pbs_dr":{"enabled":true,"storage_id":"felhom-pbs",` +
+		`"pbs_tunnel_ip":"10.77.0.1","datastore":"felhom-offsite","namespace":"peti",` +
+		`"token_id":"felhom@pbs!peti","fingerprint":"aa:bb"}},"generation":7}`
+	var resp hub.DesiredStateResponse
+	if err := json.Unmarshal([]byte(hubJSON), &resp); err != nil {
+		t.Fatalf("unmarshal: %v", err)
+	}
+	b := resp.DesiredState.PBSDR
+	if b == nil || !b.Enabled || b.StorageID != "felhom-pbs" || b.PBSTunnelIP != "10.77.0.1" ||
+		b.Datastore != "felhom-offsite" || b.Namespace != "peti" ||
+		b.TokenID != "felhom@pbs!peti" || b.Fingerprint != "aa:bb" {
+		t.Fatalf("wire mapping wrong: %+v", b)
+	}
+}
+
+func TestEscrowSeed_NeverClobbersDifferentValue(t *testing.T) {
+	r := &fakeRunner{}
+	st := &fakeStorage{found: false, active: []bool{true}}
+	c := &fakeConsumer{secret: "S"}
+	m, cfgPath := newTestManager(t, r, st, c)
+	os.WriteFile(cfgPath, []byte(`{"escrow":{"pbs_storage_id":"operator-set-id"}}`), 0o600)
+
+	m.Apply(context.Background(), true, testBlock())
+
+	raw, _ := os.ReadFile(cfgPath)
+	var doc struct {
+		Escrow struct {
+			PBSStorageID string `json:"pbs_storage_id"`
+		} `json:"escrow"`
+	}
+	json.Unmarshal(raw, &doc)
+	if doc.Escrow.PBSStorageID != "operator-set-id" {
+		t.Fatalf("an operator-set escrow.pbs_storage_id was clobbered: %s", raw)
+	}
+}