wgtunnel: S3 Part 2 — manager state machine + loop + desired raw-consumer seam

Manager: one-shot registration (marker gate; backoff cap 15m), adopt-lost-marker,
re-key-on-mismatch, REVOKED-STAYS-REVOKED teardown (marker kept, zero execs on
later ticks), no-teardown-on-absent-data, hash-gated apply (zero execs steady
state), restart-not-reload on conf change, self-heal enable. Status stanza with
latest-handshakes-ONLY wg read. Collector WireguardReporter seam. desired.Syncer
AddConsumer fan-out with panic containment. Red-proofs a/b/d run + reverted;
no-key-material-in-logs asserted.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01PSK5g6qYLknKj8u3QAFEr6

internal/desired/consumer_test.go

@@ -0,0 +1,90 @@
+package desired
+
+// S3 Group C — the raw-consumer fan-out seam: called on generation advance, NOT on no-advance,
+// and a panicking consumer is contained (the guest reconcile path must never break).
+
+import (
+	"context"
+	"io"
+	"log/slog"
+	"sync"
+	"testing"
+
+	"gitea.dooplex.hu/admin/felhom-agent/internal/hub"
+	"gitea.dooplex.hu/admin/felhom-agent/internal/reconcile"
+)
+
+type recordingConsumer struct {
+	mu    sync.Mutex
+	calls []*hub.DesiredStateResponse
+	panic bool
+}
+
+func (r *recordingConsumer) OnDesiredState(_ context.Context, resp *hub.DesiredStateResponse) {
+	r.mu.Lock()
+	r.calls = append(r.calls, resp)
+	r.mu.Unlock()
+	if r.panic {
+		panic("consumer exploded")
+	}
+}
+
+func (r *recordingConsumer) count() int { r.mu.Lock(); defer r.mu.Unlock(); return len(r.calls) }
+
+type stubFetcher struct{ resp *hub.DesiredStateResponse }
+
+func (s *stubFetcher) FetchDesiredState(context.Context) (*hub.DesiredStateResponse, error) {
+	return s.resp, nil
+}
+
+func testResp(gen int64) *hub.DesiredStateResponse {
+	return &hub.DesiredStateResponse{
+		Generation: gen,
+		DesiredState: hub.WireDesiredState{
+			Guests:    []hub.WireDesiredGuest{},
+			Wireguard: &hub.WireWireguard{Pubkey: "PK", AssignedIP: "10.77.0.2/32"},
+		},
+	}
+}
+
+func TestSyncer_ConsumerCalledOnAdvanceOnly(t *testing.T) {
+	provider := reconcile.NewCachingProvider()
+	f := &stubFetcher{resp: testResp(2)}
+	s := NewSyncer(f, provider, slog.New(slog.NewTextHandler(io.Discard, nil)))
+	c := &recordingConsumer{}
+	s.AddConsumer(c)
+
+	// Advance → fetch → consumer called with the raw doc (wireguard block intact).
+	s.OnEnvelope(context.Background(), &hub.ControlEnvelope{DesiredGeneration: 2})
+	if c.count() != 1 {
+		t.Fatalf("consumer calls = %d, want 1", c.count())
+	}
+	if c.calls[0].DesiredState.Wireguard == nil || c.calls[0].DesiredState.Wireguard.Pubkey != "PK" {
+		t.Fatalf("consumer got %+v — the raw wireguard block must ride through", c.calls[0].DesiredState.Wireguard)
+	}
+
+	// No advance → no fetch → no consumer call (the negative).
+	s.OnEnvelope(context.Background(), &hub.ControlEnvelope{DesiredGeneration: 2})
+	if c.count() != 1 {
+		t.Errorf("consumer called without a generation advance: %d", c.count())
+	}
+}
+
+func TestSyncer_PanickingConsumerContained(t *testing.T) {
+	provider := reconcile.NewCachingProvider()
+	f := &stubFetcher{resp: testResp(1)}
+	s := NewSyncer(f, provider, slog.New(slog.NewTextHandler(io.Discard, nil)))
+	bomb := &recordingConsumer{panic: true}
+	after := &recordingConsumer{}
+	s.AddConsumer(bomb)
+	s.AddConsumer(after)
+
+	// Must not panic out; the second consumer still runs; the provider still updated.
+	s.OnEnvelope(context.Background(), &hub.ControlEnvelope{DesiredGeneration: 1})
+	if after.count() != 1 {
+		t.Errorf("consumer after the panicking one not called: %d", after.count())
+	}
+	if provider.Generation() != 1 {
+		t.Errorf("provider generation = %d, want 1 (guest path unaffected)", provider.Generation())
+	}
+}

internal/desired/syncer.go

@@ -22,11 +22,28 @@ type Fetcher interface {
 	FetchDesiredState(ctx context.Context) (*hub.DesiredStateResponse, error)
 }
 
+// RawConsumer receives the FULL fetched desired-state document after each successful
+// generation-advance fetch (S3 seam — internal/wgtunnel consumes its wireguard block this way
+// without the reconcile engine learning about tunnels). Implementations must not block: do the
+// cheap store-and-nudge, never network/exec inline.
+type RawConsumer interface {
+	OnDesiredState(ctx context.Context, resp *hub.DesiredStateResponse)
+}
+
 // Syncer keeps the engine's CachingProvider in step with the hub's authoritative desired-state.
 type Syncer struct {
-	fetcher  Fetcher
-	provider *reconcile.CachingProvider
-	logger   *slog.Logger
+	fetcher   Fetcher
+	provider  *reconcile.CachingProvider
+	consumers []RawConsumer
+	logger    *slog.Logger
+}
+
+// AddConsumer registers a raw desired-state consumer (nil-safe no-op). Not concurrency-safe —
+// call during wiring, before the hub loop starts.
+func (s *Syncer) AddConsumer(c RawConsumer) {
+	if c != nil {
+		s.consumers = append(s.consumers, c)
+	}
 }
 
 // NewSyncer builds a Syncer over the hub fetcher and the engine's provider.
@@ -61,12 +78,27 @@ func (s *Syncer) OnEnvelope(ctx context.Context, env *hub.ControlEnvelope) {
 	s.provider.Update(resp.Generation, state)
 	s.logger.Info("desired: updated from hub",
 		"generation", resp.Generation, "guests", len(state.Guests))
+	// S3: fan the raw document out to registered consumers (wgtunnel etc). A panicking consumer
+	// is contained — the guest reconcile path must never break over a tunnel add-on.
+	for _, c := range s.consumers {
+		s.notifyConsumer(ctx, c, resp)
+	}
 	if env.HasSignedOps {
 		// 10A only notes the flag; fetching + verifying + executing signed ops is slice 10B.
 		s.logger.Info("desired: hub reports pending signed ops (fetch/execute is slice 10B)")
 	}
 }
 
+// notifyConsumer delivers one raw document with panic containment.
+func (s *Syncer) notifyConsumer(ctx context.Context, c RawConsumer, resp *hub.DesiredStateResponse) {
+	defer func() {
+		if r := recover(); r != nil {
+			s.logger.Error("desired: raw consumer panicked (contained)", "panic", r)
+		}
+	}()
+	c.OnDesiredState(ctx, resp)
+}
+
 // mapWire maps the hub wire desired-state to the reconcile domain. 10A acts only on guests; the
 // forward-compat fields (restore_directive — 10D — etc.) are carried on the wire and logged, but
 // not translated into actions here.