fix(fleet): harden audit and refresh boundaries

README.md

@@ -225,10 +225,11 @@ kubeconfig, endpoint, or token through the Sith collector contract. Only normali
 freshness-bounded, and stored behind forced RLS.
 Failed refreshes retain the last snapshot as explicitly stale evidence and record only a closed
 failure category. Concurrent refresh requests are authorized independently and then coalesced only
-within the same validated workspace. The shared refresh runs on a detached, locally traced context,
-so one canceled caller cannot cancel its peers and no request credential, context value, or trace
-identity crosses caller boundaries; completed and panicking refresh flights are removed. Within one
-refresh, transport and validation use a bounded worker pool: `CollectorConfig` defaults to four
+within the same validated workspace. The shared refresh runs on a locally traced context detached
+from individual callers but bounded by the hub lifecycle, so one canceled caller cannot cancel its
+peers, shutdown cancels outstanding work, and no request credential, context value, or trace identity
+crosses caller boundaries; completed and panicking refresh flights are removed. Within one refresh,
+transport and validation use a bounded worker pool: `CollectorConfig` defaults to four
 concurrent spokes and accepts only explicit limits from 1 through 64. Persistence and coverage
 mutation remain serialized, so one store failure cancels the remaining workers before returning;
 per-spoke deadlines and sorted stale/unreachable coverage remain unchanged. The pinned

internal/hubdb/policy_audit_test.go

@@ -231,14 +231,21 @@ func FuzzPolicyAuditEntryHashUsesLengthFraming(f *testing.F) {
 	f.Add("user:alice", "phase-1-read")
 	f.Add("a", "bc")
 	f.Fuzz(func(t *testing.T, left, right string) {
-		if right == "" || len(left)+len(right) > 512 {
+		if len(right) < 2 || len(left)+len(right) > 256 {
 			t.Skip()
 		}
 		first := policyAuditTestEntry()
 		first.actor, first.reasonCode = left, right
 		second := policyAuditTestEntry()
-		second.actor, second.reasonCode = left+right, ""
-		if bytes.Equal(policyAuditEntryHash(first), policyAuditEntryHash(second)) {
+		second.actor, second.reasonCode = left+right[:1], right[1:]
+		if validatePolicyAuditEntry(first) != nil || validatePolicyAuditEntry(second) != nil {
+			t.Skip()
+		}
+		firstHash, secondHash := policyAuditEntryHash(first), policyAuditEntryHash(second)
+		if len(firstHash) != 32 || len(secondHash) != 32 {
+			t.Fatal("valid audit entries did not produce SHA-256 digests")
+		}
+		if bytes.Equal(firstHash, secondHash) {
 			t.Fatal("length-delimited audit fields produced an ambiguous digest")
 		}
 	})

internal/hubfleet/collector.go

@@ -121,6 +121,7 @@ type Store interface {
 
 // CollectorConfig defines bounded collection behavior.
 type CollectorConfig struct {
+	LifecycleContext    context.Context
 	Store               Store
 	Transport           Transport
 	PEP                 *pep.Enforcer
@@ -149,8 +150,8 @@ type Collector struct {
 
 // NewCollector constructs a fail-closed collector with bounded per-spoke work.
 func NewCollector(config CollectorConfig) (*Collector, error) {
-	if config.Store == nil || config.Transport == nil || config.PEP == nil {
-		return nil, fmt.Errorf("new spoke collector: store, transport, and policy enforcer are required")
+	if config.LifecycleContext == nil || config.Store == nil || config.Transport == nil || config.PEP == nil {
+		return nil, fmt.Errorf("new spoke collector: lifecycle context, store, transport, and policy enforcer are required")
 	}
 	if config.SpokeTimeout == 0 {
 		config.SpokeTimeout = defaultSpokeTimeout
@@ -183,7 +184,7 @@ func NewCollector(config CollectorConfig) (*Collector, error) {
 		store:               config.Store,
 		transport:           config.Transport,
 		pep:                 config.PEP,
-		refreshes:           newRefreshCoordinator(),
+		refreshes:           newRefreshCoordinator(config.LifecycleContext),
 		observer:            config.Observer,
 		tracer:              config.TraceObserver,
 		spokeTimeout:        config.SpokeTimeout,

internal/hubfleet/collector_concurrency_test.go

@@ -25,8 +25,9 @@ func TestCollectorBoundsParallelSpokesAndSortsCoverage(t *testing.T) {
 	var active atomic.Int64
 	var maximum atomic.Int64
 	collector, err := NewCollector(CollectorConfig{
-		Store: store,
-		PEP:   testReadPEP(t),
+		LifecycleContext: t.Context(),
+		Store:            store,
+		PEP:              testReadPEP(t),
 		Transport: transportFunc(func(_ context.Context, _ tenancy.WorkspaceID, spoke Spoke) (Snapshot, error) {
 			current := active.Add(1)
 			defer active.Add(-1)
@@ -84,8 +85,9 @@ func TestCollectorCollapsesSerialTimeoutsIntoParallelWave(t *testing.T) {
 
 	store := newMemoryStore(testSpokes("spoke-a", "spoke-b", "spoke-c", "spoke-d"))
 	collector, err := NewCollector(CollectorConfig{
-		Store: store,
-		PEP:   testReadPEP(t),
+		LifecycleContext: t.Context(),
+		Store:            store,
+		PEP:              testReadPEP(t),
 		Transport: transportFunc(func(ctx context.Context, _ tenancy.WorkspaceID, _ Spoke) (Snapshot, error) {
 			<-ctx.Done()
 			return Snapshot{}, ctx.Err()
@@ -120,8 +122,9 @@ func TestCollectorPersistsHealthyPeerWhileAnotherWorkerIsBlocked(t *testing.T) {
 	releaseBlocked := make(chan struct{})
 	blockedStarted := make(chan struct{})
 	collector, err := NewCollector(CollectorConfig{
-		Store: store,
-		PEP:   testReadPEP(t),
+		LifecycleContext: t.Context(),
+		Store:            store,
+		PEP:              testReadPEP(t),
 		Transport: transportFunc(func(_ context.Context, _ tenancy.WorkspaceID, spoke Spoke) (Snapshot, error) {
 			if spoke.ID == "spoke-blocked" {
 				close(blockedStarted)
@@ -164,9 +167,10 @@ func TestCollectWorkspaceCancellationStopsAdmissionAndWorkers(t *testing.T) {
 	var active atomic.Int64
 	observer := &recordingSnapshotObserver{}
 	collector, err := NewCollector(CollectorConfig{
-		Store:    store,
-		PEP:      testReadPEP(t),
-		Observer: observer,
+		LifecycleContext: t.Context(),
+		Store:            store,
+		PEP:              testReadPEP(t),
+		Observer:         observer,
 		Transport: transportFunc(func(ctx context.Context, _ tenancy.WorkspaceID, spoke Spoke) (Snapshot, error) {
 			active.Add(1)
 			defer active.Add(-1)
@@ -235,8 +239,9 @@ func TestCollectorStoreFailureCancelsWorkersAndLaterAdmission(t *testing.T) {
 	secondWorkerStarted := make(chan struct{})
 	blockedCanceled := make(chan struct{})
 	collector, err := NewCollector(CollectorConfig{
-		Store: store,
-		PEP:   testReadPEP(t),
+		LifecycleContext: t.Context(),
+		Store:            store,
+		PEP:              testReadPEP(t),
 		Transport: transportFunc(func(ctx context.Context, _ tenancy.WorkspaceID, spoke Spoke) (Snapshot, error) {
 			started <- spoke.ID
 			if spoke.ID == "spoke-a" {
@@ -290,8 +295,9 @@ func TestCollectorRecoversPanickingTransportWorkerAndLaterRefresh(t *testing.T)
 	var panicFlight atomic.Bool
 	panicFlight.Store(true)
 	collector, err := NewCollector(CollectorConfig{
-		Store: store,
-		PEP:   testReadPEP(t),
+		LifecycleContext: t.Context(),
+		Store:            store,
+		PEP:              testReadPEP(t),
 		Transport: transportFunc(func(ctx context.Context, _ tenancy.WorkspaceID, spoke Spoke) (Snapshot, error) {
 			if panicFlight.Load() {
 				switch spoke.ID {
@@ -338,8 +344,9 @@ func TestCollectorJoinsWorkersWhenStorePanicsAndLaterRefresh(t *testing.T) {
 	secondWorkerStarted := make(chan struct{})
 	blockedCanceled := make(chan struct{})
 	collector, err := NewCollector(CollectorConfig{
-		Store: store,
-		PEP:   testReadPEP(t),
+		LifecycleContext: t.Context(),
+		Store:            store,
+		PEP:              testReadPEP(t),
 		Transport: transportFunc(func(ctx context.Context, _ tenancy.WorkspaceID, spoke Spoke) (Snapshot, error) {
 			if store.panicNext.Load() {
 				switch spoke.ID {

internal/hubfleet/collector_test.go

@@ -29,8 +29,9 @@ func TestCollectorIsolatesSpokeFailuresAndRetainsStaleSnapshots(t *testing.T) {
 		failures:  make(map[string]FailureKind),
 	}
 	collector, err := NewCollector(CollectorConfig{
-		Store: store,
-		PEP:   testReadPEP(t),
+		LifecycleContext: t.Context(),
+		Store:            store,
+		PEP:              testReadPEP(t),
 		Transport: transportFunc(func(_ context.Context, workspaceID tenancy.WorkspaceID, spoke Spoke) (Snapshot, error) {
 			if workspaceID != "workspace-a" {
 				return Snapshot{}, errors.New("wrong workspace")
@@ -77,8 +78,9 @@ func TestCollectorBoundsEveryTransportCall(t *testing.T) {
 		failures:  make(map[string]FailureKind),
 	}
 	collector, err := NewCollector(CollectorConfig{
-		Store: store,
-		PEP:   testReadPEP(t),
+		LifecycleContext: t.Context(),
+		Store:            store,
+		PEP:              testReadPEP(t),
 		Transport: transportFunc(func(ctx context.Context, _ tenancy.WorkspaceID, _ Spoke) (Snapshot, error) {
 			deadline, exists := ctx.Deadline()
 			if !exists || time.Until(deadline) > 1100*time.Millisecond {
@@ -170,8 +172,9 @@ func TestCollectorCopiesTransportOwnedSnapshot(t *testing.T) {
 		failures:  make(map[string]FailureKind),
 	}
 	collector, err := NewCollector(CollectorConfig{
-		Store: store,
-		PEP:   testReadPEP(t),
+		LifecycleContext: t.Context(),
+		Store:            store,
+		PEP:              testReadPEP(t),
 		Transport: transportFunc(func(context.Context, tenancy.WorkspaceID, Spoke) (Snapshot, error) {
 			return transportSnapshot, nil
 		}),
@@ -195,20 +198,28 @@ func TestCollectorAndSourceRejectUnsafeConfiguration(t *testing.T) {
 	t.Parallel()
 
 	store := &memoryStore{snapshots: make(map[string]Snapshot), failures: make(map[string]FailureKind)}
-	collector, err := NewCollector(CollectorConfig{Store: store, Transport: transportFunc(func(context.Context, tenancy.WorkspaceID, Spoke) (Snapshot, error) {
+	transport := transportFunc(func(context.Context, tenancy.WorkspaceID, Spoke) (Snapshot, error) {
 		return Snapshot{}, nil
-	}), PEP: testReadPEP(t)})
+	})
+	collector, err := NewCollector(CollectorConfig{
+		LifecycleContext: t.Context(), Store: store, Transport: transport, PEP: testReadPEP(t),
+	})
 	if err != nil || collector.maxConcurrentSpokes != defaultSpokeConcurrency {
 		t.Fatalf("default worker configuration = %d/%v", collector.maxConcurrentSpokes, err)
 	}
-	if _, err := NewCollector(CollectorConfig{Store: store, Transport: transportFunc(func(context.Context, tenancy.WorkspaceID, Spoke) (Snapshot, error) {
-		return Snapshot{}, nil
-	}), PEP: testReadPEP(t), SpokeTimeout: 999 * time.Millisecond}); err == nil {
+	if _, err := NewCollector(CollectorConfig{Store: store, Transport: transport, PEP: testReadPEP(t)}); err == nil {
+		t.Fatal("collector without lifecycle context unexpectedly accepted")
+	}
+	if _, err := NewCollector(CollectorConfig{
+		LifecycleContext: t.Context(), Store: store, Transport: transport,
+		PEP: testReadPEP(t), SpokeTimeout: 999 * time.Millisecond,
+	}); err == nil {
 		t.Fatal("sub-second collector timeout unexpectedly accepted")
 	}
-	if _, err := NewCollector(CollectorConfig{Store: store, Transport: transportFunc(func(context.Context, tenancy.WorkspaceID, Spoke) (Snapshot, error) {
-		return Snapshot{}, nil
-	}), PEP: testReadPEP(t), MaxConcurrentSpokes: maximumSpokeConcurrency + 1}); err == nil {
+	if _, err := NewCollector(CollectorConfig{
+		LifecycleContext: t.Context(), Store: store, Transport: transport,
+		PEP: testReadPEP(t), MaxConcurrentSpokes: maximumSpokeConcurrency + 1,
+	}); err == nil {
 		t.Fatal("oversized collector worker pool unexpectedly accepted")
 	}
 	if _, err := NewSource(SourceConfig{Reader: fleetReaderFunc(func(context.Context, tenancy.Scope, time.Duration, time.Time) (fleet.FleetResult, error) {

internal/hubfleet/metrics_test.go

@@ -47,7 +47,8 @@ func TestCollectorObservesClosedSnapshotOutcomes(t *testing.T) {
 				spokes: []Spoke{{ID: "spoke-a", ManagedClusterRef: "ocm/spoke-a"}}, snapshots: make(map[string]Snapshot), failures: make(map[string]FailureKind),
 			}
 			collector, err := NewCollector(CollectorConfig{
-				Store: store, Transport: test.transport, PEP: testReadPEP(t), Observer: observer, Now: func() time.Time { return now },
+				LifecycleContext: t.Context(), Store: store, Transport: test.transport,
+				PEP: testReadPEP(t), Observer: observer, Now: func() time.Time { return now },
 			})
 			if err != nil {
 				t.Fatal(err)
@@ -65,6 +66,7 @@ func TestCollectorObservesClosedSnapshotOutcomes(t *testing.T) {
 func TestCollectorRecoversFromPanickingSnapshotObserver(t *testing.T) {
 	now := time.Date(2026, time.July, 12, 20, 0, 0, 0, time.UTC)
 	collector, err := NewCollector(CollectorConfig{
+		LifecycleContext: t.Context(),
 		Store: &memoryStore{
 			spokes: []Spoke{{ID: "spoke-a", ManagedClusterRef: "ocm/spoke-a"}}, snapshots: make(map[string]Snapshot), failures: make(map[string]FailureKind),
 		},

internal/hubfleet/policy_test.go

@@ -20,7 +20,7 @@ func TestHubReadEntrypointsStopBeforeDependenciesWhenPolicyRefuses(t *testing.T)
 
 	store := &memoryStore{snapshots: make(map[string]Snapshot), failures: make(map[string]FailureKind)}
 	collector, err := NewCollector(CollectorConfig{
-		Store: store, PEP: refusal.enforcer(t),
+		LifecycleContext: t.Context(), Store: store, PEP: refusal.enforcer(t),
 		Transport: transportFunc(func(context.Context, tenancy.WorkspaceID, Spoke) (Snapshot, error) {
 			return Snapshot{}, errors.New("transport must not run")
 		}),
@@ -118,7 +118,7 @@ func (refusal *policyRefusal) enforcer(t *testing.T) *pep.Enforcer {
 
 func TestHubReadConstructorsRequirePolicyEnforcer(t *testing.T) {
 	store := &memoryStore{snapshots: make(map[string]Snapshot), failures: make(map[string]FailureKind)}
-	if _, err := NewCollector(CollectorConfig{Store: store, Transport: transportFunc(func(context.Context, tenancy.WorkspaceID, Spoke) (Snapshot, error) {
+	if _, err := NewCollector(CollectorConfig{LifecycleContext: t.Context(), Store: store, Transport: transportFunc(func(context.Context, tenancy.WorkspaceID, Spoke) (Snapshot, error) {
 		return Snapshot{}, nil
 	})}); err == nil {
 		t.Fatal("NewCollector() accepted no policy enforcer")

internal/hubfleet/refresh_coordinator.go

@@ -7,6 +7,7 @@ import (
 	"errors"
 	"fmt"
 	"sync"
+	"time"
 
 	"github.com/ArdurAI/sith/internal/fleet"
 	"github.com/ArdurAI/sith/internal/tenancy"
@@ -22,14 +23,28 @@ type refreshFlight struct {
 }
 
 type refreshCoordinator struct {
-	mu      sync.Mutex
-	flights map[tenancy.WorkspaceID]*refreshFlight
+	lifecycle context.Context
+	mu        sync.Mutex
+	flights   map[tenancy.WorkspaceID]*refreshFlight
 }
 
-func newRefreshCoordinator() *refreshCoordinator {
-	return &refreshCoordinator{flights: make(map[tenancy.WorkspaceID]*refreshFlight)}
+func newRefreshCoordinator(lifecycle context.Context) *refreshCoordinator {
+	return &refreshCoordinator{
+		lifecycle: valueFreeContext{Context: lifecycle},
+		flights:   make(map[tenancy.WorkspaceID]*refreshFlight),
+	}
+}
+
+// valueFreeContext preserves only lifecycle cancellation and deadlines. Refresh work must not
+// inherit request-scoped credentials, authorization state, or trace identity from its owner.
+type valueFreeContext struct {
+	context.Context
 }
 
+func (ctx valueFreeContext) Value(any) any { return nil }
+
+func (ctx valueFreeContext) Deadline() (time.Time, bool) { return ctx.Context.Deadline() }
+
 func (coordinator *refreshCoordinator) collect(
 	ctx context.Context,
 	scope tenancy.Scope,
@@ -48,7 +63,7 @@ func (coordinator *refreshCoordinator) collect(
 	if !exists {
 		flight = &refreshFlight{done: make(chan struct{})}
 		coordinator.flights[workspaceID] = flight
-		//nolint:gosec // A flight must outlive one caller; run supplies a fresh value-free context.
+		//nolint:gosec // A flight outlives one caller but remains bounded by the collector lifecycle.
 		go coordinator.run(workspaceID, flight, scope, collect)
 	}
 	coordinator.mu.Unlock()
@@ -72,7 +87,7 @@ func (coordinator *refreshCoordinator) run(
 		coordinator.finish(workspaceID, flight, coverage, err)
 	}()
 
-	flightContext, _, traceErr := tracing.Ensure(context.Background())
+	flightContext, _, traceErr := tracing.Ensure(coordinator.lifecycle)
 	if traceErr != nil {
 		err = fmt.Errorf("collect spoke snapshots: establish refresh flight trace: %w", traceErr)
 		return