diff --git a/.coverage-baseline b/.coverage-baseline
index c7a703d9..e59d5c49 100644
--- a/.coverage-baseline
+++ b/.coverage-baseline
@@ -1 +1 @@
-78.0
+78.2
diff --git a/internal/queue/fact_stream.go b/internal/queue/fact_stream.go
new file mode 100644
index 00000000..e475cd1d
--- /dev/null
+++ b/internal/queue/fact_stream.go
@@ -0,0 +1,369 @@
+// SPDX-License-Identifier: Apache-2.0
+
+package queue
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"slices"
+	"strconv"
+	"strings"
+	"sync"
+	"time"
+
+	"k8s.io/apimachinery/pkg/runtime/schema"
+)
+
+// Defaults for the attribution fact transport. They are shared by both implementations so a
+// conformance test, and an operator reading one implementation's flags, sees one set of numbers.
+const (
+	// DefaultFactStreamMaxLen bounds one stream so a hot type cannot grow without bound between
+	// retention trims. It is a count of ENTRIES, and one entry carries a whole audit batch's facts
+	// for one (route, group/resource), so it is far more history than the TTL horizon usually holds.
+	DefaultFactStreamMaxLen int64 = 10000
+
+	// DefaultFactStreamReadCount bounds how many entries one Next drains per stream. It also decides
+	// when a follower is considered BEHIND, which is the precondition for trim-gap detection.
+	DefaultFactStreamReadCount int64 = 512
+
+	// DefaultFactStreamBlock is how long one Next waits for a new entry before returning empty. It
+	// sets how quickly a change to the followed set takes effect: a follower re-reads its set on
+	// every Next, so a subscribe or unsubscribe lands within one block period.
+	DefaultFactStreamBlock = time.Second
+
+	// DefaultFactStreamTrimInterval is how often a stream is trimmed to the retention horizon.
+	// Trimming is amortized onto the publish path, so this bounds the extra command rate rather
+	// than the accuracy of the horizon.
+	DefaultFactStreamTrimInterval = time.Minute
+)
+
+// factStreamKeySuffix namespaces the attribution fact STREAMS, one version on from the v1 fact
+// KEYS in attribution_index.go. The two are unrelated keyspaces that happen to carry the same
+// facts, so they are deliberately not siblings: an install that rolls back keeps reading v1 keys
+// while the v2 streams age out on their own.
+const factStreamKeySuffix = ":author:v2:audit:"
+
+// factStreamEntryField is the single stream-entry field carrying one JSON-encoded fact batch.
+// One field rather than one per fact keeps an entry's shape independent of how many facts an
+// audit batch produced.
+const factStreamEntryField = "facts"
+
+// streamIDSeparator splits a stream ID into its millisecond and sequence halves.
+const streamIDSeparator = "-"
+
+// FactStreamKey identifies one attribution fact stream: the audit route the facts arrived under,
+// and the group/resource they are about. The route is part of the identity for the same reason it
+// is part of the v1 fact keys — a fact from cluster A must never name the author of an object
+// watched on cluster B.
+//
+// GroupResource is the typed identity rather than a rendered string. The rendering matters: the
+// stream name embeds the API-path form groupResourceKey produces ("configmaps",
+// "apps/deployments"), and schema.GroupResource.String() produces the reversed dotted form
+// ("deployments.apps"), so a caller that rendered its own key would publish to a stream nobody
+// follows and get no compile error for it. Holding the type and rendering at the transport boundary
+// makes that mistake unrepresentable.
+type FactStreamKey struct {
+	AuditRoute    string
+	GroupResource schema.GroupResource
+}
+
+// FactStreamKeyFor builds the key for one audit route and group/resource.
+func FactStreamKeyFor(auditRoute string, gr schema.GroupResource) FactStreamKey {
+	return FactStreamKey{AuditRoute: auditRoute, GroupResource: gr}
+}
+
+// String renders the key for logs and metrics as "<route>/<group-resource>".
+func (k FactStreamKey) String() string {
+	return k.AuditRoute + "/" + k.groupResource()
+}
+
+// groupResource renders the type the way every key and stream name spells it.
+func (k FactStreamKey) groupResource() string {
+	return groupResourceKey(k.GroupResource.Group, k.GroupResource.Resource)
+}
+
+// FactEntry is one appended batch of facts, as it comes back to a follower. ID is the
+// transport-assigned position, "<unix-millis>-<sequence>", which increases strictly within a
+// stream and is what a follower resumes from.
+type FactEntry struct {
+	Key   FactStreamKey
+	ID    string
+	Facts []AuthorFact
+}
+
+// FactStreamGap reports that a follower was trimmed past on one stream: entries it had not read
+// were dropped by retention before it got to them, so the facts they carried are lost for good.
+// It is the one loss this transport can see, and reporting it is why the transport is a log with
+// positions rather than fire-and-forget publish/subscribe.
+type FactStreamGap struct {
+	Key FactStreamKey
+	// Cursor is the position the follower had reached; FirstSurviving is the oldest entry the
+	// stream still holds. FirstSurviving is newer than Cursor, and everything between them is gone.
+	Cursor         string
+	FirstSurviving string
+}
+
+// FactDelivery is one Next's result: the entries read this round, in append order per stream, plus
+// any trim gaps noticed. Both may be empty — that is an idle block period, not an error.
+type FactDelivery struct {
+	Entries []FactEntry
+	Gaps    []FactStreamGap
+}
+
+// FactPublisher appends a batch of facts for one (route, group/resource). The audit receiver is
+// its only caller: it decodes one EventList per request, groups the facts it accepted by stream,
+// and appends once per group.
+type FactPublisher interface {
+	// PublishFacts appends facts as ONE entry on the key's stream. An empty batch is a no-op.
+	// Entries appear to followers in the order they were published per stream; there is no
+	// ordering promise across streams, and none is needed — an object belongs to exactly one
+	// group/resource and therefore to exactly one stream.
+	PublishFacts(ctx context.Context, key FactStreamKey, facts []AuthorFact) error
+}
+
+// FactFollower follows a set of streams from a horizon. The fact index is its only caller: it
+// follows the union of the types any watch covers and applies what it reads into memory.
+type FactFollower interface {
+	// FollowFacts starts following keys, reading each from horizon before now. A horizon of the
+	// fact TTL is what makes a restart cost nothing: the follower replays the whole retention
+	// window before the first watch event needs it. Entries older than the horizon are skipped,
+	// to within the millisecond granularity of a stream position.
+	FollowFacts(keys []FactStreamKey, horizon time.Duration) FactSubscription
+}
+
+// FactSubscription is one follower's live position across its followed streams. It is not safe to
+// call Next concurrently with itself; SetStreams may be called from another goroutine at any time.
+// A subscription owns no resources beyond its cursors, so it is dropped rather than closed.
+type FactSubscription interface {
+	// SetStreams replaces the followed set. A newly followed stream starts from the horizon, so
+	// it replays its retention window; an unfollowed stream's cursor is forgotten, so following it
+	// again replays too. It takes effect on the next Next, hence within one block period.
+	SetStreams(keys []FactStreamKey)
+
+	// Next returns the entries appended since the last call, waiting up to the block period for
+	// the first of them. An empty delivery with a nil error means the block period elapsed with
+	// nothing new, which is the ordinary idle case. It returns an error only when the context ends
+	// or the transport fails.
+	Next(ctx context.Context) (FactDelivery, error)
+}
+
+// FactTransport is the whole seam: publish a batch, follow a set. Everything above it — the
+// in-memory index, the TTL sweep, the waiter registry, the resolver — has one implementation and
+// never learns which transport it has. Consumers should depend on FactPublisher or FactFollower,
+// the half they use; this composition exists for wiring and for the conformance suite.
+type FactTransport interface {
+	FactPublisher
+	FactFollower
+}
+
+// encodeFactBatch renders one batch as the entry payload. Both implementations store the encoded
+// form, so an entry is an immutable snapshot in both and a follower can never see a fact mutate
+// under it.
+func encodeFactBatch(facts []AuthorFact) ([]byte, error) {
+	raw, err := json.Marshal(facts)
+	if err != nil {
+		return nil, fmt.Errorf("marshal fact batch: %w", err)
+	}
+	return raw, nil
+}
+
+// decodeFactBatch parses an entry payload back into facts.
+func decodeFactBatch(raw []byte) ([]AuthorFact, error) {
+	var facts []AuthorFact
+	if err := json.Unmarshal(raw, &facts); err != nil {
+		return nil, fmt.Errorf("unmarshal fact batch: %w", err)
+	}
+	return facts, nil
+}
+
+// streamIDAt renders the position a stream reads from for entries appended at or after t. Stream
+// IDs are millisecond timestamps, so a time horizon is a plain position and needs no side index.
+// The position is EXCLUSIVE — a follower reads entries strictly after it — which is why an entry
+// appended in the horizon's own millisecond may be skipped. That granularity is deliberate and is
+// the one place the two implementations are allowed to differ by a hair.
+func streamIDAt(t time.Time) string {
+	millis := t.UnixMilli()
+	if millis < 0 {
+		millis = 0
+	}
+	return strconv.FormatInt(millis, 10) + streamIDSeparator + "0"
+}
+
+// compareStreamIDs orders two "<millis>-<seq>" positions, returning -1, 0 or 1. A malformed half
+// reads as zero: IDs are transport-assigned, so this is defensive only, and ordering a
+// unparseable ID first is the safe direction (it can only make a follower re-read).
+func compareStreamIDs(a, b string) int {
+	aMillis, aSeq := parseStreamID(a)
+	bMillis, bSeq := parseStreamID(b)
+	if aMillis != bMillis {
+		return compareUint64(aMillis, bMillis)
+	}
+	return compareUint64(aSeq, bSeq)
+}
+
+func compareUint64(a, b uint64) int {
+	switch {
+	case a < b:
+		return -1
+	case a > b:
+		return 1
+	default:
+		return 0
+	}
+}
+
+// parseStreamID splits a stream ID into its millisecond and sequence halves.
+func parseStreamID(id string) (uint64, uint64) {
+	millisPart, seqPart, _ := strings.Cut(id, streamIDSeparator)
+	millis, _ := strconv.ParseUint(millisPart, 10, 64)
+	seq, _ := strconv.ParseUint(seqPart, 10, 64)
+	return millis, seq
+}
+
+// followState is one followed stream's position. behind records that the last read filled its
+// entry budget, so more was waiting: it is the precondition for trim-gap detection, because a
+// follower that read everything there was cannot have been trimmed past. Without it, ordinary
+// retention — the cursor's own entry aging out while the follower idles — would report a gap on
+// every stream once per TTL period. reportedGap dedupes the report while the gap persists.
+type followState struct {
+	cursor      string
+	behind      bool
+	reportedGap string
+}
+
+// followTarget is one followed stream as a reader sees it for one pass.
+type followTarget struct {
+	Key    FactStreamKey
+	Cursor string
+	Behind bool
+}
+
+// followSet is the shared cursor bookkeeping behind both implementations' subscriptions: the
+// followed set, each stream's position, and the trim-gap precondition. Sharing it is what keeps
+// the two transports from drifting on the parts that are not transport-specific at all.
+type followSet struct {
+	horizon time.Duration
+
+	mu     sync.Mutex
+	states map[FactStreamKey]*followState
+}
+
+func newFollowSet(keys []FactStreamKey, horizon time.Duration) *followSet {
+	set := &followSet{horizon: horizon, states: map[FactStreamKey]*followState{}}
+	set.set(keys)
+	return set
+}
+
+// set replaces the followed set, starting anything newly followed at the horizon.
+func (f *followSet) set(keys []FactStreamKey) {
+	wanted := make(map[FactStreamKey]struct{}, len(keys))
+	for _, key := range keys {
+		wanted[key] = struct{}{}
+	}
+	from := streamIDAt(time.Now().Add(-f.horizon))
+
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	for key := range f.states {
+		if _, ok := wanted[key]; !ok {
+			delete(f.states, key)
+		}
+	}
+	for key := range wanted {
+		if _, ok := f.states[key]; !ok {
+			f.states[key] = &followState{cursor: from}
+		}
+	}
+}
+
+// targets snapshots the followed set for one read pass, in a stable order so a follower reading
+// several streams sees them the same way every time.
+func (f *followSet) targets() []followTarget {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	targets := make([]followTarget, 0, len(f.states))
+	for key, state := range f.states {
+		targets = append(targets, followTarget{Key: key, Cursor: state.cursor, Behind: state.behind})
+	}
+	slices.SortFunc(targets, func(a, b followTarget) int {
+		if c := strings.Compare(a.Key.AuditRoute, b.Key.AuditRoute); c != 0 {
+			return c
+		}
+		return strings.Compare(a.Key.groupResource(), b.Key.groupResource())
+	})
+	return targets
+}
+
+// advance moves one stream's cursor to the last entry delivered and records whether more was
+// waiting. A stream unfollowed during the read pass is not resurrected.
+func (f *followSet) advance(key FactStreamKey, cursor string, behind bool) {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	state, ok := f.states[key]
+	if !ok {
+		return
+	}
+	state.cursor = cursor
+	state.behind = behind
+	if state.reportedGap != "" && compareStreamIDs(cursor, state.reportedGap) >= 0 {
+		state.reportedGap = ""
+	}
+}
+
+// noteGap records a trim gap and reports whether it is new. The same gap is reported once, not on
+// every pass until the follower catches up past it.
+func (f *followSet) noteGap(key FactStreamKey, firstSurviving string) bool {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	state, ok := f.states[key]
+	if !ok || state.reportedGap == firstSurviving {
+		return false
+	}
+	state.reportedGap = firstSurviving
+	return true
+}
+
+// forgetGaps drops the dedupe marks for gaps that were detected but never delivered, so the next
+// read pass reports them again.
+func (f *followSet) forgetGaps(gaps []FactStreamGap) {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	for _, gap := range gaps {
+		if state, ok := f.states[gap.Key]; ok && state.reportedGap == gap.FirstSurviving {
+			state.reportedGap = ""
+		}
+	}
+}
+
+// gapIfTrimmedPast turns a stream's oldest surviving position into a gap report when the target
+// was behind and that position has moved past its cursor. It is the whole detection rule, shared
+// so both transports decide it identically.
+func (f *followSet) gapIfTrimmedPast(target followTarget, firstSurviving string) (FactStreamGap, bool) {
+	if !target.Behind || firstSurviving == "" {
+		return FactStreamGap{}, false
+	}
+	if compareStreamIDs(target.Cursor, firstSurviving) >= 0 {
+		return FactStreamGap{}, false
+	}
+	if !f.noteGap(target.Key, firstSurviving) {
+		return FactStreamGap{}, false
+	}
+	return FactStreamGap{Key: target.Key, Cursor: target.Cursor, FirstSurviving: firstSurviving}, true
+}
+
+// waitBlock sleeps out a block period, reporting the context ending as an error so a follower's
+// Next never reports an idle round it did not actually wait through.
+func waitBlock(ctx context.Context, block time.Duration) error {
+	if block <= 0 {
+		return ctx.Err()
+	}
+	timer := time.NewTimer(block)
+	defer timer.Stop()
+	select {
+	case <-ctx.Done():
+		return ctx.Err()
+	case <-timer.C:
+		return nil
+	}
+}
diff --git a/internal/queue/fact_stream_conformance_test.go b/internal/queue/fact_stream_conformance_test.go
new file mode 100644
index 00000000..cf580c88
--- /dev/null
+++ b/internal/queue/fact_stream_conformance_test.go
@@ -0,0 +1,467 @@
+// SPDX-License-Identifier: Apache-2.0
+
+package queue
+
+import (
+	"context"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/require"
+	"k8s.io/apimachinery/pkg/runtime/schema"
+)
+
+// The attribution fact transport has two implementations and ONE test suite. They are the same
+// data structure — a capped, ordered log with per-follower cursors — so anything true of one has
+// to be true of the other, and two transports tested apart would drift in whichever one the
+// maintainers do not run daily.
+//
+// Every test here runs against both. Nothing implementation-specific belongs in this file.
+
+var (
+	_ FactTransport = (*RedisFactStream)(nil)
+	_ FactTransport = (*MemoryFactStream)(nil)
+)
+
+const (
+	// factStreamTestBlock keeps an idle Next short so a test that waits for nothing is quick.
+	factStreamTestBlock = 50 * time.Millisecond
+	// factStreamTestTimeout bounds a drain that never delivers, so a broken implementation fails
+	// with the entries it did deliver rather than hanging.
+	factStreamTestTimeout = 10 * time.Second
+	// factStreamTestTTL is short enough to age an entry out inside a test, and long enough that a
+	// slow machine does not age one out mid-round.
+	factStreamTestTTL = 100 * time.Millisecond
+	// factStreamTestAgeWait is how long a test waits for factStreamTestTTL to have elapsed.
+	factStreamTestAgeWait = 300 * time.Millisecond
+	// factStreamTestHorizon replays everything the transport still holds.
+	factStreamTestHorizon = time.Hour
+)
+
+// factStreamParams are the knobs a conformance case needs to reach a behaviour deterministically:
+// a short TTL to observe retention, a small entry cap to observe eviction, and a small read count
+// to put a follower BEHIND, which is what makes a trim gap detectable.
+type factStreamParams struct {
+	TTL       time.Duration
+	MaxLen    int64
+	ReadCount int64
+}
+
+func defaultFactStreamParams() factStreamParams {
+	return factStreamParams{
+		TTL:       time.Minute,
+		MaxLen:    DefaultFactStreamMaxLen,
+		ReadCount: DefaultFactStreamReadCount,
+	}
+}
+
+// factStreamImplementation names one transport and builds it for one case.
+type factStreamImplementation struct {
+	name  string
+	build func(t *testing.T, params factStreamParams) FactTransport
+}
+
+func factStreamImplementations() []factStreamImplementation {
+	return []factStreamImplementation{
+		{
+			name: "redis",
+			build: func(t *testing.T, params factStreamParams) FactTransport {
+				t.Helper()
+				store, _ := newTestRedisStoreWithRedis(t)
+				return store.FactStream(RedisFactStreamConfig{
+					TTL:    params.TTL,
+					MaxLen: params.MaxLen,
+					// Trim on every publish: the interval only decides how often retention is
+					// enforced, and a test that waited a minute for it would test the clock.
+					TrimInterval: time.Nanosecond,
+					Block:        factStreamTestBlock,
+					ReadCount:    params.ReadCount,
+				})
+			},
+		},
+		{
+			name: "memory",
+			build: func(t *testing.T, params factStreamParams) FactTransport {
+				t.Helper()
+				return NewMemoryFactStream(MemoryFactStreamConfig{
+					TTL:       params.TTL,
+					MaxLen:    params.MaxLen,
+					Block:     factStreamTestBlock,
+					ReadCount: params.ReadCount,
+				})
+			},
+		},
+	}
+}
+
+// runFactStreamConformance runs one case against every implementation.
+func runFactStreamConformance(
+	t *testing.T,
+	params factStreamParams,
+	test func(t *testing.T, transport FactTransport),
+) {
+	t.Helper()
+	for _, impl := range factStreamImplementations() {
+		t.Run(impl.name, func(t *testing.T) {
+			test(t, impl.build(t, params))
+		})
+	}
+}
+
+func factStreamTestKey(group, resource string) FactStreamKey {
+	return FactStreamKeyFor("prod-eu-1", schema.GroupResource{Group: group, Resource: resource})
+}
+
+// authorFacts builds one batch naming the given authors.
+func authorFacts(authors ...string) []AuthorFact {
+	facts := make([]AuthorFact, 0, len(authors))
+	for _, author := range authors {
+		facts = append(facts, AuthorFact{Author: author, Verb: "update"})
+	}
+	return facts
+}
+
+// factAuthors flattens delivered entries back to the authors they carry, in delivery order.
+func factAuthors(entries []FactEntry) []string {
+	var authors []string
+	for _, entry := range entries {
+		for _, fact := range entry.Facts {
+			authors = append(authors, fact.Author)
+		}
+	}
+	return authors
+}
+
+// drainFactEntries reads until want entries have arrived, failing with what it did get.
+func drainFactEntries(t *testing.T, sub FactSubscription, want int) []FactEntry {
+	t.Helper()
+	ctx, cancel := context.WithTimeout(t.Context(), factStreamTestTimeout)
+	defer cancel()
+	entries, err := readFactEntries(ctx, sub, want)
+	require.NoErrorf(t, err, "wanted %d entries, delivered %d", want, len(entries))
+	return entries
+}
+
+// readFactEntries is the assertion-free half of drainFactEntries, so a follower can be drained
+// from a goroutine and judged on the test's own.
+func readFactEntries(ctx context.Context, sub FactSubscription, want int) ([]FactEntry, error) {
+	var entries []FactEntry
+	for len(entries) < want {
+		delivery, err := sub.Next(ctx)
+		if err != nil {
+			return entries, err
+		}
+		entries = append(entries, delivery.Entries...)
+	}
+	return entries, nil
+}
+
+// expectNoFactEntries asserts that nothing more arrives within one block period.
+func expectNoFactEntries(t *testing.T, sub FactSubscription) {
+	t.Helper()
+	ctx, cancel := context.WithTimeout(t.Context(), factStreamTestTimeout)
+	defer cancel()
+	delivery, err := sub.Next(ctx)
+	require.NoError(t, err)
+	require.Empty(t, factAuthors(delivery.Entries))
+}
+
+func TestFactStreamConformance_AppendsAreFollowedInOrder(t *testing.T) {
+	runFactStreamConformance(t, defaultFactStreamParams(), func(t *testing.T, transport FactTransport) {
+		ctx := t.Context()
+		key := factStreamTestKey("", "configmaps")
+		sub := transport.FollowFacts([]FactStreamKey{key}, factStreamTestHorizon)
+
+		require.NoError(t, transport.PublishFacts(ctx, key, authorFacts("alice")))
+		require.NoError(t, transport.PublishFacts(ctx, key, authorFacts("bob", "carol")))
+		require.NoError(t, transport.PublishFacts(ctx, key, authorFacts("dave")))
+
+		entries := drainFactEntries(t, sub, 3)
+		require.Len(t, entries, 3)
+		// One batch is one entry, and the batch keeps its own order inside it.
+		require.Equal(t, []string{"alice", "bob", "carol", "dave"}, factAuthors(entries))
+		for _, entry := range entries {
+			require.Equal(t, key, entry.Key)
+		}
+		require.Negative(t, compareStreamIDs(entries[0].ID, entries[1].ID))
+		require.Negative(t, compareStreamIDs(entries[1].ID, entries[2].ID))
+	})
+}
+
+func TestFactStreamConformance_EmptyBatchIsNotAppended(t *testing.T) {
+	runFactStreamConformance(t, defaultFactStreamParams(), func(t *testing.T, transport FactTransport) {
+		ctx := t.Context()
+		key := factStreamTestKey("", "configmaps")
+		sub := transport.FollowFacts([]FactStreamKey{key}, factStreamTestHorizon)
+
+		require.NoError(t, transport.PublishFacts(ctx, key, nil))
+		expectNoFactEntries(t, sub)
+	})
+}
+
+func TestFactStreamConformance_LateFollowerReplaysFromHorizon(t *testing.T) {
+	runFactStreamConformance(t, defaultFactStreamParams(), func(t *testing.T, transport FactTransport) {
+		ctx := t.Context()
+		key := factStreamTestKey("apps", "deployments")
+
+		// Everything a restart, a reconnect, or a brand new watch would have missed.
+		require.NoError(t, transport.PublishFacts(ctx, key, authorFacts("alice")))
+		require.NoError(t, transport.PublishFacts(ctx, key, authorFacts("bob")))
+
+		sub := transport.FollowFacts([]FactStreamKey{key}, factStreamTestHorizon)
+		require.Equal(t, []string{"alice", "bob"}, factAuthors(drainFactEntries(t, sub, 2)))
+
+		// And it keeps up from there rather than replaying the window again.
+		require.NoError(t, transport.PublishFacts(ctx, key, authorFacts("carol")))
+		require.Equal(t, []string{"carol"}, factAuthors(drainFactEntries(t, sub, 1)))
+	})
+}
+
+func TestFactStreamConformance_RetentionTrimDropsAgedEntries(t *testing.T) {
+	params := defaultFactStreamParams()
+	params.TTL = factStreamTestTTL
+	runFactStreamConformance(t, params, func(t *testing.T, transport FactTransport) {
+		ctx := t.Context()
+		key := factStreamTestKey("", "configmaps")
+
+		require.NoError(t, transport.PublishFacts(ctx, key, authorFacts("aged-out")))
+		time.Sleep(factStreamTestAgeWait)
+		require.NoError(t, transport.PublishFacts(ctx, key, authorFacts("retained")))
+
+		// A follower asking for far more history than the TTL still only gets what survives.
+		sub := transport.FollowFacts([]FactStreamKey{key}, factStreamTestHorizon)
+		require.Equal(t, []string{"retained"}, factAuthors(drainFactEntries(t, sub, 1)))
+		expectNoFactEntries(t, sub)
+	})
+}
+
+func TestFactStreamConformance_EntryCapEvictsOldestFirst(t *testing.T) {
+	params := defaultFactStreamParams()
+	params.MaxLen = 1
+	runFactStreamConformance(t, params, func(t *testing.T, transport FactTransport) {
+		ctx := t.Context()
+		key := factStreamTestKey("", "configmaps")
+
+		require.NoError(t, transport.PublishFacts(ctx, key, authorFacts("evicted")))
+		require.NoError(t, transport.PublishFacts(ctx, key, authorFacts("kept")))
+
+		sub := transport.FollowFacts([]FactStreamKey{key}, factStreamTestHorizon)
+		require.Equal(t, []string{"kept"}, factAuthors(drainFactEntries(t, sub, 1)))
+	})
+}
+
+func TestFactStreamConformance_ReportsTrimGapWhenFollowerIsTrimmedPast(t *testing.T) {
+	params := defaultFactStreamParams()
+	params.TTL = factStreamTestTTL
+	// One entry per read leaves the follower behind after the first, which is the precondition
+	// for a gap: a follower that read everything there was cannot have been trimmed past.
+	params.ReadCount = 1
+	runFactStreamConformance(t, params, func(t *testing.T, transport FactTransport) {
+		ctx := t.Context()
+		key := factStreamTestKey("", "configmaps")
+
+		require.NoError(t, transport.PublishFacts(ctx, key, authorFacts("first")))
+		require.NoError(t, transport.PublishFacts(ctx, key, authorFacts("missed")))
+
+		sub := transport.FollowFacts([]FactStreamKey{key}, factStreamTestHorizon)
+		read := drainFactEntries(t, sub, 1)
+		require.Equal(t, []string{"first"}, factAuthors(read))
+
+		// The follower stalls long enough for retention to drop what it had not read.
+		time.Sleep(factStreamTestAgeWait)
+		require.NoError(t, transport.PublishFacts(ctx, key, authorFacts("after-the-gap")))
+
+		gap, entries := drainUntilFactStreamGap(t, sub)
+		require.Equal(t, key, gap.Key)
+		require.Equal(t, read[0].ID, gap.Cursor)
+		require.Negative(t, compareStreamIDs(gap.Cursor, gap.FirstSurviving),
+			"the surviving entry must be newer than the cursor it skipped past")
+		require.Equal(t, []string{"after-the-gap"}, factAuthors(entries))
+	})
+}
+
+func TestFactStreamConformance_CaughtUpFollowerReportsNoTrimGap(t *testing.T) {
+	params := defaultFactStreamParams()
+	params.TTL = factStreamTestTTL
+	runFactStreamConformance(t, params, func(t *testing.T, transport FactTransport) {
+		ctx := t.Context()
+		key := factStreamTestKey("", "configmaps")
+		sub := transport.FollowFacts([]FactStreamKey{key}, factStreamTestHorizon)
+
+		require.NoError(t, transport.PublishFacts(ctx, key, authorFacts("read")))
+		require.Equal(t, []string{"read"}, factAuthors(drainFactEntries(t, sub, 1)))
+
+		// Ordinary retention: the entry this follower already read ages out while it idles, and
+		// the next entry arrives after it. Nothing was lost, so nothing may be reported.
+		time.Sleep(factStreamTestAgeWait)
+		require.NoError(t, transport.PublishFacts(ctx, key, authorFacts("next")))
+
+		readCtx, cancel := context.WithTimeout(t.Context(), factStreamTestTimeout)
+		defer cancel()
+		for range 3 {
+			delivery, err := sub.Next(readCtx)
+			require.NoError(t, err)
+			require.Empty(t, delivery.Gaps)
+		}
+	})
+}
+
+func TestFactStreamConformance_FollowedSetChangesAtRuntime(t *testing.T) {
+	runFactStreamConformance(t, defaultFactStreamParams(), func(t *testing.T, transport FactTransport) {
+		ctx := t.Context()
+		watched := factStreamTestKey("", "configmaps")
+		unwatched := factStreamTestKey("", "secrets")
+
+		require.NoError(t, transport.PublishFacts(ctx, watched, authorFacts("alice")))
+		require.NoError(t, transport.PublishFacts(ctx, unwatched, authorFacts("bob")))
+
+		// A type nobody watches is written and never read.
+		sub := transport.FollowFacts([]FactStreamKey{watched}, factStreamTestHorizon)
+		require.Equal(t, []string{"alice"}, factAuthors(drainFactEntries(t, sub, 1)))
+		expectNoFactEntries(t, sub)
+
+		// A new watch on that type replays its window rather than starting empty.
+		sub.SetStreams([]FactStreamKey{watched, unwatched})
+		require.Equal(t, []string{"bob"}, factAuthors(drainFactEntries(t, sub, 1)))
+
+		// And the last watch on a type going away stops its facts arriving.
+		sub.SetStreams([]FactStreamKey{unwatched})
+		require.NoError(t, transport.PublishFacts(ctx, watched, authorFacts("carol")))
+		require.NoError(t, transport.PublishFacts(ctx, unwatched, authorFacts("dave")))
+		require.Equal(t, []string{"dave"}, factAuthors(drainFactEntries(t, sub, 1)))
+		expectNoFactEntries(t, sub)
+	})
+}
+
+func TestFactStreamConformance_ConcurrentFollowersEachReceiveEveryEntry(t *testing.T) {
+	const followers, batches = 3, 5
+	runFactStreamConformance(t, defaultFactStreamParams(), func(t *testing.T, transport FactTransport) {
+		ctx := t.Context()
+		key := factStreamTestKey("", "configmaps")
+
+		// Fan-out, not work sharing: no consumer groups, so every follower needs every entry.
+		subs := make([]FactSubscription, 0, followers)
+		for range followers {
+			subs = append(subs, transport.FollowFacts([]FactStreamKey{key}, factStreamTestHorizon))
+		}
+
+		readCtx, cancel := context.WithTimeout(t.Context(), factStreamTestTimeout)
+		defer cancel()
+		var wg sync.WaitGroup
+		delivered := make([][]string, followers)
+		errs := make([]error, followers)
+		for i, sub := range subs {
+			wg.Add(1)
+			go func() {
+				defer wg.Done()
+				entries, err := readFactEntries(readCtx, sub, batches)
+				delivered[i], errs[i] = factAuthors(entries), err
+			}()
+		}
+
+		want := make([]string, 0, batches)
+		for i := range batches {
+			author := "author-" + string(rune('a'+i))
+			want = append(want, author)
+			require.NoError(t, transport.PublishFacts(ctx, key, authorFacts(author)))
+		}
+		wg.Wait()
+
+		for i := range followers {
+			require.NoErrorf(t, errs[i], "follower %d", i)
+			require.Equalf(t, want, delivered[i], "follower %d", i)
+		}
+	})
+}
+
+func TestFactStreamConformance_StreamsAreKeyedByRouteAndType(t *testing.T) {
+	runFactStreamConformance(t, defaultFactStreamParams(), func(t *testing.T, transport FactTransport) {
+		ctx := t.Context()
+		core := schema.GroupResource{Resource: "configmaps"}
+		followed := FactStreamKeyFor("prod-eu-1", core)
+		// Another cluster posting under its own route, and a CRD that happens to share the
+		// resource name. Neither may reach a follower of the first: a fact from cluster A naming
+		// the author of an object watched on cluster B is the failure the route dimension exists
+		// to prevent, and the group is what keeps two "configmaps" apart.
+		otherRoute := FactStreamKeyFor("prod-us-1", core)
+		otherGroup := FactStreamKeyFor("prod-eu-1", schema.GroupResource{Group: "example.com", Resource: "configmaps"})
+
+		sub := transport.FollowFacts([]FactStreamKey{followed}, factStreamTestHorizon)
+		require.NoError(t, transport.PublishFacts(ctx, otherRoute, authorFacts("other-cluster")))
+		require.NoError(t, transport.PublishFacts(ctx, otherGroup, authorFacts("other-group")))
+		require.NoError(t, transport.PublishFacts(ctx, followed, authorFacts("mine")))
+
+		require.Equal(t, []string{"mine"}, factAuthors(drainFactEntries(t, sub, 1)))
+		expectNoFactEntries(t, sub)
+	})
+}
+
+func TestFactStreamConformance_EmptyFollowedSetWaitsAndDeliversNothing(t *testing.T) {
+	runFactStreamConformance(t, defaultFactStreamParams(), func(t *testing.T, transport FactTransport) {
+		// A process with no watches follows nothing, and must idle rather than spin.
+		sub := transport.FollowFacts(nil, factStreamTestHorizon)
+		ctx, cancel := context.WithTimeout(t.Context(), factStreamTestTimeout)
+		defer cancel()
+		start := time.Now()
+		delivery, err := sub.Next(ctx)
+		require.NoError(t, err)
+		require.Empty(t, delivery.Entries)
+		require.GreaterOrEqual(t, time.Since(start), factStreamTestBlock)
+	})
+}
+
+func TestFactStreamConformance_EndedContextIsReported(t *testing.T) {
+	runFactStreamConformance(t, defaultFactStreamParams(), func(t *testing.T, transport FactTransport) {
+		key := factStreamTestKey("", "configmaps")
+		ctx, cancel := context.WithCancel(t.Context())
+		cancel()
+
+		following := transport.FollowFacts([]FactStreamKey{key}, factStreamTestHorizon)
+		_, err := following.Next(ctx)
+		require.Error(t, err)
+
+		idle := transport.FollowFacts(nil, factStreamTestHorizon)
+		_, err = idle.Next(ctx)
+		require.Error(t, err)
+	})
+}
+
+func TestFactStreamConformance_ZeroConfigIsUsable(t *testing.T) {
+	// The zero value of each config is the supported configuration: every knob falls back to its
+	// shared Default… constant rather than to zero.
+	store, _ := newTestRedisStoreWithRedis(t)
+	transports := map[string]FactTransport{
+		"redis":  store.FactStream(RedisFactStreamConfig{}),
+		"memory": NewMemoryFactStream(MemoryFactStreamConfig{}),
+	}
+	for name, transport := range transports {
+		t.Run(name, func(t *testing.T) {
+			key := factStreamTestKey("", "configmaps")
+			require.NoError(t, transport.PublishFacts(t.Context(), key, authorFacts("alice")))
+			sub := transport.FollowFacts([]FactStreamKey{key}, factStreamTestHorizon)
+			require.Equal(t, []string{"alice"}, factAuthors(drainFactEntries(t, sub, 1)))
+		})
+	}
+}
+
+func TestFactStreamKey_StringNamesRouteAndType(t *testing.T) {
+	require.Equal(t, "prod-eu-1/apps/deployments", factStreamTestKey("apps", "deployments").String())
+}
+
+// drainUntilFactStreamGap reads until a trim gap is reported, returning it with everything
+// delivered alongside and after it up to that round.
+func drainUntilFactStreamGap(t *testing.T, sub FactSubscription) (FactStreamGap, []FactEntry) {
+	t.Helper()
+	ctx, cancel := context.WithTimeout(t.Context(), factStreamTestTimeout)
+	defer cancel()
+	var entries []FactEntry
+	for {
+		delivery, err := sub.Next(ctx)
+		require.NoError(t, err, "no trim gap was reported")
+		entries = append(entries, delivery.Entries...)
+		if len(delivery.Gaps) > 0 {
+			return delivery.Gaps[0], entries
+		}
+	}
+}
diff --git a/internal/queue/fact_stream_memory.go b/internal/queue/fact_stream_memory.go
new file mode 100644
index 00000000..9212bd7c
--- /dev/null
+++ b/internal/queue/fact_stream_memory.go
@@ -0,0 +1,265 @@
+// SPDX-License-Identifier: Apache-2.0
+
+package queue
+
+import (
+	"context"
+	"strconv"
+	"sync"
+	"time"
+)
+
+// MemoryFactStreamConfig configures the in-memory transport. Every zero field falls back to its
+// Default… constant, so the zero value is the supported configuration.
+type MemoryFactStreamConfig struct {
+	// TTL is the retention horizon: entries older than it are dropped from the ring.
+	TTL time.Duration
+	// MaxLen caps one ring's entry count, oldest first. Zero means DefaultFactStreamMaxLen; a
+	// negative value means no cap, leaving the TTL as the only bound.
+	MaxLen int64
+	// Block is how long one Next waits for a new entry, which bounds how long a change to the
+	// followed set waits to take effect.
+	Block time.Duration
+	// ReadCount bounds how many entries one Next drains per stream.
+	ReadCount int64
+}
+
+// MemoryFactStream is the in-process implementation of the attribution fact transport: one ring
+// buffer per (audit route, group/resource), trimmed by TTL and entry count, with one cursor per
+// follower.
+//
+// It is the same data structure as the Redis one rather than an approximation of it. A Redis
+// stream is a capped, ordered log with per-reader cursors, which is a ring buffer: replay from the
+// horizon is reading the ring from its tail, and trim-gap detection is comparing a follower's
+// cursor against the ring's oldest surviving entry. Both fall out; neither is simulated.
+//
+// It only works when the audit receiver and the resolver are the same process. Selecting it
+// alongside more than one replica is a configuration error, and belongs at startup validation
+// rather than here — a transport cannot see how many replicas it has.
+type MemoryFactStream struct {
+	ttl       time.Duration
+	maxLen    int
+	block     time.Duration
+	readCount int
+
+	mu      sync.Mutex
+	streams map[FactStreamKey]*memFactRing
+	// signal is closed and replaced on every append, so a parked follower wakes on the next
+	// publish rather than at the end of its block period.
+	signal chan struct{}
+}
+
+// NewMemoryFactStream builds the in-process fact transport.
+func NewMemoryFactStream(cfg MemoryFactStreamConfig) *MemoryFactStream {
+	stream := &MemoryFactStream{
+		ttl:       cfg.TTL,
+		maxLen:    int(cfg.MaxLen),
+		block:     cfg.Block,
+		readCount: int(cfg.ReadCount),
+		streams:   map[FactStreamKey]*memFactRing{},
+		signal:    make(chan struct{}),
+	}
+	if stream.ttl <= 0 {
+		stream.ttl = DefaultAttributionFactTTL
+	}
+	if cfg.MaxLen == 0 {
+		stream.maxLen = int(DefaultFactStreamMaxLen)
+	}
+	if stream.block <= 0 {
+		stream.block = DefaultFactStreamBlock
+	}
+	if stream.readCount <= 0 {
+		stream.readCount = int(DefaultFactStreamReadCount)
+	}
+	return stream
+}
+
+// PublishFacts appends the batch as one entry on the key's ring and wakes every parked follower.
+func (m *MemoryFactStream) PublishFacts(ctx context.Context, key FactStreamKey, facts []AuthorFact) error {
+	if len(facts) == 0 {
+		return nil
+	}
+	if err := ctx.Err(); err != nil {
+		return err
+	}
+	raw, err := encodeFactBatch(facts)
+	if err != nil {
+		return err
+	}
+	now := time.Now()
+
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	ring, ok := m.streams[key]
+	if !ok {
+		ring = &memFactRing{}
+		m.streams[key] = ring
+	}
+	ring.append(raw, now)
+	ring.trim(now.Add(-m.ttl), m.maxLen)
+	close(m.signal)
+	m.signal = make(chan struct{})
+	return nil
+}
+
+// FollowFacts starts following keys from horizon before now.
+func (m *MemoryFactStream) FollowFacts(keys []FactStreamKey, horizon time.Duration) FactSubscription {
+	return &memFactSubscription{stream: m, follow: newFollowSet(keys, horizon)}
+}
+
+// wakeup returns the channel closed by the next publish.
+func (m *MemoryFactStream) wakeup() <-chan struct{} {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	return m.signal
+}
+
+// readRing reads one stream under the transport lock.
+func (m *MemoryFactStream) readRing(key FactStreamKey, cursor string, limit int) ([]memFactEntry, string) {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	ring, ok := m.streams[key]
+	if !ok {
+		return nil, ""
+	}
+	return ring.read(cursor, limit)
+}
+
+// memFactEntry is one appended batch, held encoded so a follower can never see it mutate.
+type memFactEntry struct {
+	id  string
+	at  time.Time
+	raw []byte
+}
+
+// memFactRing is one stream's bounded, ordered log. The slice is used as a FIFO ring: entries are
+// appended at the back and trimmed from the front, so its backing array is reused rather than
+// growing with the stream's lifetime.
+type memFactRing struct {
+	entries    []memFactEntry
+	lastMillis int64
+	lastSeq    int64
+}
+
+// append adds one entry, assigning the next strictly increasing "<millis>-<seq>" position. The
+// sequence half is what keeps several appends within one millisecond distinct and ordered, which
+// is the same rule the Redis transport gets from stream IDs.
+func (r *memFactRing) append(raw []byte, now time.Time) {
+	millis := now.UnixMilli()
+	if millis < 0 {
+		millis = 0
+	}
+	if millis > r.lastMillis {
+		r.lastMillis, r.lastSeq = millis, 0
+	} else {
+		// Same millisecond, or a clock that stepped backwards: keep the position monotonic and
+		// take the next sequence. A follower's cursor is only ever compared, never interpreted.
+		r.lastSeq++
+	}
+	id := strconv.FormatInt(r.lastMillis, 10) + streamIDSeparator + strconv.FormatInt(r.lastSeq, 10)
+	r.entries = append(r.entries, memFactEntry{id: id, at: now, raw: raw})
+}
+
+// trim drops entries appended before the horizon, then any beyond the entry cap, oldest first.
+func (r *memFactRing) trim(horizon time.Time, maxLen int) {
+	drop := 0
+	for drop < len(r.entries) && r.entries[drop].at.Before(horizon) {
+		drop++
+	}
+	if maxLen > 0 {
+		if excess := len(r.entries) - drop - maxLen; excess > 0 {
+			drop += excess
+		}
+	}
+	if drop > 0 {
+		r.entries = r.entries[drop:]
+	}
+}
+
+// read returns up to limit entries appended after cursor, plus the oldest surviving position so a
+// follower can tell it has been trimmed past.
+func (r *memFactRing) read(cursor string, limit int) ([]memFactEntry, string) {
+	if len(r.entries) == 0 {
+		return nil, ""
+	}
+	oldest := r.entries[0].id
+	var out []memFactEntry
+	for i := range r.entries {
+		if len(out) == limit {
+			break
+		}
+		if compareStreamIDs(r.entries[i].id, cursor) <= 0 {
+			continue
+		}
+		out = append(out, r.entries[i])
+	}
+	return out, oldest
+}
+
+// memFactSubscription is one follower's position across its followed rings.
+type memFactSubscription struct {
+	stream *MemoryFactStream
+	follow *followSet
+}
+
+// SetStreams replaces the followed set; it takes effect on the next Next.
+func (sub *memFactSubscription) SetStreams(keys []FactStreamKey) {
+	sub.follow.set(keys)
+}
+
+// Next drains every followed ring past this follower's cursors, waiting up to the block period for
+// something to arrive. It wakes early on a publish, so the block period bounds an idle round and a
+// subscription change rather than delivery latency.
+func (sub *memFactSubscription) Next(ctx context.Context) (FactDelivery, error) {
+	deadline := time.Now().Add(sub.stream.block)
+	for {
+		if err := ctx.Err(); err != nil {
+			return FactDelivery{}, err
+		}
+		// Take the wake-up channel BEFORE reading, so a publish that lands between the read and
+		// the wait is not missed: it closes the channel this iteration already holds.
+		signal := sub.stream.wakeup()
+		delivery := sub.collect()
+		if len(delivery.Entries) > 0 || len(delivery.Gaps) > 0 {
+			return delivery, nil
+		}
+		remaining := time.Until(deadline)
+		if remaining <= 0 {
+			return FactDelivery{}, nil
+		}
+		timer := time.NewTimer(remaining)
+		select {
+		case <-ctx.Done():
+			timer.Stop()
+			return FactDelivery{}, ctx.Err()
+		case <-signal:
+			timer.Stop()
+		case <-timer.C:
+			return FactDelivery{}, nil
+		}
+	}
+}
+
+// collect reads every followed ring once, advancing the cursors it delivered.
+func (sub *memFactSubscription) collect() FactDelivery {
+	targets := sub.follow.targets()
+	var delivery FactDelivery
+	for _, target := range targets {
+		entries, oldest := sub.stream.readRing(target.Key, target.Cursor, sub.stream.readCount)
+		if gap, ok := sub.follow.gapIfTrimmedPast(target, oldest); ok {
+			delivery.Gaps = append(delivery.Gaps, gap)
+		}
+		if len(entries) == 0 {
+			continue
+		}
+		for _, entry := range entries {
+			facts, err := decodeFactBatch(entry.raw)
+			if err != nil {
+				continue
+			}
+			delivery.Entries = append(delivery.Entries, FactEntry{Key: target.Key, ID: entry.id, Facts: facts})
+		}
+		sub.follow.advance(target.Key, entries[len(entries)-1].id, len(entries) >= sub.stream.readCount)
+	}
+	return delivery
+}
diff --git a/internal/queue/fact_stream_redis.go b/internal/queue/fact_stream_redis.go
new file mode 100644
index 00000000..678ac788
--- /dev/null
+++ b/internal/queue/fact_stream_redis.go
@@ -0,0 +1,266 @@
+// SPDX-License-Identifier: Apache-2.0
+
+package queue
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"strconv"
+	"sync"
+	"time"
+
+	"github.com/redis/go-redis/v9"
+)
+
+// RedisFactStreamConfig configures the Redis Streams transport. Every zero field falls back to
+// its Default… constant, so the zero value is the supported configuration.
+type RedisFactStreamConfig struct {
+	// TTL is the retention horizon: entries older than it are trimmed away. It is the same number
+	// as --author-attribution-ttl, which now bounds stream retention and the in-memory index
+	// together.
+	TTL time.Duration
+	// MaxLen caps one stream's entry count via XADD MAXLEN ~, so a hot type cannot grow without
+	// bound between retention trims. Zero means DefaultFactStreamMaxLen; a negative value means no
+	// cap, leaving the TTL as the only bound.
+	MaxLen int64
+	// TrimInterval is how often one stream is XTRIMmed to the retention horizon, zero meaning
+	// DefaultFactStreamTrimInterval. The trim is amortized onto the publish path rather than run
+	// from a goroutine: a stream nobody writes to needs no trimming, since nothing is arriving to
+	// age. It bounds a command round trip per publish, which is why the in-memory transport, whose
+	// trim is a slice re-slice, has no equivalent knob and trims on every append.
+	TrimInterval time.Duration
+	// Block is the XREAD BLOCK period, which bounds how long a change to the followed set waits to
+	// take effect.
+	Block time.Duration
+	// ReadCount is the XREAD COUNT per stream per read.
+	ReadCount int64
+}
+
+// RedisFactStream is the Redis Streams implementation of the attribution fact transport: one
+// stream per (audit route, group/resource), appended to with XADD and followed with a single
+// blocking XREAD across the followed set.
+//
+// It uses no consumer groups, deliberately. A consumer group distributes entries BETWEEN
+// consumers, and this is a fan-out: every process watching a type needs every fact for that type.
+// Each follower reads independently from its own in-memory cursor.
+type RedisFactStream struct {
+	client    *redis.Client
+	keyPrefix string
+	ttl       time.Duration
+	maxLen    int64
+	trimEvery time.Duration
+	block     time.Duration
+	readCount int64
+
+	trimMu   sync.Mutex
+	lastTrim map[string]time.Time
+}
+
+// FactStream builds the Redis Streams attribution fact transport on this store's connection, in
+// the same keyspace as its other keys. The blocking follower parks on one pooled connection for
+// the duration of each block period, so a process runs one follower rather than one per type.
+func (s *RedisStore) FactStream(cfg RedisFactStreamConfig) *RedisFactStream {
+	stream := &RedisFactStream{
+		client:    s.client,
+		keyPrefix: s.keyPrefix,
+		ttl:       cfg.TTL,
+		maxLen:    cfg.MaxLen,
+		trimEvery: cfg.TrimInterval,
+		block:     cfg.Block,
+		readCount: cfg.ReadCount,
+		lastTrim:  map[string]time.Time{},
+	}
+	if stream.ttl <= 0 {
+		stream.ttl = DefaultAttributionFactTTL
+	}
+	if stream.maxLen == 0 {
+		stream.maxLen = DefaultFactStreamMaxLen
+	}
+	if stream.trimEvery <= 0 {
+		stream.trimEvery = DefaultFactStreamTrimInterval
+	}
+	if stream.block <= 0 {
+		stream.block = DefaultFactStreamBlock
+	}
+	if stream.readCount <= 0 {
+		stream.readCount = DefaultFactStreamReadCount
+	}
+	return stream
+}
+
+// PublishFacts appends the batch as one XADD on the key's stream, capped with MAXLEN ~, and trims
+// the stream to the retention horizon when one is due.
+func (s *RedisFactStream) PublishFacts(ctx context.Context, key FactStreamKey, facts []AuthorFact) error {
+	if len(facts) == 0 {
+		return nil
+	}
+	raw, err := encodeFactBatch(facts)
+	if err != nil {
+		return err
+	}
+	streamKey := s.streamKey(key)
+	args := &redis.XAddArgs{
+		Stream: streamKey,
+		Values: map[string]any{factStreamEntryField: raw},
+	}
+	if s.maxLen > 0 {
+		args.MaxLen = s.maxLen
+		args.Approx = true
+	}
+	if err := s.client.XAdd(ctx, args).Err(); err != nil {
+		return fmt.Errorf("append facts to %q: %w", streamKey, err)
+	}
+	s.trimIfDue(ctx, streamKey)
+	return nil
+}
+
+// FollowFacts starts following keys from horizon before now.
+func (s *RedisFactStream) FollowFacts(keys []FactStreamKey, horizon time.Duration) FactSubscription {
+	return &redisFactSubscription{stream: s, follow: newFollowSet(keys, horizon)}
+}
+
+// streamKey renders one stream's Redis key, e.g.
+// "gitops-reverser:author:v2:audit:route:prod-eu-1:apps/deployments". The route sits directly
+// after the domain so one route's streams share a single glob prefix.
+func (s *RedisFactStream) streamKey(key FactStreamKey) string {
+	return resolveKeyPrefix(s.keyPrefix) + factStreamKeySuffix + routeKeyInfix +
+		escapeKeyField(key.AuditRoute) + ":" + key.groupResource()
+}
+
+// trimIfDue drops entries past the retention horizon with XTRIM MINID, at most once per stream per
+// trim interval. Stream IDs are millisecond timestamps, so the horizon is a plain MINID bound.
+//
+// A trim failure is deliberately not returned: the MAXLEN cap on every XADD already bounds the
+// stream, so a missed trim costs retention accuracy rather than safety, and failing an audit
+// request over it would trade a bounded cost for an unbounded one.
+func (s *RedisFactStream) trimIfDue(ctx context.Context, streamKey string) {
+	now := time.Now()
+	s.trimMu.Lock()
+	due := now.Sub(s.lastTrim[streamKey]) >= s.trimEvery
+	if due {
+		s.lastTrim[streamKey] = now
+	}
+	s.trimMu.Unlock()
+	if !due {
+		return
+	}
+	minID := strconv.FormatInt(now.Add(-s.ttl).UnixMilli(), 10)
+	_ = s.client.XTrimMinID(ctx, streamKey, minID).Err()
+}
+
+// redisFactSubscription is one follower's position across its followed streams: the cursors live
+// in memory only, because a process that lost them also lost its in-memory index and is starting
+// from the horizon anyway.
+type redisFactSubscription struct {
+	stream *RedisFactStream
+	follow *followSet
+}
+
+// SetStreams replaces the followed set; it takes effect on the next Next.
+func (sub *redisFactSubscription) SetStreams(keys []FactStreamKey) {
+	sub.follow.set(keys)
+}
+
+// Next issues one blocking XREAD across the whole followed set and returns what it read, plus any
+// stream this follower has been trimmed past. Re-issuing per call, rather than holding one long
+// read open, is what lets a subscription change take effect within one block period.
+func (sub *redisFactSubscription) Next(ctx context.Context) (FactDelivery, error) {
+	targets := sub.follow.targets()
+	if len(targets) == 0 {
+		return FactDelivery{}, waitBlock(ctx, sub.stream.block)
+	}
+	gaps := sub.detectGaps(ctx, targets)
+
+	byStreamKey := make(map[string]FactStreamKey, len(targets))
+	streams := make([]string, 0, len(targets))
+	cursors := make([]string, 0, len(targets))
+	for _, target := range targets {
+		streamKey := sub.stream.streamKey(target.Key)
+		byStreamKey[streamKey] = target.Key
+		streams = append(streams, streamKey)
+		cursors = append(cursors, target.Cursor)
+	}
+	// XREAD takes every key first and then every cursor, positionally paired.
+	streamArgs := make([]string, 0, len(streams)+len(cursors))
+	streamArgs = append(streamArgs, streams...)
+	streamArgs = append(streamArgs, cursors...)
+
+	res, err := sub.stream.client.XRead(ctx, &redis.XReadArgs{
+		Streams: streamArgs,
+		Count:   sub.stream.readCount,
+		Block:   sub.stream.block,
+	}).Result()
+	if errors.Is(err, redis.Nil) {
+		return FactDelivery{Gaps: gaps}, nil
+	}
+	if err != nil {
+		// This round's gap reports go nowhere, so forget them: a gap deduped against a report
+		// nobody received is exactly the silent loss the detection exists to prevent.
+		sub.follow.forgetGaps(gaps)
+		return FactDelivery{}, fmt.Errorf("read fact streams: %w", err)
+	}
+	return FactDelivery{Entries: sub.collect(res, byStreamKey), Gaps: gaps}, nil
+}
+
+// collect turns one XREAD result into entries and advances the cursors it delivered. An entry
+// whose payload does not decode is skipped rather than retried: it can never decode, and stalling
+// the follower on it would cost every later fact on that stream.
+func (sub *redisFactSubscription) collect(
+	res []redis.XStream,
+	byStreamKey map[string]FactStreamKey,
+) []FactEntry {
+	var entries []FactEntry
+	for i := range res {
+		key, ok := byStreamKey[res[i].Stream]
+		if !ok {
+			continue
+		}
+		messages := res[i].Messages
+		if len(messages) == 0 {
+			continue
+		}
+		for j := range messages {
+			facts, err := factsFromMessage(messages[j])
+			if err != nil {
+				continue
+			}
+			entries = append(entries, FactEntry{Key: key, ID: messages[j].ID, Facts: facts})
+		}
+		sub.follow.advance(key, messages[len(messages)-1].ID, int64(len(messages)) >= sub.stream.readCount)
+	}
+	return entries
+}
+
+// detectGaps asks each stream this follower is BEHIND on for its oldest surviving entry, and
+// reports the ones that have been trimmed past the follower's cursor. Only a behind follower is
+// asked, so a caught-up one costs nothing: it cannot have been trimmed past.
+func (sub *redisFactSubscription) detectGaps(ctx context.Context, targets []followTarget) []FactStreamGap {
+	var gaps []FactStreamGap
+	for _, target := range targets {
+		if !target.Behind {
+			continue
+		}
+		oldest, err := sub.stream.client.XRangeN(ctx, sub.stream.streamKey(target.Key), "-", "+", 1).Result()
+		if err != nil || len(oldest) == 0 {
+			continue
+		}
+		if gap, ok := sub.follow.gapIfTrimmedPast(target, oldest[0].ID); ok {
+			gaps = append(gaps, gap)
+		}
+	}
+	return gaps
+}
+
+// factsFromMessage decodes one stream entry's fact batch.
+func factsFromMessage(message redis.XMessage) ([]AuthorFact, error) {
+	value, ok := message.Values[factStreamEntryField]
+	if !ok {
+		return nil, fmt.Errorf("stream entry %q carries no %q field", message.ID, factStreamEntryField)
+	}
+	raw, ok := value.(string)
+	if !ok {
+		return nil, fmt.Errorf("stream entry %q carries a non-string %q field", message.ID, factStreamEntryField)
+	}
+	return decodeFactBatch([]byte(raw))
+}
diff --git a/internal/queue/fact_stream_redis_test.go b/internal/queue/fact_stream_redis_test.go
new file mode 100644
index 00000000..32b3f5d3
--- /dev/null
+++ b/internal/queue/fact_stream_redis_test.go
@@ -0,0 +1,51 @@
+// SPDX-License-Identifier: Apache-2.0
+
+package queue
+
+import (
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/require"
+	"k8s.io/apimachinery/pkg/runtime/schema"
+)
+
+// The behaviour of the Redis transport is covered by the conformance suite, which runs against
+// both implementations. What is left here is what only the Redis side has: the key layout, which
+// is an operator-visible contract — a stream is inspected, trimmed, and reasoned about by key.
+func TestRedisFactStream_StreamKeyLayout(t *testing.T) {
+	store, _ := newTestRedisStoreWithRedis(t)
+	stream := store.FactStream(RedisFactStreamConfig{})
+
+	// The API-path form, NOT schema.GroupResource.String()'s reversed "deployments.apps".
+	require.Equal(t,
+		"gitops-reverser:author:v2:audit:route:prod-eu-1:apps/deployments",
+		stream.streamKey(FactStreamKeyFor("prod-eu-1", schema.GroupResource{Group: "apps", Resource: "deployments"})))
+	// The core group has no prefix, matching the fact keys' group/resource form.
+	require.Equal(t,
+		"gitops-reverser:author:v2:audit:route:default:configmaps",
+		stream.streamKey(FactStreamKeyFor("default", schema.GroupResource{Resource: "configmaps"})))
+	// A route carrying the key delimiter is escaped, so it can never spill into the next segment.
+	require.Equal(t,
+		"gitops-reverser:author:v2:audit:route:odd%3Aroute:configmaps",
+		stream.streamKey(FactStreamKeyFor("odd:route", schema.GroupResource{Resource: "configmaps"})))
+}
+
+// TestRedisFactStream_TrimIsRateLimited pins the amortized trim: the first publish on a stream
+// trims it, and a publish inside the same interval does not.
+func TestRedisFactStream_TrimIsRateLimited(t *testing.T) {
+	store, mr := newTestRedisStoreWithRedis(t)
+	stream := store.FactStream(RedisFactStreamConfig{TTL: time.Millisecond, TrimInterval: time.Hour})
+	key := factStreamTestKey("", "configmaps")
+	ctx := t.Context()
+
+	require.NoError(t, stream.PublishFacts(ctx, key, authorFacts("first")))
+	time.Sleep(10 * time.Millisecond)
+	require.NoError(t, stream.PublishFacts(ctx, key, authorFacts("second")))
+
+	// The second publish is inside the trim interval, so the aged first entry is still there and
+	// the TTL horizon is enforced only when the next trim comes due.
+	entries, err := mr.Stream(stream.streamKey(key))
+	require.NoError(t, err)
+	require.Len(t, entries, 2)
+}